Restore global request_id uniqueness and missing usage event columns

docs/schema.md

@@ -8,6 +8,17 @@ This document describes the core tables that power the Callora API marketplace,
 
 ## Tables
 
+### `usage_events` and `usage_event_request_ids`
+
+`usage_events` is hash partitioned by `developer_id`; its per-partition primary
+key includes that partition key, so it cannot enforce global request ID
+uniqueness. Every writer reserves `request_id` in `usage_event_request_ids`
+within the same transaction before inserting the event. The registry primary
+key provides the global idempotency guarantee and its unique `usage_event_id`
+is the target for `revenue_ledger.usage_event_id`. The partitioned parent also
+contains `api_key VARCHAR(255)` and `status_code INTEGER NOT NULL DEFAULT 200`
+for persistent usage-store reads and writes.
+
 ### `apis`
 
 Stores the top-level API products created by developers.

migrations/0028_usage_event_request_ids.down.sql

@@ -0,0 +1,16 @@
+-- dialect: postgres
+-- Rollback: 0028_usage_event_request_ids
+BEGIN;
+ALTER TABLE revenue_ledger DROP CONSTRAINT IF EXISTS revenue_ledger_usage_event_id_fkey;
+DO $$
+BEGIN
+  IF to_regclass('usage_events_old') IS NOT NULL THEN
+    ALTER TABLE revenue_ledger
+      ADD CONSTRAINT revenue_ledger_usage_event_id_fkey
+      FOREIGN KEY (usage_event_id) REFERENCES usage_events_old(id);
+  END IF;
+END$$;
+DROP TABLE IF EXISTS usage_event_request_ids;
+ALTER TABLE usage_events DROP COLUMN IF EXISTS status_code;
+ALTER TABLE usage_events DROP COLUMN IF EXISTS api_key;
+COMMIT;

migrations/0028_usage_event_request_ids.sql

@@ -0,0 +1,54 @@
+-- dialect: postgres
+-- Global idempotency registry for hash-partitioned usage events. PostgreSQL
+-- cannot enforce request_id uniqueness on a partitioned table without the
+-- partition key, so request IDs are reserved here in the same transaction.
+-- destructive-approved: #1486
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS usage_event_request_ids (
+  request_id VARCHAR(255) PRIMARY KEY,
+  usage_event_id BIGINT NOT NULL UNIQUE,
+  developer_id VARCHAR(255) NOT NULL,
+  created_at TIMESTAMP NOT NULL DEFAULT NOW()
+);
+
+-- Preserve the legacy persistent-store fields on the new partitioned parent.
+ALTER TABLE usage_events
+  ADD COLUMN IF NOT EXISTS api_key VARCHAR(255),
+  ADD COLUMN IF NOT EXISTS status_code INTEGER NOT NULL DEFAULT 200;
+
+-- Carry existing IDs forward before attaching the ledger FK. The standalone
+-- backfill script performs a duplicate audit and writes any collisions to a
+-- report; this SQL migration remains safe to rerun on a partially copied set.
+DO $$
+BEGIN
+  IF to_regclass('usage_events_old') IS NOT NULL THEN
+    INSERT INTO usage_event_request_ids (request_id, usage_event_id, developer_id)
+    SELECT o.request_id, o.id, COALESCE(NULLIF(o.developer_id, ''), a.developer_id::text, '')
+      FROM usage_events_old o LEFT JOIN apis a ON a.id::text = o.api_id
+    ON CONFLICT (request_id) DO NOTHING;
+  END IF;
+END$$;
+
+INSERT INTO usage_event_request_ids (request_id, usage_event_id, developer_id)
+SELECT ue.request_id, ue.id, ue.developer_id
+  FROM usage_events ue
+ON CONFLICT (request_id) DO NOTHING;
+
+-- Existing IDs were allocated by the pre-partitioned sequence. Ensure the new
+-- parent sequence starts above every ID reserved in the global registry.
+SELECT setval(
+  pg_get_serial_sequence('usage_events', 'id'),
+  GREATEST(COALESCE(MAX(usage_event_id), 1), 1),
+  COALESCE(MAX(usage_event_id), 0) > 0
+)
+FROM usage_event_request_ids;
+
+-- Replace the stale FK (which remained attached to usage_events_old after the
+-- table swap) with a link to the global registry's unique event id.
+ALTER TABLE revenue_ledger DROP CONSTRAINT IF EXISTS revenue_ledger_usage_event_id_fkey;
+ALTER TABLE revenue_ledger
+  ADD CONSTRAINT revenue_ledger_usage_event_id_fkey
+  FOREIGN KEY (usage_event_id) REFERENCES usage_event_request_ids(usage_event_id);
+
+COMMIT;

scripts/backfill-usage-partitions.ts

@@ -3,8 +3,7 @@
  * Backfill script: copy rows from usage_events_old into the new
  * hash-partitioned usage_events table.
  *
- * Idempotent: uses ON CONFLICT (request_id, developer_id) DO NOTHING so
- * re-running after a partial copy is safe.
+ * Idempotent: each batch reserves request IDs and copies events transactionally.
  *
  * Usage:
  *   DATABASE_URL=postgres://... tsx scripts/backfill-usage-partitions.ts
@@ -15,6 +14,7 @@
  */
 
 import pg from 'pg';
+import { writeFile } from 'node:fs/promises';
 import { logger } from '../src/logger.js';
 
 const { Pool } = pg;
@@ -47,6 +47,13 @@ async function main(): Promise<void> {
       );
       process.exit(1);
     }
+    const { rows: registryCheck } = await pool.query<{ exists: boolean }>(`
+      SELECT to_regclass('usage_event_request_ids') IS NOT NULL AS exists
+    `);
+    if (!registryCheck[0]?.exists) {
+      logger.error('usage_event_request_ids not found. Run migrations/0028_usage_event_request_ids.sql first.');
+      process.exit(1);
+    }
 
     const { rows: totalRows } = await pool.query<{ count: string }>(
       'SELECT COUNT(*)::text AS count FROM usage_events_old',
@@ -59,14 +66,50 @@ async function main(): Promise<void> {
       return;
     }
 
+    const { rows: duplicates } = await pool.query<{ request_id: string; developers: string }>(
+      `WITH source AS (
+         SELECT o.request_id, o.id,
+                COALESCE(NULLIF(o.developer_id, ''), a.developer_id::text, '') AS developer_id
+           FROM usage_events_old o LEFT JOIN apis a ON a.id::text = o.api_id
+       ), duplicate_ids AS (
+         SELECT request_id,
+                string_agg(DISTINCT developer_id, ',' ORDER BY developer_id) AS developers
+           FROM source GROUP BY request_id HAVING COUNT(*) > 1
+       ), registry_conflicts AS (
+         SELECT s.request_id,
+                string_agg(DISTINCT s.developer_id || ' <> ' || r.developer_id, ',' ORDER BY s.developer_id || ' <> ' || r.developer_id) AS developers
+           FROM source s JOIN usage_event_request_ids r USING (request_id)
+          WHERE r.usage_event_id <> s.id OR r.developer_id <> s.developer_id
+          GROUP BY s.request_id
+       )
+       SELECT request_id, developers FROM duplicate_ids
+       UNION
+       SELECT request_id, developers FROM registry_conflicts
+       ORDER BY request_id`,
+    );
+    if (duplicates.length) {
+      const reportPath = process.env['DUPLICATE_REPORT'] ?? 'usage-event-request-id-duplicates.csv';
+      await writeFile(reportPath, `request_id,developers\n${duplicates.map((row) => `"${row.request_id.replaceAll('"', '""')}","${row.developers.replaceAll('"', '""')}"`).join('\n')}\n`);
+      throw new Error(`Found ${duplicates.length} duplicate request_id values; report written to ${reportPath}`);
+    }
+
     let offset = 0;
     let copied = 0;
     let skipped = 0;
 
-    while (true) {
+    while (offset < total) {
       // Fetch a batch ordered by id for deterministic cursor progress
-      const { rows: batch } = await pool.query<{ id: string }>(
-        `INSERT INTO usage_events (
+      const client = await pool.connect();
+      let batch: { id: string }[];
+      try {
+        await client.query('BEGIN');
+        await client.query(
+        `INSERT INTO usage_event_request_ids (request_id, usage_event_id, developer_id)
+         SELECT o.request_id, o.id, COALESCE(NULLIF(o.developer_id, ''), a.developer_id::text, '')
+         FROM (SELECT * FROM usage_events_old ORDER BY id LIMIT $1 OFFSET $2) o
+         LEFT JOIN apis a ON a.id::text = o.api_id
+         ON CONFLICT (request_id) DO NOTHING`, [BATCH_SIZE, offset]);
+        const inserted = await client.query<{ id: string }>(`INSERT INTO usage_events (
            id,
            user_id,
            api_id,
@@ -84,7 +127,7 @@ async function main(): Promise<void> {
            o.api_id,
            o.endpoint_id,
            o.api_key_id,
-           COALESCE(o.developer_id, COALESCE(a.developer_id::text, '')),
+           COALESCE(NULLIF(o.developer_id, ''), COALESCE(a.developer_id::text, '')),
            o.amount_usdc,
            o.request_id,
            o.stellar_tx_hash,
@@ -95,12 +138,13 @@ async function main(): Promise<void> {
            LIMIT $1 OFFSET $2
          ) o
          LEFT JOIN apis a ON a.id::text = o.api_id
-         ON CONFLICT (request_id, developer_id) DO NOTHING
-         RETURNING id`,
-        [BATCH_SIZE, offset],
-      );
-
-      if (batch.length === 0) break;
+         ON CONFLICT (id, developer_id) DO NOTHING RETURNING id`, [BATCH_SIZE, offset]);
+        batch = inserted.rows;
+        await client.query('COMMIT');
+      } catch (error) {
+        await client.query('ROLLBACK');
+        throw error;
+      } finally { client.release(); }
 
       const batchInserted = batch.length;
       // batch from the SELECT could be up to BATCH_SIZE; inserted may be less due to conflicts
@@ -112,6 +156,14 @@ async function main(): Promise<void> {
 
     skipped = total - copied;
 
+    await pool.query(`
+      SELECT setval(
+        pg_get_serial_sequence('usage_events', 'id'),
+        GREATEST(COALESCE(MAX(usage_event_id), 1), 1),
+        COALESCE(MAX(usage_event_id), 0) > 0
+      ) FROM usage_event_request_ids
+    `);
+
     logger.info(`Backfill complete: ${copied} inserted, ${skipped} skipped (already present).`);
   } finally {
     await pool.end();

src/repositories/usageEventsRepository.pg.ts

@@ -339,7 +339,6 @@ export class PgUsageEventsRepository implements UsageEventsPgRepository {
     const apiId = assertNonEmpty(event.apiId, 'apiId');
     const endpointId = assertNonEmpty(event.endpointId, 'endpointId');
     const apiKeyId = assertNonEmpty(event.apiKeyId, 'apiKeyId');
-    const developerId = (event.developerId ?? '').trim();
     const requestId = assertNonEmpty(event.requestId, 'requestId');
     const amountVal = typeof event.amount === 'bigint'
       ? assertAmount(event.amount)
@@ -348,49 +347,55 @@ export class PgUsageEventsRepository implements UsageEventsPgRepository {
 
     const result = await this.writeDb.query<UsageEventRow>(
       `
-      INSERT INTO usage_events (
-        user_id,
-        api_id,
-        endpoint_id,
-        api_key_id,
-        developer_id,
-        amount_usdc,
-        request_id,
-        stellar_tx_hash,
-        created_at
+      WITH owner AS (
+        SELECT developer_id::text AS developer_id FROM apis WHERE id::text = $2 LIMIT 1
+      ), reservation AS (
+        INSERT INTO usage_event_request_ids (request_id, usage_event_id, developer_id)
+        SELECT $6, nextval(pg_get_serial_sequence('usage_events', 'id')), developer_id FROM owner
+        ON CONFLICT (request_id) DO NOTHING
+        RETURNING usage_event_id, developer_id
+      ), inserted AS (
+        INSERT INTO usage_events (
+          id, user_id, api_id, endpoint_id, api_key_id, developer_id,
+          amount_usdc, request_id, stellar_tx_hash, created_at
+        )
+        SELECT r.usage_event_id, $1, $2, $3, $4, r.developer_id,
+               $5, $6, $7, COALESCE($8, NOW())
+        FROM reservation r
+        RETURNING id, user_id, api_id, endpoint_id, api_key_id, developer_id,
+                  amount_usdc, request_id, stellar_tx_hash, created_at
       )
-      VALUES ($1, $2, $3, $4, $5, $6, $7, $8, COALESCE($9, NOW()))
-      ON CONFLICT (request_id, developer_id)
-      DO UPDATE SET request_id = EXCLUDED.request_id
-      RETURNING
-        id,
-        user_id,
-        api_id,
-        endpoint_id,
-        api_key_id,
-        developer_id,
-        amount_usdc,
-        request_id,
-        stellar_tx_hash,
-        created_at
+      SELECT * FROM inserted
+      UNION ALL
+      SELECT ue.id, ue.user_id, ue.api_id, ue.endpoint_id, ue.api_key_id,
+             ue.developer_id, ue.amount_usdc, ue.request_id, ue.stellar_tx_hash, ue.created_at
+      FROM usage_event_request_ids r
+      JOIN usage_events ue ON ue.id = r.usage_event_id
+      WHERE r.request_id = $6 AND NOT EXISTS (SELECT 1 FROM inserted)
     `,
       [
-        userId,
-        apiId,
-        endpointId,
-        apiKeyId,
-        developerId,
-        amount,
-        requestId,
-        event.stellarTxHash ?? null,
-        event.createdAt ?? null,
+        userId, apiId, endpointId, apiKeyId, amount, requestId,
+        event.stellarTxHash ?? null, event.createdAt ?? null,
       ],
     );
 
     const row = result.rows[0];
 
     if (!row) {
-      throw new Error(`Failed to create or retrieve usage event for requestId "${requestId}".`);
+      const api = await this.writeDb.query('SELECT 1 FROM apis WHERE id::text = $1 LIMIT 1', [apiId]);
+      if (!api.rows[0]) {
+        throw Object.assign(new Error(`Unknown API: ${apiId}`), { statusCode: 404, code: 'API_NOT_FOUND' });
+      }
+      // A concurrent writer may have committed the reservation after this
+      // statement's snapshot was taken. A fresh statement sees its complete row.
+      const retry = await this.writeDb.query<UsageEventRow>(
+        `SELECT ue.id, ue.user_id, ue.api_id, ue.endpoint_id, ue.api_key_id,
+                ue.developer_id, ue.amount_usdc, ue.request_id, ue.stellar_tx_hash, ue.created_at
+           FROM usage_event_request_ids r JOIN usage_events ue ON ue.id = r.usage_event_id
+          WHERE r.request_id = $1`, [requestId],
+      );
+      if (!retry.rows[0]) throw new Error(`Reserved request id has no usage event: ${requestId}`);
+      return mapUsageEventRow(retry.rows[0]);
     }
 
     return mapUsageEventRow(row);

src/routes/billing/deduct.ts

@@ -146,6 +146,10 @@ router.post(
       await res.locals.idempotency?.renew();
 
       if (!result.success) {
+        if (result.errorCode === "API_NOT_FOUND") {
+          next(new NotFoundError(result.error ?? "API not found"));
+          return;
+        }
         if (result.simulationDetails) {
           next(
             simulationFailureError(

src/routes/billing/deduct/bulk.ts

@@ -1,6 +1,6 @@
 import { Router, type Response, type NextFunction, type Request } from 'express';
 import { z } from 'zod';
-import { UnauthorizedError, PaymentRequiredError, InternalServerError } from '../../../errors/index.js';
+import { UnauthorizedError, PaymentRequiredError, InternalServerError, NotFoundError } from '../../../errors/index.js';
 import { requireAuth, type AuthenticatedLocals } from '../../../middleware/requireAuth.js';
 import { validate } from '../../../middleware/validate.js';
 import { idempotencyMiddleware } from '../../../middleware/idempotency.js';
@@ -78,6 +78,10 @@ router.post(
       );
 
       if (!result.success) {
+        if (result.errorCode === 'API_NOT_FOUND') {
+          next(new NotFoundError(result.error ?? 'API not found'));
+          return;
+        }
         if (result.error?.toLowerCase().includes('insufficient balance')) {
           next(new PaymentRequiredError(result.error, 'INSUFFICIENT_BALANCE'));
           return;