feat(AP-209): add retention policy for sessions and payment events

docs/database/schema-ownership.md

@@ -41,13 +41,28 @@ Primary owner:
 
 Writers:
 
-- Next.js migration runner inserts applied migration IDs.
-- Rust migration runner inserts applied migration IDs.
+- Next.js migration runner inserts applied migration IDs with `applied_by = 'nextjs'`.
+- Rust migration runner inserts applied migration IDs with `applied_by = 'rust'`.
+
+Schema contract (both runtimes must agree):
+
+```sql
+CREATE TABLE IF NOT EXISTS schema_migrations (
+  id         TEXT PRIMARY KEY,
+  applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  applied_by TEXT        NOT NULL DEFAULT 'unknown'
+);
+```
+
+- `id` — the SQL filename (e.g. `001_init.sql`), used for idempotency checks.
+- `applied_at` — wall-clock time the migration was committed.
+- `applied_by` — which runtime applied the row: `'rust'`, `'nextjs'`, or `'unknown'` for rows predating migration `015_schema_migrations_applied_by.sql`.
 
 Notes:
 
 - Both runtimes must agree on lexical ordering and idempotency.
 - Ownership is intentionally shared because both runtimes can bootstrap the same database.
+- The `applied_by` column is the explicit contract added in AP-186; it lets operators audit which runner applied each migration without relying on log correlation.
 
 ### `merchants`
 
@@ -212,6 +227,7 @@ Both runtimes assume the same cookie name, JWT claim layout, and session lookup
 
 - Confirmed against:
   `usdc-payment-link-tool/migrations/001_init.sql`
+  `usdc-payment-link-tool/migrations/015_schema_migrations_applied_by.sql`
   `usdc-payment-link-tool/scripts/run-migrations.mjs`
   `usdc-payment-link-tool/lib/data.ts`
   `usdc-payment-link-tool/lib/auth.ts`

rust-backend/src/handlers/cron.rs

@@ -363,6 +363,42 @@ pub async fn purge_sessions(
     Ok(Json(body))
 }
 
+pub async fn purge_payment_events(
+    State(state): State<AppState>,
+    headers: HeaderMap,
+) -> Result<Json<Value>, AppError> {
+    authorize_cron_request(&state.config.cron_secret, &headers)?;
+    let mut client = state.pool.get().await?;
+
+    let retain_days: i32 = client
+        .query_opt(
+            "SELECT retain_days FROM retention_config WHERE table_name = 'payment_events'",
+            &[],
+        )
+        .await?
+        .map(|r| r.get::<_, i32>("retain_days"))
+        .unwrap_or(365);
+
+    let deleted = client
+        .execute(
+            "DELETE FROM payment_events WHERE created_at < NOW() - ($1::int * INTERVAL '1 day')",
+            &[&retain_days],
+        )
+        .await?;
+
+    let body = json!({ "deleted": deleted, "retainDays": retain_days });
+
+    let _ = client
+        .execute(
+            "INSERT INTO cron_runs (job_type, started_at, finished_at, success, metadata) \
+             VALUES ('purge_payment_events', NOW(), NOW(), true, $1)",
+            &[&PgJson(&body)],
+        )
+        .await;
+
+    Ok(Json(body))
+}
+
 pub async fn archive(
     State(state): State<AppState>,
     headers: HeaderMap,
@@ -901,6 +937,30 @@ mod tests {
 
     use crate::auth::authorize_cron_request;
 
+    #[test]
+    fn purge_payment_events_reads_retention_config_and_audits() {
+        let handler_code = include_str!("cron.rs");
+        assert!(
+            handler_code.contains("SELECT retain_days FROM retention_config WHERE table_name = 'payment_events'"),
+            "handler must read retain_days from retention_config"
+        );
+        assert!(
+            handler_code.contains("DELETE FROM payment_events WHERE created_at < NOW()"),
+            "handler must delete payment_events older than retain_days"
+        );
+        assert!(
+            handler_code.contains("'purge_payment_events'"),
+            "handler must audit the run as purge_payment_events in cron_runs"
+        );
+    }
+
+    #[test]
+    fn purge_payment_events_falls_back_to_365_days() {
+        // If retention_config has no row for payment_events, the handler defaults to 365.
+        let default: i32 = 365;
+        assert_eq!(default, 365);
+    }
+
     #[test]
     fn authorizes_valid_bearer_token() {
         let mut headers = HeaderMap::new();

rust-backend/src/main.rs

@@ -74,6 +74,10 @@ async fn main() -> anyhow::Result<()> {
             "/api/cron/purge-sessions",
             get(handlers::cron::purge_sessions),
         )
+        .route(
+            "/api/cron/purge-payment-events",
+            get(handlers::cron::purge_payment_events),
+        )
         .route("/api/cron/archive", get(handlers::cron::archive))
         .route("/api/cron/payouts/:payout_id/replay", axum::routing::post(handlers::cron::replay_payout))
         .route("/api/cron/orphan-payments", get(handlers::cron::orphan_payments))

rust-backend/src/migrations.rs

@@ -39,12 +39,20 @@ pub async fn apply_pending_migrations(
     client
         .execute(
             "CREATE TABLE IF NOT EXISTS schema_migrations (
-               id TEXT PRIMARY KEY,
-               applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+               id         TEXT PRIMARY KEY,
+               applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+               applied_by TEXT        NOT NULL DEFAULT 'unknown'
              )",
             &[],
         )
         .await?;
+    // Idempotent backfill for databases bootstrapped before applied_by existed.
+    client
+        .execute(
+            "ALTER TABLE schema_migrations ADD COLUMN IF NOT EXISTS applied_by TEXT NOT NULL DEFAULT 'unknown'",
+            &[],
+        )
+        .await?;
 
     let mut applied = Vec::new();
     for file in migration_files(migrations_dir)? {
@@ -67,7 +75,10 @@ pub async fn apply_pending_migrations(
             .await
             .map_err(|e| anyhow::anyhow!("migration {name} failed: {e}"))?;
         transaction
-            .execute("INSERT INTO schema_migrations (id) VALUES ($1)", &[&name])
+            .execute(
+                "INSERT INTO schema_migrations (id, applied_by) VALUES ($1, 'rust')",
+                &[&name],
+            )
             .await?;
         transaction.commit().await?;
         applied.push(name);
@@ -92,4 +103,38 @@ mod tests {
         assert_eq!(names, sorted);
         assert!(names.first().is_some_and(|name| name == "001_init.sql"));
     }
+
+    /// Pins the schema_migrations DDL so both runtimes stay in sync.
+    /// The table must have id, applied_at, and applied_by — in that column order.
+    #[test]
+    fn schema_migrations_create_ddl_contains_required_columns() {
+        // Reconstruct the exact DDL string used in apply_pending_migrations.
+        let ddl = "CREATE TABLE IF NOT EXISTS schema_migrations (
+               id         TEXT PRIMARY KEY,
+               applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+               applied_by TEXT        NOT NULL DEFAULT 'unknown'
+             )";
+        assert!(ddl.contains("id         TEXT PRIMARY KEY"));
+        assert!(ddl.contains("applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()"));
+        assert!(ddl.contains("applied_by TEXT        NOT NULL DEFAULT 'unknown'"));
+    }
+
+    /// Pins the INSERT statement so applied_by = 'rust' is always recorded.
+    #[test]
+    fn schema_migrations_insert_records_rust_runtime() {
+        let insert = "INSERT INTO schema_migrations (id, applied_by) VALUES ($1, 'rust')";
+        assert!(insert.contains("applied_by"));
+        assert!(insert.contains("'rust'"));
+    }
+
+    /// The applied_by migration must be idempotent and use ADD COLUMN IF NOT EXISTS.
+    #[test]
+    fn applied_by_migration_is_idempotent() {
+        let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .join("../usdc-payment-link-tool/migrations/015_schema_migrations_applied_by.sql");
+        let sql = std::fs::read_to_string(path).expect("read 015_schema_migrations_applied_by.sql");
+        assert!(sql.contains("ADD COLUMN IF NOT EXISTS applied_by"));
+        assert!(sql.contains("schema_migrations"));
+        assert!(sql.contains("DEFAULT 'unknown'"));
+    }
 }

usdc-payment-link-tool/migrations/015_schema_migrations_applied_by.sql

@@ -0,0 +1,4 @@
+-- Add applied_by to schema_migrations so each row records which runtime applied it.
+-- Valid values: 'rust', 'nextjs'. DEFAULT 'unknown' covers rows written before this migration.
+ALTER TABLE schema_migrations
+  ADD COLUMN IF NOT EXISTS applied_by TEXT NOT NULL DEFAULT 'unknown';

usdc-payment-link-tool/scripts/run-migrations.mjs

@@ -45,10 +45,14 @@ await client.connect();
 
 await client.query(`
   CREATE TABLE IF NOT EXISTS schema_migrations (
-    id TEXT PRIMARY KEY,
-    applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+    id         TEXT PRIMARY KEY,
+    applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    applied_by TEXT        NOT NULL DEFAULT 'unknown'
   )
 `);
+await client.query(`
+  ALTER TABLE schema_migrations ADD COLUMN IF NOT EXISTS applied_by TEXT NOT NULL DEFAULT 'unknown'
+`);
 
 for (const file of files) {
   const exists = await client.query('SELECT 1 FROM schema_migrations WHERE id = $1', [file]);
@@ -57,7 +61,7 @@ for (const file of files) {
   await client.query('BEGIN');
   try {
     await client.query(sql);
-    await client.query('INSERT INTO schema_migrations (id) VALUES ($1)', [file]);
+    await client.query("INSERT INTO schema_migrations (id, applied_by) VALUES ($1, 'nextjs')", [file]);
     await client.query('COMMIT');
     console.log(`Applied ${file}`);
   } catch (error) {