Stream versioned user settings updates across devices

.github/workflows/ci.yml

@@ -97,8 +97,17 @@ jobs:
       - name: Run API tests (DB-less)
         run: pnpm --filter @uoa/api exec vitest run --exclude 'tests/integration/**'
 
-      - name: Run API integration tests (Postgres)
-        run: pnpm --filter @uoa/api exec vitest run tests/integration
+      # Schemas do not isolate PostgreSQL advisory locks. This suite deliberately
+      # holds a shared product-policy lock while filling the admin pool; an
+      # exclusive policy writer from another suite can queue ahead of its readers.
+      - name: Run pool contention integration tests (Postgres)
+        run: pnpm --filter @uoa/api exec vitest run tests/integration/org-me-pool-contention.test.ts
+        env:
+          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
+          BILLING_FUNDING_DATABASE_TESTS: 'true'
+
+      - name: Run remaining API integration tests (Postgres)
+        run: pnpm --filter @uoa/api exec vitest run tests/integration --exclude 'tests/integration/org-me-pool-contention.test.ts'
         env:
           DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
           BILLING_FUNDING_DATABASE_TESTS: 'true'

API/package.json

@@ -45,6 +45,7 @@
     "jose": "^5.9.6",
     "nodemailer": "^10.0.6",
     "pdf-lib": "^1.17.1",
+    "pg": "8.23.1",
     "qrcode": "^1.5.4",
     "reflect-metadata": "^0.2.2",
     "stripe": "22.3.2",
@@ -56,13 +57,15 @@
     "@redocly/cli": "^2.40.0",
     "@types/node": "^22.13.1",
     "@types/nodemailer": "^6.4.17",
+    "@types/pg": "8.23.1",
     "@types/qrcode": "^1.5.6",
     "@types/ws": "^8.18.1",
     "@types/xml-encryption": "^1.2.4",
     "prisma": "^6.3.1",
     "tsx": "^4.19.2",
     "typescript": "^5.7.3",
     "vitest": "^3.2.6",
+    "ws": "8.21.0",
     "xml-crypto": "^6.4.0",
     "xml-encryption": "^3.1.0"
   }

API/prisma/migrations/20261104160000_settings_change_notifications/migration.sql

@@ -0,0 +1,23 @@
+SET lock_timeout = '5s';
+SET statement_timeout = '120s';
+
+-- Commit-scoped hints for every settings writer. Values never enter NOTIFY.
+CREATE OR REPLACE FUNCTION notify_user_setting_change() RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+  IF TG_OP = 'UPDATE' AND OLD.value IS NOT DISTINCT FROM NEW.value THEN
+    RETURN NEW;
+  END IF;
+  IF TG_OP = 'DELETE' THEN
+    PERFORM pg_notify('uoa_settings_changed', json_build_object(
+      'userId', OLD.user_id, 'namespace', OLD.namespace, 'key', OLD.key)::text);
+    RETURN OLD;
+  END IF;
+  PERFORM pg_notify('uoa_settings_changed', json_build_object(
+    'userId', NEW.user_id, 'namespace', NEW.namespace, 'key', NEW.key)::text);
+  RETURN NEW;
+END;
+$$;
+CREATE TRIGGER user_settings_changed
+AFTER INSERT OR UPDATE OR DELETE ON user_settings
+FOR EACH ROW EXECUTE FUNCTION notify_user_setting_change();

API/src/routes/oauth/index.ts

@@ -1,3 +1,4 @@
+import { registerSettingsEvents } from './settings-events.js';
 import { registerOAuthLifecycleStatus } from './lifecycle-status.js';
 import type { FastifyInstance } from 'fastify';
 import { registerNativeAppIcon } from './native-app-icon.js';
@@ -20,6 +21,7 @@ export function registerOAuthRoutes(app: FastifyInstance): void {
   registerNativeAppIcon(app);
   registerPublicSocialRoutes(app);
   registerOAuthAccountRoutes(app);
+  registerSettingsEvents(app);
   registerKelpieConnectRoutes(app);
   registerKelpiePairRoutes(app);
   registerOAuthMetadataRoute(app);

API/src/routes/oauth/settings-events.ts

@@ -0,0 +1,121 @@
+import type { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify';
+import type { WebSocket } from 'ws';
+import { z } from 'zod';
+import { getEnv } from '../../config/env.js';
+import { verifyPublicAccountToken, authorizePublicAccount } from '../../services/oauth/account.service.js';
+import { settingVersion } from '../../services/setting-version.js';
+import { settingsDelta } from '../../services/settings-delta.js';
+import { SettingsNotifications, type SettingsTopic } from '../../services/settings-notifications.js';
+import { SettingsStreamQueue } from '../../services/settings-stream-queue.js';
+import { USER_SETTINGS_NAMESPACE_PATTERN, USER_SETTINGS_KEY_PATTERN } from '../../services/user-settings.service.js';
+import { AppError } from '../../utils/errors.js';
+import { requireMcpOAuthPublicProfile } from './public-profile-guard.js';
+
+const paramsSchema = z.object({ namespace: z.string().regex(USER_SETTINGS_NAMESPACE_PATTERN), key: z.string().regex(USER_SETTINGS_KEY_PATTERN) });
+const MAX_SOCKETS = 32; // Live Cloud Run request concurrency is 80; leave capacity for sign-in.
+
+export function registerSettingsEvents(app: FastifyInstance): void {
+  const lane = new SettingsStreamQueue();
+  const subjects = new Map<string, number>();
+  const reservations = new WeakMap<FastifyRequest, SettingsTopic>();
+  const sockets = new Set<WebSocket>();
+  let admitted = 0;
+  let notifications: SettingsNotifications | undefined;
+  const release = (userId: string) => {
+    admitted -= 1;
+    const remaining = (subjects.get(userId) ?? 1) - 1;
+    if (remaining > 0) subjects.set(userId, remaining); else subjects.delete(userId);
+  };
+  const reserve = async (request: FastifyRequest, reply: FastifyReply) => {
+    reply.header('Retry-After', '5');
+    if (request.headers.upgrade?.toLowerCase() !== 'websocket') throw new AppError('BAD_REQUEST', 400, 'INVALID_REQUEST');
+    const params = paramsSchema.parse(request.params);
+    z.object({}).strict().parse(request.query);
+    if (admitted >= MAX_SOCKETS) throw new AppError('SERVICE_UNAVAILABLE', 503, 'SETTINGS_STREAM_BUSY');
+    admitted += 1;
+    try {
+      const account = await lane.run(() => verifyPublicAccountToken(request.headers.authorization, 'settings.read'));
+      const count = subjects.get(account.userId) ?? 0;
+      if (count >= 8) throw new AppError('SERVICE_UNAVAILABLE', 503, 'SETTINGS_STREAM_BUSY');
+      if (request.raw.socket.destroyed) throw new Error('Connection closed');
+      subjects.set(account.userId, count + 1);
+      reservations.set(request, { userId: account.userId, ...params });
+      request.raw.socket.once('close', () => {
+        const pending = reservations.get(request);
+        if (pending) { reservations.delete(request); release(pending.userId); }
+      });
+    } catch (error) { admitted -= 1; throw error; }
+  };
+  app.get('/oauth/me/settings/:namespace/:key/events', {
+    websocket: true,
+    preHandler: [requireMcpOAuthPublicProfile, reserve],
+  }, (socket, request) => {
+    const topic = reservations.get(request);
+    if (!topic) { socket.close(1011); return; }
+    reservations.delete(request);
+    sockets.add(socket);
+    let closed = false;
+    let pending = false;
+    let dirty = false;
+    let ready = false;
+    let previous: { value: unknown; version: string } | undefined;
+    let unsubscribe: (() => void) | undefined;
+    const cleanup = () => {
+      if (closed) return;
+      closed = true;
+      sockets.delete(socket);
+      release(topic.userId);
+      clearInterval(repair); clearTimeout(recycle);
+      unsubscribe?.();
+    };
+    const check = () => {
+      if (closed || !ready) return;
+      if (pending) { dirty = true; return; }
+      pending = true;
+      void lane.run(async () => {
+        if (closed) return;
+        // Recheck current client/scopes as well as live subject authority.
+        const account = await verifyPublicAccountToken(request.headers.authorization, 'settings.read');
+        if (account.userId !== topic.userId) throw new Error('Subject changed');
+        const value = await request.adminDb.$transaction(async tx => {
+          await authorizePublicAccount(account, tx);
+          const row = await tx.userSetting.findUnique({ where: { userId_namespace_key: topic }, select: { value: true } });
+          return row?.value ?? null;
+        });
+        if (closed) return;
+        const version = settingVersion(value);
+        if (version === previous?.version) { socket.ping(); return; }
+        if (socket.bufferedAmount > 1024 * 1024) { socket.close(1013); return; }
+        const frame = previous
+          ? { type: 'settings.delta', protocolVersion: 1, namespace: topic.namespace, key: topic.key,
+            baseVersion: previous.version, version, operation: settingsDelta(previous.value, value) }
+          : { type: 'settings.snapshot', protocolVersion: 1, namespace: topic.namespace, key: topic.key, version, value };
+        socket.send(JSON.stringify(frame), error => { if (error) socket.close(1011); });
+        previous = { value, version };
+      }).catch(() => { if (!closed) socket.close(1008); }).finally(() => {
+        pending = false;
+        if (dirty && !closed) { dirty = false; check(); }
+      });
+    };
+    const repair = setInterval(check, 30_000);
+    const recycle = setTimeout(() => socket.close(1000), 240_000 + Math.random() * 40_000);
+    repair.unref(); recycle.unref();
+    socket.on('close', cleanup);
+    socket.on('error', cleanup);
+    socket.on('message', () => socket.close(1008));
+    void (async () => {
+      const env = getEnv();
+      const databaseUrl = env.DATABASE_ADMIN_URL ?? env.DATABASE_URL;
+      if (!databaseUrl) throw new Error('Database unavailable');
+      notifications ??= new SettingsNotifications(databaseUrl);
+      unsubscribe = await notifications.subscribe(topic, check);
+      if (closed) { unsubscribe(); return; }
+      ready = true;
+      check();
+    })().catch(() => socket.close(1013));
+  });
+  app.addHook('onClose', async () => {
+    sockets.forEach(socket => socket.close(1001));
+    await notifications?.stop();
+  });
+}

API/src/routes/root/index.ts

@@ -70,6 +70,7 @@ export function registerRootRoute(app: FastifyInstance): void {
         sms_admission: '/apps/v1/nessie/phone-profile/sms-admission',
         personal_owner_membership: 'optional owner_subject in service assertion; exact-team ownerActive or owner_inactive',
         prepaid_delivery: '/apps/v1/nessie/phone-profile/call-proof/reserve', caller_id_authentication: false },
+      settings_events: { path: '/oauth/me/settings/:namespace/:key/events', transport: 'WebSocket', scope: 'settings.read', protocol_version: 1, recovery: 'snapshot on reconnect; version-fenced splice or replace deltas' },
       native_apps: { admin: '/admin/apps', registration: '/oauth/register', public_identifier: 'app_id', client_secret_required: false },
       enterprise_sso: { version: 1, settings: '/auth/organisation-access',
         entry: '/sso/:connectionId/start', domain_discovery: false,

API/src/routes/root/llm.ts

@@ -39,6 +39,8 @@ GET /auth/social/x starts X OAuth 2.0 authorization-code sign-in with S256 PKCE
 Facebook's basic profile has no reliable verified-email claim. Its first callback sends a UOA code to the claimed mailbox and opens flow=facebook_proof. POST /auth/facebook-proof/verify with {code} and the signed HttpOnly proof cookie returns a one-use same-origin callback URL after mailbox proof. UOA then binds Facebook's stable subject to the verified account; later logins require that exact binding and current email, or fresh mailbox proof. Five wrong codes or expiry fail generically.
 ## Billing provider availability
 Billing calls that cannot resolve the configured payment-provider account return 503 BILLING_PAYMENT_PROVIDER_UNAVAILABLE. Preserve the user session and report billing temporarily unavailable. Retry reads with backoff; never substitute zero credits, infer insufficient funds, or admit paid work from this response. Payment mutations keep their existing idempotency and reconciliation requirements.
+## Personal settings live updates
+Native public OAuth clients with settings.read can open GET /oauth/me/settings/:namespace/:key/events as a WebSocket using the Authorization header. No query tokens or user selectors. protocolVersion 1 sends a settings.snapshot with topic, opaque version and value, then settings.delta with baseVersion, version and operation. Operations are root-array splice(index,deleteCount,items) or replace(value). Apply only to the exact baseVersion; discard malformed frames and reconnect for a fresh snapshot. Versions support equality only. Writes continue through conditional PUT; this stream cannot mutate settings. Reconnect with backoff and jitter; normal recycling happens after 240–280 seconds. Live authority is checked before every delivery and every 30 seconds. No historical replay guarantee; rapid changes may coalesce. Busy subscriptions receive 503 and Retry-After.
 ## Enterprise organisation access (version1)
 UOA alone owns enterprise settings at /auth/organisation-access. Products read GET /org/organisations/:orgId/sso/status and owners request POST its /management-continuation subroute; domain backend authority never substitutes for fresh independent owner proof. The same-origin cookie/CSRF management API reserves a draft callback/SP identity before credentials are configured. OIDC discovery never widens the server-owned secret destination allowlist. Tests use real browser-bound protocol callbacks and never provision/link users or issue product credentials. Live activation requires verified authority/provenance/revocation readiness; enforcement also requires two independently authenticated custodians and a completed recovery drill. One custodian must be the canonical owner; no backup codes or global account-recovery capability are granted. No Nessie billing gate exists. Dedicated scoped SCIM resources never link global users by email/externalId. Provider tiles/bookmarks use /sso/:connectionId/start and the relying product’s own PKCE/state flow. Pre-authentication domain routing is deferred: no discovery endpoint, owner checkbox or writable discovery policy ships. Operators deploy genuine fresh proof issuance and the global validation contract before a strict consumer. Prepare the exact product’s assurance floor and fixed entry/config URLs as a separately reviewed legacy cutover; migration/revocation revisions may remain explicitly null during preparation, which cannot activate SSO or SCIM. Record audited revisions and the actual strict receiver acknowledgement only after consumer verification.
 Backend token responses after new completed independent authorization carry assurance_ref and assurance_expires_at even before server-owned minimum-floor enrollment. Existing proof-bearing families retain that same original reference and absolute deadline on refresh; legacy families without original evidence remain legacy and cannot upgrade through refresh. Explicit encrypted custody is required and missing custody fails closed. SSO/recovery activation gates are unchanged. Server-enrolled assurance-version1 clients require proof; only the backend retains them encrypted. Human subject and billing assertions include top-level assurance_ref. Signed UOA access tokens carry opaque assurance_id, not the custody reference. POST /org/assurance/validate accepts the original backend reference with its fresh confidential subject assertion (or existing UOA access-token transport). Omit org_id/team_id only for an independent original whose scope is exactly null; team_id requires org_id. The person credential must resolve the same assurance as the supplied reference, even below the compatibility floor. No public assurance_id introspection is provided. Every reference names exact client/subject/epoch/org/team/family and original expiry; missing proof cannot downgrade an enrolled product. GET /domain/revocations is an authenticated ordered exact-client feed; persist committed handling before advancing its opaque cursor. See /api for exact routes and Docs/Auth/enterprise-sso.md for deployment gates and remaining work.

API/src/routes/root/schema.oauth-account.ts

@@ -23,6 +23,10 @@ export const oauthAccountEndpoints: EndpointSchema[] = [
     response: { 200: 'image bytes; no-store' } },
   { method: 'GET', path: '/oauth/me/settings/:namespace/:key', description: 'Read an opaque personal setting. settings.read scope required.', auth,
     response: { value: 'JSON value or null when absent', ETag: 'response header; opaque version for conditional writes' } },
+  { method: 'GET', path: '/oauth/me/settings/:namespace/:key/events', description: 'WebSocket subscription to one current-user setting; settings.read scope required.', auth,
+    response: { snapshot: 'settings.snapshot: protocolVersion=1, namespace, key, version (opaque ETag), value',
+      delta: 'settings.delta: same topic, baseVersion, version, operation: replace(value) or splice(index,deleteCount,items)' },
+    notes: 'Authorization header only; no query parameters. LISTEN before initial snapshot. Revalidates live authority before each delivery and every 30 seconds. Reconnect with jitter after close to obtain a fresh snapshot; reject a delta whose baseVersion does not match. No durable replay cursor or event-history guarantee. Limits: 32 streams/instance, 8/subject/instance; busy returns 503 with Retry-After. Connections recycle after 240–280 seconds.' },
   { method: 'PUT', path: '/oauth/me/settings/:namespace/:key', description: 'Replace or delete one personal setting. settings.write scope required.', auth,
     body: { value: 'JSON value; null deletes' }, response: { value: 'saved JSON value or null' },
     notes: 'Requires If-Match with the ETag from GET. Missing/invalid precondition: 428; stale value: 409. Same store and quotas as /settings/me. Never supply a user id. No-store responses.' },

API/src/services/settings-delta.ts

@@ -0,0 +1,21 @@
+import { settingVersion } from './setting-version.js';
+
+export type SettingsOperation =
+  | { type: 'replace'; value: unknown }
+  | { type: 'splice'; index: number; deleteCount: number; items: unknown[] };
+
+/** One bounded root-array edit. Other JSON values deliberately replace whole. */
+export function settingsDelta(before: unknown, after: unknown): SettingsOperation {
+  const replacement: SettingsOperation = { type: 'replace', value: after };
+  if (!Array.isArray(before) || !Array.isArray(after) || before.length + after.length > 20_000) return replacement;
+  let first = 0;
+  const equal = (a: unknown, b: unknown) => settingVersion(a) === settingVersion(b);
+  while (first < before.length && first < after.length && equal(before[first], after[first])) first += 1;
+  let oldEnd = before.length;
+  let newEnd = after.length;
+  while (oldEnd > first && newEnd > first && equal(before[oldEnd - 1], after[newEnd - 1])) {
+    oldEnd -= 1; newEnd -= 1;
+  }
+  const splice: SettingsOperation = { type: 'splice', index: first, deleteCount: oldEnd - first, items: after.slice(first, newEnd) };
+  return Buffer.byteLength(JSON.stringify(splice)) < Buffer.byteLength(JSON.stringify(replacement)) ? splice : replacement;
+}