fix(security): harden sessions, financial bounds, and administrator invariants

docs/DEPLOYMENT.md

@@ -60,6 +60,45 @@ headroom for the worker process's own pool. If you see `P2024` errors, increase
 `DATABASE_POOL_SIZE` rather than `DATABASE_POOL_TIMEOUT` — a longer timeout
 merely delays failures under sustained pressure.
 
+## Restricted worker execution
+
+The current worker binary deliberately does **not** run `blockchain-indexer`,
+`notifications`, or `fraud-activity-cleanup`. It registers no consumers or recurring
+jobs for them and does not replay historical events. Pending jobs and existing
+repeat definitions remain stored; they are not drained, cancelled or executed by
+this binary. Polling/checkpoint progress, queued notification delivery and fraud
+retention cleanup are unavailable. API endpoints remain available under their
+existing authorization rules; this does not make the application fully operational.
+
+Historically the worker intended to schedule polling and daily cleanup and start
+all three consumers. Missing imports prevented startup. The explicit restriction
+replaces that accidental failure while restoring compilation. The process now
+runs only its liveness heartbeat and shutdown handling. Startup logs report
+`workerExecution.status: blocked`, each queue's reason, and `workerCount: 0`.
+`GET /health/queue` includes the same fixed `workerExecution` state and reports
+`degraded` even for empty queues (existing backend/stall failures still return503).
+`GET /health/indexer` reports503/degraded with the same restriction and
+`healthy: false` for each contract, even when its lag snapshot is small.
+The heartbeat and `GET /health` prove process/dependency liveness only, **not job
+availability or deployment readiness**.
+
+There is no environment override for activation. A separately reviewed code change
+and rollout decision must first establish:
+
+- Indexer consumer completion, ordering/idempotency and recovery, complete event
+  coverage and safe checkpoint progression; historical replay needs its own policy.
+- Notification intent/outbox deduplication and recovery for ambiguous SMTP outcomes,
+  including pending backlog treatment without duplicate or unapproved sends.
+- Retention/window consistency, a reliable processing frontier aligned with evaluator
+  time, bounded deletion and restart/concurrency recovery without loss of required data.
+
+Do not substitute the unfinished M10 system or globally enable execution to make
+status green. Before rollout, inventory existing workers and recurring schedules:
+older binaries must not keep consuming these queues. This binary does not remotely
+pause queues or disable other processes. Expect backlog/storage growth and stalled
+indexer lag; monitor capacity and communicate the unavailable services. Existing
+schema/session migration and recovery requirements still apply.
+
 ## Health Checks
 
 - `GET /health` — liveness/readiness: database + Redis reachability (see `src/shared/http/routes/health.ts`). **Point your orchestrator's readiness probe at this endpoint.**

docs/SESSION_SECURITY.md

@@ -0,0 +1,13 @@
+# Session revocation policy
+
+Each new login creates an independent persisted refresh-token family. Rotation locks the user row and commits token consumption and successor insertion together. A cryptographically valid, unexpired token whose persisted row is already revoked is a reuse signal: the whole family is revoked, including any successor committed by a concurrent rotation. There is no retry grace window. Clients must serialize refresh requests; concurrent reuse may invalidate the winning response too. Unknown/mismatched/expired tokens do not revoke unrelated sessions.
+
+Family revocation affects refresh capability. Already issued access JWTs remain usable until their normal expiry; this change does not claim family-scoped immediate access revocation. Password reset and administrator role changes invalidate access through the user token version.
+
+Historical refresh-token ancestry is unknown. The additive migration leaves historical family IDs null and revokes historical refresh tokens. Users must log in again; existing access tokens retain their previous expiry/version behavior. Do not backfill guessed families. During deployment quiesce old application instances before migrating and restart only upgraded instances: old binaries do not enforce family revocation. Retain revoked token rows for replay detection through their JWT expiry. No cleanup is part of this change.
+
+Password reset serializes against rotation and login using the user row lock. Password hash, all refresh revocations and token-version increment commit atomically. Failed transactions preserve the old credentials. Login rechecks the current password under the same lock, so a pending login cannot issue a new family based on pre-reset credentials.
+
+Access authentication reads role/version from the primary database on every request; Redis auth-context entries are no longer trusted or populated by the guard. This deliberately trades caching performance for a cross-instance revocation guarantee. Redis eviction failures and stale fills cannot extend access. Database failure denies authorization. A request whose authoritative check completed before reset committed can already be in flight; this does not cancel running handlers. New checks after commit reject the old version. Deployment must replace all old guard instances; no replica lag is permitted for this lookup.
+
+Administrator role changes acquire PostgreSQL transaction advisory lock `(1835101729, 1)` before reading or mutating the administrator set. The existing transaction includes count, role change, refresh-family/token revocation, version increment and audit insertion. Rollback/crash releases the lock. No automatic transaction retries are introduced. All future runtime role-mutation or account-deletion paths must share this lock and check the invariant; Phase 1 and the implementation check found no runtime user-deletion endpoint. Direct DBA writes and bootstrap seed scripts are outside the HTTP invariant and must not be used as account-removal APIs.

docs/contracts/escrow-token-verification.md

@@ -0,0 +1,35 @@
+# Escrow token verification
+
+`Escrow.isVerified` records acceptance by the token policy when `escrow_funded`
+is ingested. It is off-chain state, separate from `User.emailVerifiedAt`, wallet
+signature verification and driver KYC. It is not a transaction authorization
+credential. The chain reader cannot supply the local allowlist decision.
+
+The introducing change (`7ebc2b5`, issue #383) persisted the decision and filtered
+released GMV by it. The schema field and repository write were lost in merge
+`5c96b86`; the policy, domain field and analytics predicate survived. No historical
+migration added `is_verified`. Restore the persisted contract rather than remove
+the GMV restriction or derive historical acceptance from today's configuration.
+
+New ingestion keeps the existing policy: a configured allowlist decides membership;
+without one, the existing compatibility policy accepts tokens. This repair does
+not change that configuration policy. Repository callers without a decision and
+rows predating the migration default to false. False means acceptance has not
+been established, including unknown historical state. Only explicit true enters
+released GMV. Status updates retain the stored decision. Changing the allowlist
+does not retroactively reclassify stored rows.
+
+No reliable historical allowlist snapshots or per-row decisions are available.
+The additive migration therefore excludes historical rows from GMV; it does not
+infer truth from token names, release status, email or wallet timestamps. Any
+future historical reclassification needs separately reviewed reliable evidence.
+The public escrow DTO continues to omit this internal flag. Authentication,
+role restrictions, email verification and wallet verification are unchanged.
+
+Before rollout, inspect redacted target migration history and catalog. Apply the
+new migration through the approved process before starting a matching generated
+Prisma client/binary. An unexpected existing column is schema drift requiring
+investigation, not a reason to replay or overwrite its values. Preserve all
+applied migrations and stored decisions during recovery. Fresh replay currently
+fails earlier at the duplicate analytics index; a component upgrade test does
+not certify the full chain or a representative deployment upgrade.

docs/contracts/mp4-evidence-header.md

@@ -0,0 +1,26 @@
+# MP4 evidence header recognition
+
+Dispute upload checks bytes before hashing, saving or inserting evidence. The
+service, signature helper and exported schema helper share `matchesMp4Header`.
+The HTTP schema still checks the MIME allowlist and upload sizes; the upload
+service performs the header check after the existing dispute/ownership guards.
+No file extension or client MIME declaration substitutes for matching bytes.
+
+The recognizer follows the basic box and FileTypeBox syntax in
+[ISO/IEC 14496-12:2008, sections 4.2–4.3](https://cdn.standards.iteh.ai/samples/51533/c8233bd62f144bdda13c99156e984616/ISO-IEC-14496-12-2008.pdf).
+The leading box type is at byte4, after its size. Standard, extended-size and
+size-to-EOF forms are supported. The complete declared box must fit the supplied
+view, contain major-brand/minor-version fields and an aligned array of printable
+four-character brands. Extended sizes are compared exactly with bigint.
+
+This is bounded container-header recognition, **not whole-video validation or a
+content-safety guarantee**. It does not scan for a marker elsewhere, traverse
+preceding boxes, verify a brand registry/profile, inspect later boxes or decode
+tracks/codecs. A matching header alone does not establish that a file is a valid,
+playable or safe video. A leading `ftyp` remains the supported recognition scope.
+
+Malformed/truncated MP4 headers retain the existing400/VALIDATION_ERROR response
+before storage or metadata writes. Authorization, stable evidence ownership,
+SHA256 of the exact uploaded bytes, configured decoded/base64/body limits and
+storage/failure behavior are unchanged. Other MIME checks are unchanged. Worker
+execution remains restricted; this repair does not enable any background job.

prisma/migrations/20261030120000_session_families/migration.sql

@@ -0,0 +1,10 @@
+-- No historical ancestry is known: legacy tokens stay unassigned and are invalidated.
+CREATE TABLE "session_families" (
+  "id" TEXT NOT NULL PRIMARY KEY,
+  "user_id" TEXT NOT NULL REFERENCES "users"("id") ON DELETE CASCADE ON UPDATE CASCADE,
+  "revoked_at" TIMESTAMP(3)
+);
+CREATE INDEX "session_families_user_id_idx" ON "session_families"("user_id");
+ALTER TABLE "refresh_tokens" ADD COLUMN "family_id" TEXT REFERENCES "session_families"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+CREATE INDEX "refresh_tokens_family_id_idx" ON "refresh_tokens"("family_id");
+UPDATE "refresh_tokens" SET "revoked_at" = CURRENT_TIMESTAMP, "updated_at" = CURRENT_TIMESTAMP WHERE "revoked_at" IS NULL;

prisma/migrations/20261103120000_escrow_token_verification/migration.sql

@@ -0,0 +1,4 @@
+-- Restore the escrow token-policy result introduced in 7ebc2b5, lost at merge.
+-- No historical allowlist snapshot exists: unknown rows must not contribute to GMV.
+-- Do not infer verification from RELEASED status, token name, or user timestamps.
+ALTER TABLE "escrows" ADD COLUMN "is_verified" BOOLEAN NOT NULL DEFAULT false;

prisma/schema.prisma

@@ -138,6 +138,7 @@ model User {
 
   wallets          WalletAddress[]
   refreshTokens    RefreshToken[]
+  sessionFamilies  SessionFamily[]
   notifications    Notification[]
   auditLogs        AuditLog[]
   ownedFleets      Fleet[]
@@ -147,18 +148,32 @@ model User {
   @@map("users")
 }
 
+model SessionFamily {
+  id        String         @id @default(uuid())
+  userId    String         @map("user_id")
+  revokedAt DateTime?      @map("revoked_at")
+  user      User           @relation(fields: [userId], references: [id], onDelete: Cascade)
+  tokens    RefreshToken[]
+
+  @@index([userId])
+  @@map("session_families")
+}
+
 model RefreshToken {
-  id        String    @id @default(uuid())
-  userId    String    @map("user_id")
-  tokenHash String    @unique @map("token_hash")
-  expiresAt DateTime  @map("expires_at")
-  revokedAt DateTime? @map("revoked_at")
-  createdAt DateTime  @default(now()) @map("created_at")
-  updatedAt DateTime  @updatedAt @map("updated_at")
+  familyId  String?        @map("family_id")
+  family    SessionFamily? @relation(fields: [familyId], references: [id], onDelete: Cascade)
+  id        String         @id @default(uuid())
+  userId    String         @map("user_id")
+  tokenHash String         @unique @map("token_hash")
+  expiresAt DateTime       @map("expires_at")
+  revokedAt DateTime?      @map("revoked_at")
+  createdAt DateTime       @default(now()) @map("created_at")
+  updatedAt DateTime       @updatedAt @map("updated_at")
 
   user User @relation(fields: [userId], references: [id], onDelete: Cascade)
 
   @@index([userId])
+  @@index([familyId])
   @@map("refresh_tokens")
 }
 
@@ -244,6 +259,9 @@ model Escrow {
   releasedAt       DateTime?    @map("released_at")
   refundedAt       DateTime?    @map("refunded_at")
   createdAtChain   DateTime     @map("created_at_chain")
+  /// Token accepted by the ingestion allowlist policy; unrelated to user verification.
+  /// False also covers historical rows with no recorded verification evidence.
+  isVerified       Boolean      @default(false) @map("is_verified")
   createdAt        DateTime     @default(now()) @map("created_at")
   updatedAt        DateTime     @updatedAt @map("updated_at")
 
@@ -475,6 +493,7 @@ model Notification {
   type          String
   payload       Json
   status        NotificationStatus  @default(PENDING)
+  readAt        DateTime?           @map("read_at")
   sentAt        DateTime?           @map("sent_at")
   failureReason String?             @map("failure_reason")
   attempts      Int                 @default(0)

src/blockchain/xdr/sc-val.spec.ts

@@ -1,6 +1,7 @@
 import { Address, Keypair, xdr } from '@stellar/stellar-sdk';
 import { describe, expect, it } from 'vitest';
 import {
+  i128ToScVal,
   addressToScVal,
   boolToScVal,
   namedStructToScVal,
@@ -156,3 +157,15 @@ describe('native -> ScVal encoders', () => {
     });
   });
 });
+
+describe('signed i128 financial encoding', () => {
+  it.each([-(1n << 127n), (1n << 127n) - 1n, -1n, 0n, 1n, 9007199254740993n])(
+    'round-trips exactly %s',
+    (value) => {
+      expect(scValToNative(i128ToScVal(value))).toBe(value.toString());
+    },
+  );
+  it.each([-(1n << 127n) - 1n, 1n << 127n])('rejects overflow %s instead of wrapping', (value) => {
+    expect(() => i128ToScVal(value)).toThrow('signed 128-bit bounds');
+  });
+});

src/blockchain/xdr/sc-val.ts

@@ -1,4 +1,5 @@
 import { Address, xdr } from '@stellar/stellar-sdk';
+import { assertI128 } from '../../shared/validation/i128.js';
 
 /**
  * Converts a Soroban `xdr.ScVal` into a plain JS value. Covers the variants
@@ -118,6 +119,7 @@ export function u64ToScVal(value: bigint | number): xdr.ScVal {
  * always non-negative in practice, but the Rust type is signed (`i128`), so
  * this accepts any bigint rather than assuming non-negativity. */
 export function i128ToScVal(value: bigint): xdr.ScVal {
+  assertI128(value);
   const mask64 = (1n << 64n) - 1n;
   const lo = value & mask64;
   const hi = value >> 64n;

src/modules/admin/infrastructure/admin-security.integration.spec.ts

@@ -0,0 +1,98 @@
+import { randomUUID } from 'node:crypto';
+import { PrismaClient } from '@prisma/client';
+import { afterAll, afterEach, describe, expect, it } from 'vitest';
+import { createUpdateUserRoleUseCase } from '../application/update-user-role.js';
+import { createPrismaAdminUnitOfWork } from './prisma-admin-unit-of-work.js';
+
+const prisma = new PrismaClient();
+const ids: string[] = [];
+afterEach(async () => {
+  await prisma.auditLog.deleteMany({ where: { actorId: { in: ids } } });
+  await prisma.user.deleteMany({ where: { id: { in: ids } } });
+  ids.length = 0;
+});
+afterAll(async () => {
+  await prisma.$disconnect();
+});
+const change = createUpdateUserRoleUseCase({
+  unitOfWork: createPrismaAdminUnitOfWork(prisma),
+  authCacheInvalidator: { async invalidateUser() {} },
+});
+
+async function admins() {
+  // This test requires a disposable database, not a shared account catalog.
+  expect(await prisma.user.count({ where: { role: 'ADMIN' } })).toBe(0);
+  const first = await prisma.user.create({
+    data: { email: `${randomUUID()}@admin.invalid`, passwordHash: 'fixture', role: 'ADMIN' },
+  });
+  const second = await prisma.user.create({
+    data: { email: `${randomUUID()}@admin.invalid`, passwordHash: 'fixture', role: 'ADMIN' },
+  });
+  ids.push(first.id, second.id);
+  return { first, second };
+}
+
+describe('administrator set invariant on PostgreSQL', () => {
+  it.each([1, 2, 3, 4, 5])(
+    'preserves one administrator under simultaneous mutual demotions, round %s',
+    async () => {
+      const { first, second } = await admins();
+      const results = await Promise.allSettled([
+        change({ actorId: first.id, userId: second.id, role: 'CUSTOMER' }),
+        change({ actorId: second.id, userId: first.id, role: 'CUSTOMER' }),
+      ]);
+      expect(results.filter((x) => x.status === 'fulfilled')).toHaveLength(1);
+      expect(results.filter((x) => x.status === 'rejected')).toHaveLength(1);
+      expect(await prisma.user.count({ where: { role: 'ADMIN' } })).toBe(1);
+      expect(await prisma.auditLog.count({ where: { actorId: { in: ids } } })).toBe(1);
+      const demoted = await prisma.user.findFirstOrThrow({
+        where: { id: { in: ids }, role: 'CUSTOMER' },
+      });
+      expect(demoted.tokenVersion).toBe(1);
+    },
+  );
+
+  it('rolls back role, sessions and audit and releases the invariant lock', async () => {
+    const { first, second } = await admins();
+    const family = await prisma.sessionFamily.create({ data: { userId: second.id } });
+    const refresh = await prisma.refreshToken.create({
+      data: {
+        userId: second.id,
+        familyId: family.id,
+        tokenHash: randomUUID(),
+        expiresAt: new Date(Date.now() + 60000),
+      },
+    });
+    const real = createPrismaAdminUnitOfWork(prisma);
+    const failing = createUpdateUserRoleUseCase({
+      unitOfWork: {
+        run: (work) =>
+          real.run(async (ports) => {
+            await work(ports);
+            throw new Error('precommit failure');
+          }),
+      },
+      authCacheInvalidator: { async invalidateUser() {} },
+    });
+    await expect(
+      failing({ actorId: first.id, userId: second.id, role: 'CUSTOMER' }),
+    ).rejects.toThrow('precommit failure');
+    expect(await prisma.user.findUnique({ where: { id: second.id } })).toMatchObject({
+      role: 'ADMIN',
+      tokenVersion: 0,
+    });
+    expect(await prisma.refreshToken.findUnique({ where: { id: refresh.id } })).toMatchObject({
+      revokedAt: null,
+    });
+    expect(await prisma.sessionFamily.findUnique({ where: { id: family.id } })).toMatchObject({
+      revokedAt: null,
+    });
+    expect(await prisma.auditLog.count({ where: { actorId: { in: ids } } })).toBe(0);
+    await expect(
+      change({ actorId: first.id, userId: second.id, role: 'CUSTOMER' }),
+    ).resolves.toMatchObject({ role: 'CUSTOMER' });
+    expect(await prisma.sessionFamily.findUnique({ where: { id: family.id } })).toMatchObject({
+      revokedAt: expect.any(Date),
+    });
+  });
+});

src/modules/admin/infrastructure/prisma-admin-unit-of-work.ts

@@ -16,13 +16,16 @@ import { createPrismaUserRoleRepository } from './prisma-user-role-repository.js
 export function createPrismaAdminUnitOfWork(prisma: PrismaClient): AdminUnitOfWork {
   return {
     async run(work) {
-      return prisma.$transaction(async (tx) =>
-        work({
+      return prisma.$transaction(async (tx) => {
+        // Serialize the global administrator set before reading its count.
+        // Transaction-scoped: rollback/crash releases the lock automatically.
+        await tx.$queryRaw`SELECT pg_advisory_xact_lock(1835101729, 1)::text`;
+        return work({
           userRoleRepository: createPrismaUserRoleRepository(tx),
           auditLogRepository: createPrismaAuditLogRepository(tx),
           sessionRevoker: createPrismaSessionRevoker(tx),
-        }),
-      );
+        });
+      });
     },
   };
 }