Add PricingPolicyService + admin REST + LISTEN/NOTIFY cache invalidation

PR-I1 (service half). Built off #6460 — sibling of #6464 (lineage primitives),
both stacked on payg-i2-document-classifier so they can review independently.

Lookup precedence:
  - PaygTeamExtensions.pricingPolicyId override → load that policy
  - else load the pricing_policy row with is_default = TRUE
  - if the override points at a deleted row, fall back to default (logs a warn —
    safety net for racing deletes)

Cache:
  - 30s Caffeine, keyed by teamId, max 10k entries — correctness floor.
  - Invalidated on PolicyChangedEvent from any source (admin mutation here, or
    cross-instance via the Postgres LISTEN runner).
  - Admin reads bypass the cache (getEffectivePolicyUncached) so admins always
    see their own writes.

Writes (service layer, transactional, fire-after-commit event):
  - create(draft) — rejects pre-set policy_id or is_default=true (promotion
    must go through setDefault so the partial unique idx is freed first).
  - setDefault(id) — clearDefaultFlag + flip; idempotent no-op if already
    default (no event fired in that case).
  - setTeamOverride(teamId, policyId|null) — validate policy exists before save.

Admin REST surface — /api/v1/admin/payg/...
  - GET  /policies, GET /policies/{id}, POST /policies
  - POST /policies/{id}/set-default
  - PUT  /teams/{teamId}/policy-override
  - GET  /teams/{teamId}/effective-policy  (bypasses cache)
  All gated by @PreAuthorize("hasRole('ADMIN')"). Validation errors return 400,
  unknown rows return 404.

LISTEN/NOTIFY runner (PolicyChangeListener):
  - Opens a dedicated raw JDBC connection via DriverManager (not HikariCP —
    Hikari would evict idle LISTEN connections) and polls getNotifications()
    on a daemon thread.
  - Reconnects with 5s backoff on SQLException; the 30s TTL is the correctness
    floor during outages.
  - Disabled by setting payg.policy.listen.enabled=false (default on).
  - PgJDBC moved to compileOnly on :saas; the runtime artifact is still
    bundled via :proprietary.

V11 migration: seed the V1 default policy (25 pages/unit, 5 MiB/unit,
min charge 1, file cap 1000) + step limits per JobSource (WEB/API/DESKTOP_APP
10, PIPELINE 20). Idempotent — only inserts when no default exists.

Tests: 17 PricingPolicyServiceTest + 14 PricingPolicyAdminControllerTest, all
passing. Coverage moves 10.87% → 12.40% LINE and 11.89% → 13.85% INSTRUCTION.

Counterpart Supabase migration (NOTIFY trigger function + per-table triggers
on pricing_policy*, payg_team_extensions overrides, wallet_policy; plus the
default-policy seed) ships on a separate SaaS PR on payg-i1-pricing-policy-service.

Author: ConnorYoh

app/saas/build.gradle

@@ -40,6 +40,10 @@ dependencies {
 
     api 'org.flywaydb:flyway-core'
     runtimeOnly 'org.flywaydb:flyway-database-postgresql'
+    // PgJDBC is bundled at runtime via :proprietary. We need `org.postgresql.PGConnection` at
+    // compile time for the LISTEN/NOTIFY runner (PricingPolicyService cache invalidation) —
+    // `compileOnly` keeps the artifact out of our build output.
+    compileOnly 'org.postgresql:postgresql:42.7.11'
 
     testImplementation 'com.tngtech.archunit:archunit-junit5:1.4.2'
 }

app/saas/src/main/java/stirling/software/saas/payg/policy/PolicyChangeListener.java

@@ -0,0 +1,192 @@
+package stirling.software.saas.payg.policy;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.postgresql.PGConnection;
+import org.postgresql.PGNotification;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Component;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Listens on the Postgres {@code policy_changed} channel and publishes a {@link PolicyChangedEvent}
+ * when a {@code pricing_policy*} row or a {@code payg_team_extensions} override changes. Triggers
+ * on the database side call {@code pg_notify('policy_changed', ...)}; this runner translates them
+ * into Spring events that {@link PricingPolicyService} (and any other cache holder) listens for.
+ *
+ * <h2>Why a dedicated raw JDBC connection</h2>
+ *
+ * <p>PgJDBC's {@code LISTEN} binds notifications to a specific {@link Connection} for that
+ * connection's lifetime. HikariCP would eventually evict an idle connection (Hikari's {@code
+ * idleTimeout} / {@code maxLifetime}), losing notifications without us noticing — so we open our
+ * own raw connection via {@link DriverManager} and hold it. Loss of one DB connection is an
+ * acceptable cost; if it ever becomes contentious we can carve out a tiny separate {@code
+ * DataSource}.
+ *
+ * <h2>Failure mode</h2>
+ *
+ * <p>If the connection drops (network blip, Postgres restart), the polling loop catches the {@link
+ * SQLException}, logs, sleeps {@value #RECONNECT_BACKOFF_MS} ms, and retries. The 30s {@link
+ * PricingPolicyService} cache TTL acts as a correctness floor during outages — admin mutations
+ * still propagate within at most 30 seconds even if LISTEN is wholly broken.
+ *
+ * <h2>Disabling</h2>
+ *
+ * <p>Set {@code payg.policy.listen.enabled=false} to skip the runner entirely (useful in tests and
+ * single-instance dev where the 30s TTL is more than enough).
+ */
+@Component
+@Profile("saas")
+@ConditionalOnProperty(
+        prefix = "payg.policy.listen",
+        name = "enabled",
+        havingValue = "true",
+        matchIfMissing = true)
+@Slf4j
+public class PolicyChangeListener {
+
+    static final String CHANNEL = "policy_changed";
+    private static final long POLL_TIMEOUT_MS = 5_000L;
+    private static final long RECONNECT_BACKOFF_MS = 5_000L;
+
+    private final String jdbcUrl;
+    private final String username;
+    private final String password;
+    private final ApplicationEventPublisher eventPublisher;
+
+    private final AtomicBoolean running = new AtomicBoolean(false);
+    private ExecutorService executor;
+    private Connection listenConnection;
+
+    public PolicyChangeListener(
+            @Value("${spring.datasource.url}") String jdbcUrl,
+            @Value("${spring.datasource.username}") String username,
+            @Value("${spring.datasource.password:}") String password,
+            ApplicationEventPublisher eventPublisher) {
+        this.jdbcUrl = jdbcUrl;
+        this.username = username;
+        this.password = password;
+        this.eventPublisher = eventPublisher;
+    }
+
+    @PostConstruct
+    void start() {
+        if (jdbcUrl == null || jdbcUrl.isBlank()) {
+            log.warn(
+                    "spring.datasource.url is empty; PolicyChangeListener will not start."
+                            + " PricingPolicyService falls back to its 30s TTL.");
+            return;
+        }
+        running.set(true);
+        executor =
+                Executors.newSingleThreadExecutor(
+                        r -> {
+                            Thread t = new Thread(r, "payg-policy-listen");
+                            t.setDaemon(true);
+                            return t;
+                        });
+        executor.submit(this::pollLoop);
+        log.info("PolicyChangeListener started on channel '{}'.", CHANNEL);
+    }
+
+    @PreDestroy
+    void stop() {
+        running.set(false);
+        closeConnectionQuietly();
+        if (executor != null) {
+            executor.shutdownNow();
+            try {
+                if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
+                    log.warn("PolicyChangeListener executor did not terminate within 5s.");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        log.info("PolicyChangeListener stopped.");
+    }
+
+    private void pollLoop() {
+        while (running.get()) {
+            try {
+                if (listenConnection == null || listenConnection.isClosed()) {
+                    listenConnection = openListenConnection();
+                }
+                drainNotifications(listenConnection);
+            } catch (SQLException e) {
+                log.warn(
+                        "PolicyChangeListener IO error ({}). Reconnecting in {}ms.",
+                        e.getMessage(),
+                        RECONNECT_BACKOFF_MS);
+                closeConnectionQuietly();
+                sleepQuietly(RECONNECT_BACKOFF_MS);
+            } catch (RuntimeException e) {
+                // Don't let a bug in the polling loop kill the thread silently.
+                log.error("PolicyChangeListener unexpected error; restarting after backoff.", e);
+                closeConnectionQuietly();
+                sleepQuietly(RECONNECT_BACKOFF_MS);
+            }
+        }
+    }
+
+    private Connection openListenConnection() throws SQLException {
+        Connection conn = DriverManager.getConnection(jdbcUrl, username, password);
+        try (Statement st = conn.createStatement()) {
+            st.execute("LISTEN " + CHANNEL);
+        }
+        return conn;
+    }
+
+    private void drainNotifications(Connection conn) throws SQLException {
+        PGConnection pg = conn.unwrap(PGConnection.class);
+        PGNotification[] notifications = pg.getNotifications((int) POLL_TIMEOUT_MS);
+        if (notifications == null) {
+            return;
+        }
+        for (PGNotification n : notifications) {
+            String payload = n.getParameter();
+            log.debug("PolicyChangeListener received notification: '{}'.", payload);
+            try {
+                eventPublisher.publishEvent(new PolicyChangedEvent(this, payload));
+            } catch (RuntimeException publishError) {
+                // A bad listener mustn't break this loop or stop future notifications.
+                log.warn(
+                        "PolicyChangedEvent listener threw ({}); continuing.",
+                        publishError.getMessage());
+            }
+        }
+    }
+
+    private void closeConnectionQuietly() {
+        if (listenConnection != null) {
+            try {
+                listenConnection.close();
+            } catch (SQLException e) {
+                log.debug("Ignoring close error on listen connection: {}", e.getMessage());
+            }
+            listenConnection = null;
+        }
+    }
+
+    private static void sleepQuietly(long ms) {
+        try {
+            Thread.sleep(ms);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}

app/saas/src/main/java/stirling/software/saas/payg/policy/PolicyChangedEvent.java

@@ -0,0 +1,28 @@
+package stirling.software.saas.payg.policy;
+
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * Fires when a {@code pricing_policy*} or {@code payg_team_extensions.pricing_policy_id} row
+ * changes — published by the Postgres LISTEN runner (see {@link PolicyChangeListener}) or by admin
+ * REST mutations directly. {@link PricingPolicyService} listens and invalidates its cache.
+ *
+ * <p>The {@code payload} echoes whatever the {@code pg_notify} channel carried (kept loose because
+ * the invalidation strategy is "blow the whole cache" — payload detail doesn't change behaviour).
+ * Field is exposed for logging/observability only.
+ */
+public class PolicyChangedEvent extends ApplicationEvent {
+
+    private static final long serialVersionUID = 1L;
+
+    private final String payload;
+
+    public PolicyChangedEvent(Object source, String payload) {
+        super(source);
+        this.payload = payload;
+    }
+
+    public String getPayload() {
+        return payload;
+    }
+}