Implement SEP-24 callback auth, worker query indexes, unified error responses, and graceful shutdown

README.md

@@ -281,6 +281,19 @@ Retry budgets are exponential with jitter and fully configurable via env vars
   (default 5000), `WORKER_ANCHOR_RETRY_MAX_DELAY_MS` (default 120000),
   `WORKER_ANCHOR_RETRY_JITTER_RATIO` (default 0.25)
 
+### Graceful shutdown
+
+Both processes shut down gracefully on `SIGTERM`/`SIGINT` (see
+`src/lib/shutdown.ts`). The API stops accepting new connections, lets in-flight
+requests finish, then disconnects Prisma; the worker stops claiming new jobs,
+lets the active job cycle finish through its normal path, then releases its
+leases so the next worker picks the work up immediately. Cleanup is bounded:
+`SHUTDOWN_TIMEOUT_MS` for the API, `WORKER_SHUTDOWN_TIMEOUT_MS` for the worker.
+A dependency that hangs past the deadline forces the process to exit with a
+non-zero status instead of lingering. Repeated signals are ignored once
+shutdown has begun, so a deployment sending SIGTERM then SIGINT cannot run
+cleanup twice.
+
 ## How it works
 
 ### SEP-10 login
@@ -342,8 +355,13 @@ treasury account and, when `treasuryRequiredSigners > 1`, returned in
 ### Anchors (SEP-24)
 `POST /anchors/deposit|withdraw` creates a session and fetches a SEP-10 challenge
 **from the anchor**. The wallet signs it; `POST /anchors/sessions/:id/complete`
-exchanges it for an anchor JWT and the interactive deposit/withdraw URL. A signed
-`POST /anchors/webhook` updates session status; the worker also polls.
+exchanges it for an anchor JWT and the interactive deposit/withdraw URL. Status
+updates arrive either as callbacks authenticated by the configured shared
+secret (`POST /api/sep24/callback` and the legacy `POST /anchors/webhook`) or
+via the worker's polling; both paths run through the same idempotent,
+terminal-protected transition maps in `src/services/anchor-status.ts` and
+`src/services/withdrawal-status.ts`. See
+[docs/api-contract.md](docs/api-contract.md#sep-24-anchor-callback).
 
 ## Endpoints
 
@@ -359,6 +377,7 @@ exchanges it for an anchor JWT and the interactive deposit/withdraw URL. A signe
 | GET | `/groups/:id/balances` · `/groups/:id/ledger` | Balances & ledger |
 | POST/GET | `/groups/:id/treasury/*` · `/treasury-transactions/:id/confirm` | Treasury |
 | GET/POST | `/anchors` · `/anchors/deposit` · `/anchors/withdraw` · `/anchors/sessions/:id/complete` · `/anchors/sessions` · `/anchors/webhook` | Anchors |
+| POST | `/api/sep24/callback` | SEP-24 anchor status callback (shared-secret auth) |
 | GET | `/history` | Cross-group history |
 | POST/GET | `/uploads/receipt` · `/uploads/:file` | Receipts |
 | GET | `/health` · `/health/live` · `/health/ready` | Liveness & readiness probes (see [HEALTH.md](HEALTH.md)) |

docs/api-contract.md

@@ -11,20 +11,26 @@ or query convention changes.
 
 ## Error envelope
 
-Every error — validation, authorization, rate limiting, upstream — uses one shape:
+Every error — validation, authorization, rate limiting, upstream — uses one shape.
+The HTTP status code carries the error class; the body never repeats it:
 
 ```json
 {
+  "code": "NOT_FOUND",
   "error": "NOT_FOUND",
   "message": "Settlement not found",
-  "statusCode": 404,
-  "details": { "…": "optional, structured" },
-  "requestId": "01J…"
+  "requestId": "01J…",
+  "details": { "…": "optional, structured" }
 }
 ```
 
-`error` is a stable machine-readable code from `ErrorCode` in
-[../src/lib/errors.ts](../src/lib/errors.ts). Codes worth calling out:
+`code` is the canonical stable machine-readable code from `ErrorCode` in
+[../src/lib/errors.ts](../src/lib/errors.ts). `error` is a deprecated alias of
+`code`, retained so existing clients keep working; new code should read `code`.
+`details` is present only when there is structured information to convey (e.g.
+the offending fields of a `VALIDATION_ERROR`). Error bodies never include stack
+traces, SQL, credentials, signed XDRs, or upstream response bodies — those go to
+the server log only, correlated by `requestId`. Codes worth calling out:
 
 | Code | Status | Meaning |
 | --- | --- | --- |
@@ -33,6 +39,7 @@ Every error — validation, authorization, rate limiting, upstream — uses one
 | `INTENT_EXPIRED` | 400 | The unsigned transaction's signing window has closed — request a new one |
 | `XDR_MISMATCH` | 400 | The signed envelope does not match the intent it was built for |
 | `XDR_MALFORMED` | 400 | The envelope could not be parsed at all |
+| `PAYLOAD_TOO_LARGE` | 413 | The request body exceeded the route's size limit |
 | `INVALID_IDEMPOTENCY_KEY` | 400 | `Idempotency-Key` is outside 1–255 characters of `A–Z a–z 0–9 - _ . :` |
 | `MISSING_IDEMPOTENCY_KEY` | 400 | The route requires an `Idempotency-Key` header and none was sent |
 | `IDEMPOTENCY_CONFLICT` | 409 | The key was already used with a different payload |
@@ -385,13 +392,66 @@ submission, and anchor routes each get their own bucket. See the table in
 [../README.md](../README.md#rate-limiting) and the policy definitions in
 [../src/lib/rate-limit.ts](../src/lib/rate-limit.ts).
 
-A 429 uses the standard error envelope with `error: "RATE_LIMITED"` and, where
+A 429 uses the standard error envelope with `code: "RATE_LIMITED"` and, where
 available, `details.retryAfterSeconds`, alongside the usual `Retry-After` and
 `X-RateLimit-*` headers. It reveals nothing about the caller's identity or
 whether a wallet account is known to the API.
 
 ---
 
+## SEP-24 anchor callback
+
+`POST /api/sep24/callback` receives asynchronous SEP-24 deposit and withdrawal
+status updates from the configured anchor (see `src/services/sep24.ts`). It is
+registered outside the authenticated route scopes — the anchor has no Mergepay
+session — so its own credential is what authenticates it.
+
+### Authentication
+
+| | |
+| --- | --- |
+| Header | `x-anchor-signature` (alias: `x-webhook-secret`) |
+| Value | The configured shared secret `ANCHOR_WEBHOOK_SECRET` |
+| Verification | Constant-time comparison, before the body is parsed or any database read |
+
+A missing or incorrect secret is a `401` with the standard error envelope. The
+rejection never discloses *why* it failed (missing vs. wrong secret), and the
+secret is never logged or persisted.
+
+### Payload
+
+Either the SEP-24 transaction envelope or the flattened shape, with unknown
+fields ignored:
+
+```json
+{ "transaction": { "id": "anchor_tx_1", "status": "completed", "asset_code": "USDC", "kind": "deposit" } }
+{ "id": "anchor_tx_1", "status": "completed" }
+```
+
+`id` and `status` are required (400 `VALIDATION_ERROR` otherwise).
+`asset_code`, `asset_issuer`, and `kind`, when present, scope the match to
+local records carrying the same values.
+
+### Behavior
+
+- The callback's transaction id is used to look up local records **only after**
+  the secret is verified, and is scoped to the configured anchor, asset, and
+  kind — never trusted on its own.
+- Matching `AnchorSession` rows are advanced through the finite transition map
+  in `src/services/anchor-status.ts`; `Withdrawal` rows keyed by the same
+  anchor transaction id (`anchorTxId`) are advanced through their own map in
+  `src/services/withdrawal-status.ts`. Both are idempotent and both protect
+  terminal states: a duplicate delivery is a no-op and a stale or contradictory
+  callback can never regress a completed/refunded record.
+- Every applied transition writes its audit record in the same database
+  transaction as the status change.
+- The response is `200` even when nothing matched or the transition was
+  disallowed (anchors retry non-2xx responses, so a correctly-processed no-op
+  must not amplify load). A callback for an unknown transaction is audited as
+  `sep24.callback.unmatched`.
+
+---
+
 ## Health endpoints
 
 Operational probes for deployments and load balancers. They require no

prisma/migrations/20260828000001_add_worker_query_indexes/migration.sql

@@ -0,0 +1,26 @@
+-- Worker-query indexes (issue #200).
+--
+-- Each index below mirrors the actual filter/order of a high-frequency
+-- worker or sweep query in src/worker/index.ts and src/worker/*.ts, so the
+-- database can serve those queries without scanning the table:
+
+-- recoverStaleSettlements(): status IN (submitted, verifying) AND
+-- lease_expires_at < now. The composite covers both the status filter and
+-- the expired-lease predicate in one index.
+CREATE INDEX IF NOT EXISTS "settlements_status_lease_expires_at_idx"
+  ON "settlements" ("status", "lease_expires_at");
+
+-- recoverStaleAnchorSessions(): lease_expires_at < now. Anchor sessions had
+-- no index on this column at all, so every recovery sweep scanned the table.
+CREATE INDEX IF NOT EXISTS "anchor_sessions_lease_expires_at_idx"
+  ON "anchor_sessions" ("lease_expires_at");
+
+-- expireInvites(): expires_at IS NOT NULL AND expires_at < now.
+CREATE INDEX IF NOT EXISTS "invites_expires_at_idx"
+  ON "invites" ("expires_at");
+
+-- expireStaleProposals(): status = 'pending' AND created_at < cutoff
+-- ORDER BY created_at ASC (oldest first). The composite matches the sweep's
+-- filter and its sort order in one index.
+CREATE INDEX IF NOT EXISTS "treasury_proposals_status_created_at_idx"
+  ON "treasury_proposals" ("status", "created_at");

prisma/schema.prisma

@@ -181,6 +181,7 @@ model Settlement {
   @@index([status, updatedAt]) // For worker reconciliation queries
   @@index([status, nextAttemptAt]) // For picking up eligible worker jobs
   @@index([leaseExpiresAt]) // For recovering leases from crashed workers
+  @@index([status, leaseExpiresAt]) // For the worker's stale-lease recovery sweep
   @@index([expiresAt])
   @@index([expenseShareId])
   @@unique([expenseShareId, idempotencyKey], name: "settlement_expense_share_idempotency")
@@ -232,6 +233,7 @@ model Invite {
   createdBy User  @relation(fields: [createdByUserId], references: [id])
 
   @@index([groupId])
+  @@index([expiresAt]) // For the worker's expired-invite sweep
   @@map("invites")
 }
 
@@ -284,6 +286,7 @@ model AnchorSession {
   @@index([status])
   @@index([anchorToken, externalTransactionId]) // For worker polling queries
   @@index([status, lastPolledAt]) // For worker polling with time-based filtering
+  @@index([leaseExpiresAt]) // For recovering leases from crashed workers
   @@map("anchor_sessions")
 }
 
@@ -351,6 +354,7 @@ model TreasuryProposal {
   @@index([groupId])
   @@index([status])
   @@index([creatorId])
+  @@index([status, createdAt]) // For the worker's stale-proposal sweep (oldest first)
   @@map("treasury_proposals")
 }
 

src/app.ts

@@ -236,6 +236,7 @@ export async function buildApp(): Promise<FastifyInstance> {
     addHeaders: { "x-ratelimit-limit": true, "x-ratelimit-remaining": true, "x-ratelimit-reset": true, "retry-after": true } as any,
     errorResponseBuilder: (request: FastifyRequest) => ({
       code: "RATE_LIMITED",
+      error: "RATE_LIMITED",
       message: "Too many requests. Please retry later.",
       requestId: request.id,
     }),
@@ -315,6 +316,7 @@ export async function buildApp(): Promise<FastifyInstance> {
     reply.header("x-correlation-id", correlationId);
     reply.code(404).send({
       code: "NOT_FOUND",
+      error: "NOT_FOUND",
       message: "Route not found",
       requestId: correlationId,
     });

src/config.ts

@@ -26,7 +26,6 @@ const schema = z.object({
   // database concurrency across Fastify workers on a single instance.
   DATABASE_CONNECTION_LIMIT: z.coerce.number().int().positive().default(5),
   PORT: z.coerce.number().int().positive().default(4000),
-  SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(10_000),
   API_PUBLIC_URL: urlSchema,
   LOG_LEVEL: z
     .enum(["fatal", "error", "warn", "info", "debug", "trace", "silent"])
@@ -129,6 +128,13 @@ const schema = z.object({
     .max(1024 * 1024)
     .default(64 * 1024),
 
+  // Graceful shutdown bounds. The API closes the HTTP server then disconnects
+  // Prisma; the worker waits for its in-flight job cycle. If either exceeds its
+  // bound, a stuck dependency cannot keep the process alive forever — the
+  // process force-exits with a non-zero status. See src/lib/shutdown.ts.
+  SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(10_000),
+  WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(60_000),
+
   // Worker configuration
   WORKER_INTERVAL_MS: z.coerce.number().positive().default(30000),
   // How long a worker's claim on a job survives without renewal. A process that

src/lib/constant-time.ts

@@ -0,0 +1,18 @@
+import { timingSafeEqual } from "node:crypto";
+
+/**
+ * Constant-time comparison of two strings, for verifying shared secrets
+ * (webhook and callback signatures) without leaking how much of the expected
+ * value a guess matched.
+ *
+ * `timingSafeEqual` requires equal buffer lengths, so length is checked
+ * first. A length mismatch reveals only the secret's length — never its
+ * contents — and that information is already implied by the constant-time
+ * comparison itself, so this is the standard safe pattern.
+ */
+export function safeEqual(a: string, b: string): boolean {
+  const ab = Buffer.from(a);
+  const bb = Buffer.from(b);
+  if (ab.length !== bb.length) return false;
+  return timingSafeEqual(ab, bb);
+}

src/lib/shutdown.ts

@@ -0,0 +1,118 @@
+/**
+ * Shared graceful-shutdown coordination for the API and worker processes.
+ *
+ * Both entry points (`src/index.ts`, `src/worker/index.ts`) need the same
+ * three guarantees, so the coordination lives here rather than being
+ * reimplemented per process:
+ *
+ *   - **Idempotent.** Only the first signal runs the cleanup. A deployment
+ *     sending SIGTERM and then SIGINT (or a user pressing Ctrl-C twice) must
+ *     not trigger duplicate cleanup.
+ *   - **Bounded.** Cleanup gets a deadline (`timeoutMs`). If a dependency
+ *     stays stuck — a hanging Horizon call, a Prisma query that never
+ *     returns — the process is force-exited instead of lingering forever.
+ *   - **Observable.** Every phase (start, complete, error, timeout) is logged
+ *     with the process name and signal, so operators can tell *what* shut
+ *     down and *how*. No secrets are ever logged here.
+ *
+ * The caller supplies `onComplete`/`onTimeout` (typically `process.exit(0)`
+ * and `process.exit(1)`), so tests can exercise the coordinator without
+ * exiting the test runner.
+ */
+import pino from "pino";
+
+/**
+ * The small slice of a logger the coordinator needs. Typed structurally so
+ * both a pino logger (worker) and Fastify's logger (API) can be passed.
+ */
+export interface ShutdownLogger {
+  info(obj: Record<string, unknown>, msg?: string): void;
+  warn(obj: Record<string, unknown>, msg?: string): void;
+  error(obj: Record<string, unknown>, msg?: string): void;
+}
+
+export interface ShutdownCoordinatorOptions {
+  /** Identifies this process in shutdown logs (e.g. "api", "worker"). */
+  name: string;
+  /** Logger for shutdown-phase lines; defaults to a dedicated pino logger. */
+  logger?: ShutdownLogger;
+  /** Upper bound (ms) for cleanup; the process is force-exited after this. */
+  timeoutMs: number;
+  /** Called after cleanup completes within the deadline (e.g. exit(0)). */
+  onComplete?: () => void;
+  /** Called when the deadline passes or cleanup throws (e.g. exit(1)). */
+  onTimeout?: () => void;
+}
+
+export interface ShutdownCoordinator {
+  /**
+   * Begin graceful shutdown. Safe to call more than once — only the first
+   * call runs the cleanup; repeated signals are logged and ignored.
+   */
+  begin(signal: string, cleanup: () => Promise<void>): void;
+  /** Resolves when the first cleanup attempt finishes, by completion or deadline. */
+  readonly done: Promise<void>;
+}
+
+export function createShutdownCoordinator(
+  options: ShutdownCoordinatorOptions
+): ShutdownCoordinator {
+  const log: ShutdownLogger =
+    options.logger ?? (pino({ name: `shutdown:${options.name}` }) as unknown as ShutdownLogger);
+  let started = false;
+  let outcomeCalled = false;
+  let resolveDone: () => void = () => undefined;
+  const done = new Promise<void>((resolve) => {
+    resolveDone = resolve;
+  });
+
+  const callOutcome = (fn: (() => void) | undefined): void => {
+    if (fn && !outcomeCalled) {
+      outcomeCalled = true;
+      fn();
+    }
+  };
+
+  return {
+    done,
+    begin(signal, cleanup) {
+      if (started) {
+        log.warn(
+          { signal, phase: "duplicate" },
+          `${options.name} shutdown already in progress; ignoring repeated signal`
+        );
+        return;
+      }
+      started = true;
+
+      const deadline = setTimeout(() => {
+        log.error(
+          { signal, phase: "timeout", timeoutMs: options.timeoutMs },
+          `${options.name} shutdown exceeded its ${options.timeoutMs}ms deadline; force-exiting`
+        );
+        callOutcome(options.onTimeout);
+        resolveDone();
+      }, options.timeoutMs);
+
+      log.info({ signal, phase: "start" }, `${options.name} shutting down`);
+
+      void (async () => {
+        try {
+          await cleanup();
+          clearTimeout(deadline);
+          log.info({ signal, phase: "complete" }, `${options.name} shutdown complete`);
+          callOutcome(options.onComplete);
+        } catch (error) {
+          clearTimeout(deadline);
+          log.error(
+            { signal, phase: "error", err: error },
+            `${options.name} shutdown cleanup failed`
+          );
+          callOutcome(options.onTimeout);
+        } finally {
+          resolveDone();
+        }
+      })();
+    },
+  };
+}