From 3d1f0855cffef62f7498c40fec26be4d4b97e3f6 Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Thu, 23 Jul 2026 22:58:08 -0700 Subject: [PATCH 1/4] feat(cloud): broadcast-signal transport with capability probe Pre-commit hook ran. Total eslint: 1, total circular: 0 --- docs/h4-h5-design-2026-07-24/h4-broadcast.md | 90 +++++ .../h5-storage-offload.md | 72 ++++ .../Org2Cloud/org2CloudCapabilities.test.ts | 72 ++++ .../Org2Cloud/org2CloudCapabilities.ts | 62 ++++ src/features/Org2Cloud/org2CloudClient.ts | 10 + .../Org2Cloud/org2CloudControlBus.test.ts | 12 + src/features/Org2Cloud/org2CloudControlBus.ts | 29 ++ .../Org2Cloud/org2CloudRealtimeClient.test.ts | 18 + .../Org2Cloud/org2CloudRealtimeClient.ts | 10 +- .../Org2Cloud/useOrg2CloudRealtime.ts | 311 +++++++++++------- 10 files changed, 573 insertions(+), 113 deletions(-) create mode 100644 docs/h4-h5-design-2026-07-24/h4-broadcast.md create mode 100644 docs/h4-h5-design-2026-07-24/h5-storage-offload.md create mode 100644 src/features/Org2Cloud/org2CloudCapabilities.test.ts create mode 100644 src/features/Org2Cloud/org2CloudCapabilities.ts diff --git a/docs/h4-h5-design-2026-07-24/h4-broadcast.md b/docs/h4-h5-design-2026-07-24/h4-broadcast.md new file mode 100644 index 000000000..655d1a667 --- /dev/null +++ b/docs/h4-h5-design-2026-07-24/h4-broadcast.md @@ -0,0 +1,90 @@ +# H4 — Change Signals via Broadcast-from-Database + +Date: 2026-07-24. Basis: realtime signal-chain architecture map (agent survey +of 0001/0003 server path + full client subscription topology). Companion: +`h5-storage-offload.md`. + +## Why + +`postgres_changes` evaluates RLS per change × per subscriber on a single +shared WAL poller — the platform's first scalability wall (server-schema-audit +H4). The client consumes **zero row data** from any of its three +postgres_changes subscriptions (all onChange callbacks are 0-arg edges), so +the entire contract is "something in scope changed" — a perfect fit for +fire-and-forget broadcast with join-time authorization. + +## Decision summary + +| Question | Decision | Why | +| ------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------- | -------- | --------- | ------ | -------------------------------------------------------------------------------------------- | +| Topic | **Reuse the existing private org channel `presence:org:`** | Already open on every active client, already authorized by the `realtime.messages` policies (0001:5694-5706), already carries client-side broadcast events (`org-control-changed` etc.). Server-side `realtime.send` from a definer trigger needs no new policy. Channels per client drop 4 → 2. | +| New event | `org-db-changed` with payload `{kind}` | Distinct from client-sent `org-control-changed`. `kind` ∈ `sessions | comments | projects | workItems | roster | policy`derived from`TG_TABLE_NAME` / call site — the discriminator today's signal row lacks. | +| Debounce | Broadcast fires **only when the debounced signal-row bump actually fired** (the 0003 250ms `WHERE` clause) — per (org, kind)? No: per org, same as today | Identical delivery semantics to the postgres_changes transport; the client's 60s coarse throttle and 5min control TTL stay unchanged. | +| Slice A (self-roster eviction) | **Stays postgres_changes** on `org_memberships` `user_id=eq.` | A member removed while disconnected can never be reached by join-time-auth broadcast (`is_org_member()` already false). Exactly the audit's recommendation. | +| Slice B (org signal) + B-roster (org memberships) | Replaced by broadcast events on the org channel when the backend supports it | The two org-scoped postgres_changes channels disappear; SUBSCRIBED-edge recovery bookkeeping moves to the org channel's join edge. | +| Old signal row / publication | **Kept and still bumped** (org_change_signals stays in the publication) | Old clients keep working; removal is a later cleanup migration once the fleet has upgraded. | +| Capability detection | New RPC `get_cloud_capabilities()` → `{broadcastSignals: true}`; client probes once per endpoint, PGRST202 ⇒ legacy | Same probe-and-remember pattern as the 0004 paged listings. | +| Roster nudges for members | New AFTER trigger on `org_memberships` → send `kind:'roster'` to the org topic | Replaces the org-filtered memberships channel for present members; the removed user is covered by Slice A. | +| Policy nudges | The 3 RPCs with direct in-RPC bumps (`cloud_rename_org`, `cloud_set_org_sharing_floor`, `cloud_set_member_sharing_floor`) recreated with `kind:'policy'` send | They bypass the trigger today; same bypass, now with a broadcast. | + +## Server (cloud-infra 0005) + +1. Helper `org2_cloud.nudge_org_signal(p_org_id uuid, p_kind text)`: + debounced signal-row upsert (unchanged 250ms suppress) and, **iff the + bump fired**, `realtime.send(jsonb_build_object('kind', p_kind), +'org-db-changed', 'presence:org:' || p_org_id, true)`. +2. `touch_org_change_signal()` trigger fn delegates to the helper with kind + mapped from `TG_TABLE_NAME` (`cloud_projects`→projects, + `cloud_work_items`→workItems, `cloud_sessions`→sessions, + `cloud_session_comments`→comments). +3. New `AFTER INSERT OR UPDATE OR DELETE` trigger on `org_memberships` → + `nudge_org_signal(org_id, 'roster')`. +4. Recreate the 3 policy RPCs replacing their inline bump with the helper + (`kind:'policy'`). +5. `get_cloud_capabilities()` → `jsonb_build_object('broadcastSignals', +true)`, grant `authenticated`. +6. Lock-order invariant preserved: the helper is still the statement-final + signal-row acquisition (canonical slot 6); `realtime.send` inserts into + `realtime.messages` (a partitioned insert, no user-table locks) after it. +7. Verify tail: single-signature checks; trigger presence; capabilities + callable. + +Offline validation: brew-PG harness stubs `realtime.send` (records rows into +a scratch table) — assert one broadcast per debounce window per kind source, +none when suppressed, roster/policy events fire, legacy signal row byte-same. + +## Client (ORGII) + +1. `org2CloudClient` (or syncClient): `getCloudCapabilities(accessToken)` + with PGRST202→null; per-endpoint memory (`capabilitiesByEndpoint`). +2. `org2CloudRealtimeClient.joinPresence`: surface server broadcasts — + `org-db-changed` events dispatch to a new `onOrgDbChanged(kind)` option + next to the existing broadcast handlers; expose the channel's SUBSCRIBED + edge (already surfaced for presence re-track) to the caller for recovery + bookkeeping. +3. `useOrg2CloudRealtime`: when capabilities report `broadcastSignals`, + skip creating Slice B and B-roster postgres_changes channels; wire + `onOrgDbChanged`: `roster` → `bumpRosterVersion(orgId)` (+ the + rosterRealtimeConnected atom keyed to the org channel's join edge); + every other kind → `scheduleCoarseSignalRefresh()` (behavior-identical + coarse path; per-kind narrowing is a later optimization, the payload + already carries it). SUBSCRIBED-edge full/delta recovery + (`decideSubscribedEdgeRecovery`) moves onto the org channel's join edge. + Slice A unchanged. +4. Legacy backends (probe fails): topology unchanged (3 pg channels + 1 + presence). Tests: transport tests reworked in + `org2CloudRealtimeClient.test.ts`; recovery/lease/scope tests untouched. + +## Quota effect + +Broadcast messages still bill per delivery, but: no per-change × per-client +RLS evaluation, no WAL-poller serialization, and the `kind` discriminator +unlocks future per-plane refetch narrowing (today every signal shotguns all +planes after the 60s throttle). The 250ms server debounce and 60s client +throttle keep message volume identical to today's. + +## Rollout order + +0005 applies live (additive, old clients unaffected) → client PR merges → +fleet upgrades → later cleanup migration drops org_change_signals from the +publication (and eventually the table, once fleet-wide). diff --git a/docs/h4-h5-design-2026-07-24/h5-storage-offload.md b/docs/h4-h5-design-2026-07-24/h5-storage-offload.md new file mode 100644 index 000000000..60b4019f6 --- /dev/null +++ b/docs/h4-h5-design-2026-07-24/h5-storage-offload.md @@ -0,0 +1,72 @@ +# H5 — Replay Segment Payload Offload to Supabase Storage + +Date: 2026-07-24. Basis: segment-lifecycle architecture map (agent survey of +0001/0002/0003 + client push/pull paths). Companion: `h4-broadcast.md`. + +## Decision summary + +| Question | Decision | Why | +| ----------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| What moves to Storage | **Frozen segments only** (`seq >= 1`) | Immutable + append-only + the bulk of bytes. No overwrite churn, no orphan risk on the hot path. | +| What stays in Postgres | **Tail (`seq = 0`) stays `payload_gz` inline** | Mutable, rewritten every push, small (recent non-terminal events). Keeps append OCC single-transaction. | +| Object encoding | raw gzip bytes (`application/gzip`) | Drops the base64 +33% leg entirely; codec already isolates the base64 step (`segmentCodec.ts`). | +| Object key | `replay/{org_id}/{session_id}/{epoch}/{seq}-{segment_hash}.gz` | Immutable-by-name; hash in key makes retry idempotent and collision-safe inside the org/session auth domain (no cross-tenant dedup by design — see SharedSessionPerformance.md side-channel note). | +| Member reads/writes | **Direct Storage REST with user JWT + storage RLS** | The whole member auth ladder (membership → retention → metadata_only → restricted-visibility grant) is already SQL (`assert_session_readable`); a `security definer` helper called from the storage policy reuses it verbatim. | +| Share-token guest reads | **Vercel signer route** (`apps/org2-cloud-web`, has `SUPABASE_SERVICE_ROLE_KEY`) | Guest tier has no `auth` role, inexpressible in storage RLS. RPC mints a short-lived grant; the route redeems it and returns signed URLs. Rare path. | +| Atomicity | **Upload blobs first, then commit manifest RPC** | RPC verifies each object exists in `storage.objects` and reads its真实 size from object metadata — server-measured bytes, no client attestation. Missing object ⇒ `ORG2_VALIDATION`, nothing committed. | +| Quota unit | `byte_size` = raw object bytes (frozen) / `octet_length(payload_gz)` (tail, unchanged) | −25% vs base64 accounting: quotas get slightly looser, matches what is actually stored. Documented, not compensated. | +| Orphan GC | SQL view `replay_orphan_objects` (objects with no matching segment row, older than 24h) + service-role sweep via Storage API (Vercel cron) | Postgres FK cascade cannot delete bucket objects; rewrite/delete-account leave objects behind by design and the sweep reaps them. | +| Rollout | Additive: `cloud_session_segments.storage_path text null`; null = legacy inline. Read RPCs return `storagePath` XOR `payloadGz` per segment. | Old clients fail closed on new-format sessions (zod `payloadGz` missing ⇒ import error, no corruption). Pre-release population is auto-updating; accepted. | + +## Server (cloud-infra migration, offline-validatable parts) + +1. `cloud_session_segments.storage_path text` + relax `payload_gz` to nullable + with `check ((payload_gz is not null) or (storage_path is not null))` and + `check (seq = 0 → payload_gz is not null)` (tail always inline). +2. Bucket `replay` (private) + storage RLS policies delegating to + `org2_cloud.can_read_replay_object(name)` / `can_write_replay_object(name)` + security-definer helpers (path parse → session ladder / owner+entitlement + gates + epoch/seq sanity). +3. `cloud_append_session_events` / `cloud_rewrite_session_events`: frozen + segment wire accepts `{seq, storagePath, eventCount, segmentHash}` as an + alternative to `payloadGz`; on the storage form the RPC looks the object up + in `storage.objects` (existence + `(metadata->>'size')::bigint` + key must + embed the claimed `segment_hash`), charges quota with the object size, and + stores `storage_path`. Legacy inline form byte-identical behavior. +4. `cloud_get_session_events(_page)`: rows return `storagePath` when set, + `payloadGz` otherwise. Guest grant RPC `cloud_authorize_replay_read` + + redeem RPC (service-role) for the signer route. +5. `reconcile_org_stored_bytes` unchanged (byte_size stays truthful). +6. `replay_orphan_objects` view + `gc_replay_orphans` bookkeeping (row side); + object deletion happens in the cron sweeper, not SQL. + +## Client (ORGII) + +- Push seam: inside `org2CloudSyncClient.appendSessionEvents`/`rewrite…` — + before the RPC, upload each frozen segment's raw gzip bytes via + `PUT /storage/v1/object/replay/{key}` (JWT), bounded concurrency (reuse + `mapSegmentsBounded` cap 4); segment wire swaps `payloadGz` → + `storagePath`. Tail keeps the base64 inline path. +- Pull seam: `decodeCloudSegments` (`org2CloudBackendAdapter.ts`) — when a + segment carries `storagePath`, `GET /storage/v1/object/replay/{key}` (JWT) + → gunzip raw bytes; integrity check unchanged (sha256 of pre-gzip bytes vs + `segmentHash`). Share-token imports call the signer route first. +- Capability fallback mirror of the 0004 pattern: on storage-form rejection + (pre-migration backend), remember per endpoint and fall back to inline + wire. Full backward compatibility both directions. + +## Explicitly out of scope + +- Migrating EXISTING inline segments (backfill mover is a later service-role + batch job; inline reads stay supported indefinitely). +- TUS/resumable uploads (objects are ≤ ~256 KiB pre-gzip; far below 6 MB). +- Cross-org content dedup (side channel, rejected in SharedSessionPerformance). + +## Validation plan + +- Offline (brew PG + storage schema stub): migration idempotency, wire A/B + (inline form byte-identical vs 0004 DB), manifest commit vs missing object, + quota deltas with mixed inline/storage segments, orphan view correctness. +- Live: real Storage bucket + policies (user pastes migration, bucket created + via dashboard or SQL insert into storage.buckets), dual-instance push/pull, + cmd+5 + CPU/RAM per standing rule. diff --git a/src/features/Org2Cloud/org2CloudCapabilities.test.ts b/src/features/Org2Cloud/org2CloudCapabilities.test.ts new file mode 100644 index 000000000..665b6047f --- /dev/null +++ b/src/features/Org2Cloud/org2CloudCapabilities.test.ts @@ -0,0 +1,72 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { + __CAPABILITIES_INTERNALS, + getCloudCapabilities, +} from "./org2CloudCapabilities"; +import { getCloudCapabilitiesRaw } from "./org2CloudClient"; + +vi.mock("./org2CloudClient", () => ({ + getCloudCapabilitiesRaw: vi.fn(), +})); + +const rawMock = vi.mocked(getCloudCapabilitiesRaw); + +beforeEach(() => { + __CAPABILITIES_INTERNALS.reset(); +}); + +afterEach(() => { + rawMock.mockReset(); +}); + +describe("getCloudCapabilities", () => { + it("parses a 0005 payload and caches it per endpoint", async () => { + rawMock.mockResolvedValueOnce({ broadcastSignals: true }); + expect(await getCloudCapabilities("jwt-1")).toEqual({ + broadcastSignals: true, + }); + expect(await getCloudCapabilities("jwt-1")).toEqual({ + broadcastSignals: true, + }); + expect(rawMock).toHaveBeenCalledTimes(1); + }); + + it("answers legacy on failure without caching so the next probe retries", async () => { + rawMock.mockResolvedValueOnce(null); + expect(await getCloudCapabilities("jwt-1")).toEqual({ + broadcastSignals: false, + }); + rawMock.mockResolvedValueOnce({ broadcastSignals: true }); + expect(await getCloudCapabilities("jwt-1")).toEqual({ + broadcastSignals: true, + }); + expect(rawMock).toHaveBeenCalledTimes(2); + }); + + it("degrades a malformed flag to false and still caches the answer", async () => { + rawMock.mockResolvedValueOnce({ broadcastSignals: "yes" }); + expect(await getCloudCapabilities("jwt-1")).toEqual({ + broadcastSignals: false, + }); + expect(await getCloudCapabilities("jwt-1")).toEqual({ + broadcastSignals: false, + }); + expect(rawMock).toHaveBeenCalledTimes(1); + }); + + it("coalesces concurrent probes into one request", async () => { + let release: (value: unknown) => void = () => undefined; + rawMock.mockReturnValueOnce( + new Promise((resolve) => { + release = resolve; + }) + ); + const first = getCloudCapabilities("jwt-1"); + const second = getCloudCapabilities("jwt-1"); + release({ broadcastSignals: true }); + expect(await first).toEqual({ broadcastSignals: true }); + expect(await second).toEqual({ broadcastSignals: true }); + expect(rawMock).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/features/Org2Cloud/org2CloudCapabilities.ts b/src/features/Org2Cloud/org2CloudCapabilities.ts new file mode 100644 index 000000000..3c0ad0d8a --- /dev/null +++ b/src/features/Org2Cloud/org2CloudCapabilities.ts @@ -0,0 +1,62 @@ +/** + * Endpoint-level backend capability probe (0005+). `get_cloud_capabilities` + * is additive: a pre-0005 backend answers PGRST202/404 and the probe resolves + * to the legacy shape with every flag false. Cached per endpoint for the app + * session (the 0004 remember-per-endpoint pattern); a restart re-probes, so + * a live backend upgrade is picked up without client logic. + */ +import { z } from "zod/v4"; + +import { getCloudEndpoint } from "./config"; +import { getCloudCapabilitiesRaw } from "./org2CloudClient"; + +const CloudCapabilitiesWireSchema = z.object({ + broadcastSignals: z.boolean().nullish().catch(undefined), +}); + +export interface CloudCapabilities { + broadcastSignals: boolean; +} + +const LEGACY_CAPABILITIES: CloudCapabilities = { broadcastSignals: false }; + +const capabilitiesByEndpoint = new Map(); +const inFlightByEndpoint = new Map>(); + +export async function getCloudCapabilities( + accessToken: string +): Promise { + const endpointKey = getCloudEndpoint().supabaseUrl; + const cached = capabilitiesByEndpoint.get(endpointKey); + if (cached) return cached; + const inFlight = inFlightByEndpoint.get(endpointKey); + if (inFlight) return inFlight; + const probe = (async () => { + const payload = await getCloudCapabilitiesRaw(accessToken); + const parsed = CloudCapabilitiesWireSchema.safeParse(payload); + if (payload === null || !parsed.success) { + // 404 (pre-0005) and transient failures are indistinguishable here, so + // answer legacy but do NOT cache — the next connection generation + // re-probes instead of pinning a healthy backend to the legacy path. + return LEGACY_CAPABILITIES; + } + const capabilities: CloudCapabilities = { + broadcastSignals: parsed.data.broadcastSignals ?? false, + }; + capabilitiesByEndpoint.set(endpointKey, capabilities); + return capabilities; + })(); + inFlightByEndpoint.set(endpointKey, probe); + try { + return await probe; + } finally { + inFlightByEndpoint.delete(endpointKey); + } +} + +export const __CAPABILITIES_INTERNALS = { + reset: () => { + capabilitiesByEndpoint.clear(); + inFlightByEndpoint.clear(); + }, +}; diff --git a/src/features/Org2Cloud/org2CloudClient.ts b/src/features/Org2Cloud/org2CloudClient.ts index fac993882..806e3b11d 100644 --- a/src/features/Org2Cloud/org2CloudClient.ts +++ b/src/features/Org2Cloud/org2CloudClient.ts @@ -127,6 +127,16 @@ export async function schemaVersion(): Promise { return typeof payload === "number" ? payload : null; } +/** + * Raw 0005+ capability read; `null` on pre-0005 backends (PGRST202) and on + * transport failure. Interpretation/caching live in `org2CloudCapabilities`. + */ +export async function getCloudCapabilitiesRaw( + accessToken: string +): Promise { + return callRpc("get_cloud_capabilities", accessToken); +} + /** * Fetch the signed-in user's cloud profile. Returns `null` on any failure * or when the server returns an empty object (no profile row yet). diff --git a/src/features/Org2Cloud/org2CloudControlBus.test.ts b/src/features/Org2Cloud/org2CloudControlBus.test.ts index ac21fece0..bd699c342 100644 --- a/src/features/Org2Cloud/org2CloudControlBus.test.ts +++ b/src/features/Org2Cloud/org2CloudControlBus.test.ts @@ -4,6 +4,7 @@ import { ORG_CONTROL_CHANGED_EVENT, broadcastOrgControlChangedToPeers, parseOrgControlChangeKind, + parseOrgDbChangeKind, registerOrgControlBroadcaster, } from "./org2CloudControlBus"; @@ -44,4 +45,15 @@ describe("org2 cloud control bus", () => { expect(parseOrgControlChangeKind({ kind: "projects" })).toBeNull(); expect(parseOrgControlChangeKind({ kind: "scopes" })).toBe("scopes"); }); + + it("parses server db-change kinds and rejects unknown ones", () => { + expect(parseOrgDbChangeKind({ kind: "sessions" })).toBe("sessions"); + expect(parseOrgDbChangeKind({ kind: "comments" })).toBe("comments"); + expect(parseOrgDbChangeKind({ kind: "projects" })).toBe("projects"); + expect(parseOrgDbChangeKind({ kind: "workItems" })).toBe("workItems"); + expect(parseOrgDbChangeKind({ kind: "roster" })).toBe("roster"); + expect(parseOrgDbChangeKind({ kind: "policy" })).toBe("policy"); + expect(parseOrgDbChangeKind({ kind: "entitlement" })).toBeNull(); + expect(parseOrgDbChangeKind({})).toBeNull(); + }); }); diff --git a/src/features/Org2Cloud/org2CloudControlBus.ts b/src/features/Org2Cloud/org2CloudControlBus.ts index af4040902..b2d4f027f 100644 --- a/src/features/Org2Cloud/org2CloudControlBus.ts +++ b/src/features/Org2Cloud/org2CloudControlBus.ts @@ -69,3 +69,32 @@ export function parseOrgControlChangeKind( ? kind : null; } + +/** + * Server-originated change signal (0005 Broadcast-from-Database): DB triggers + * `realtime.send` this event on the same org channel. Distinct from the + * client-sent ORG_CONTROL_CHANGED_EVENT so neither side misparses the other. + */ +export const ORG_DB_CHANGED_EVENT = "org-db-changed"; + +export type Org2CloudDbChangeKind = + | "sessions" + | "comments" + | "projects" + | "workItems" + | "roster" + | "policy"; + +export function parseOrgDbChangeKind( + payload: Record +): Org2CloudDbChangeKind | null { + const kind = payload.kind; + return kind === "sessions" || + kind === "comments" || + kind === "projects" || + kind === "workItems" || + kind === "roster" || + kind === "policy" + ? kind + : null; +} diff --git a/src/features/Org2Cloud/org2CloudRealtimeClient.test.ts b/src/features/Org2Cloud/org2CloudRealtimeClient.test.ts index 6ef334a35..afa275c61 100644 --- a/src/features/Org2Cloud/org2CloudRealtimeClient.test.ts +++ b/src/features/Org2Cloud/org2CloudRealtimeClient.test.ts @@ -81,6 +81,24 @@ describe("createOrg2CloudRealtimeConnection presence privacy", () => { }); }); + it("surfaces presence-channel subscription edges through onStatus", () => { + const conn = createOrg2CloudRealtimeConnection("token-abc"); + const edges: boolean[] = []; + conn.joinPresence({ + scope: "org:org-123", + key: "user-9", + payload: null, + onSync: () => undefined, + onStatus: (subscribed) => edges.push(subscribed), + }); + const channel = createdChannels.at(-1); + channel?.emitStatus("SUBSCRIBED"); + channel?.emitStatus("CHANNEL_ERROR"); + channel?.emitStatus("SUBSCRIBED"); + channel?.emitStatus("CLOSED"); + expect(edges).toEqual([true, false, true, false]); + }); + it("authorizes the socket with the access token before joining (RLS private-channel requirement)", () => { createOrg2CloudRealtimeConnection("token-abc"); expect(setAuthMock).toHaveBeenCalledWith("token-abc"); diff --git a/src/features/Org2Cloud/org2CloudRealtimeClient.ts b/src/features/Org2Cloud/org2CloudRealtimeClient.ts index 0ecc742bf..bb693998a 100644 --- a/src/features/Org2Cloud/org2CloudRealtimeClient.ts +++ b/src/features/Org2Cloud/org2CloudRealtimeClient.ts @@ -100,6 +100,13 @@ export interface Org2CloudPresenceOptions { event: string, payload: Record ) => void; + /** + * Fired on subscription status edges, like `Org2CloudSubscribeOptions. + * onStatus`. When the backend broadcasts change signals on this channel + * (0005), callers use the true-edge for missed-signal recovery — the same + * role the dedicated signal channel's edge played on legacy backends. + */ + readonly onStatus?: (subscribed: boolean) => void; } export interface Org2CloudPresenceHandle { @@ -260,7 +267,7 @@ export function createOrg2CloudRealtimeConnection( leave: () => undefined, }; } - const { scope, key, payload, onSync, onBroadcast } = options; + const { scope, key, payload, onSync, onBroadcast, onStatus } = options; // Private: presence roster + broadcast nudges are org-scoped, gated by the // RLS policy on realtime.messages for topic `presence:` (setAuth // carries the JWT the authorization check needs). @@ -487,6 +494,7 @@ export function createOrg2CloudRealtimeConnection( subscribed = false; published = false; } + onStatus?.(status === "SUBSCRIBED"); }); channels.add(channel); return { diff --git a/src/features/Org2Cloud/useOrg2CloudRealtime.ts b/src/features/Org2Cloud/useOrg2CloudRealtime.ts index d137509a3..6a4c3523e 100644 --- a/src/features/Org2Cloud/useOrg2CloudRealtime.ts +++ b/src/features/Org2Cloud/useOrg2CloudRealtime.ts @@ -35,6 +35,7 @@ import { useLayoutEffect, useMemo, useRef, + useState, } from "react"; import { @@ -53,6 +54,7 @@ import { org2CloudAuthAtom, org2CloudAuthIdentityKey, } from "./org2CloudAuthAtom"; +import { getCloudCapabilities } from "./org2CloudCapabilities"; import { ensureFreshSession } from "./org2CloudClient"; import { COMMENTS_CHANGED_EVENT, @@ -64,7 +66,9 @@ import { } from "./org2CloudCommentsBus"; import { ORG_CONTROL_CHANGED_EVENT, + ORG_DB_CHANGED_EVENT, parseOrgControlChangeKind, + parseOrgDbChangeKind, registerOrgControlBroadcaster, } from "./org2CloudControlBus"; import { refreshOrgEntitlement } from "./org2CloudEntitlementCoordinator"; @@ -265,6 +269,27 @@ export function useOrg2CloudRealtime(): void { authRef.current = auth; }, [auth]); + // 0005 capability: change signals arrive as server broadcasts on the org + // channel instead of postgres_changes. `false` covers legacy backends AND + // the unresolved-probe window — the legacy channels stay up until the probe + // flips, so no signal is ever lost; pre-0005 backends never flip, so their + // subscription topology never churns. + const [broadcastSignals, setBroadcastSignals] = useState(false); + useEffect(() => { + let cancelled = false; + setBroadcastSignals(false); + const current = authRef.current; + if (!userId || !current) return undefined; + void getCloudCapabilities(current.accessToken).then((capabilities) => { + if (!cancelled && capabilities.broadcastSignals) { + setBroadcastSignals(true); + } + }); + return () => { + cancelled = true; + }; + }, [userId, endpointUrl]); + // `org_change_signals` also carries rare sharing-floor changes. Refresh only // the affected org's entitlement through the shared coordinator // (store-keyed single-flight + TTL) instead of using list_my_orgs as a @@ -345,124 +370,152 @@ export function useOrg2CloudRealtime(): void { } }, [auth?.accessToken]); + // Shared by the Slice B postgres_changes path (legacy backends) and the + // Slice C org-channel broadcast path (0005 backends): identical throttle + // windows and refresh behavior regardless of transport. + const runCoarseSignalRefresh = useCallback(() => { + const orgId = activeRealtimeOrgId; + if (!orgId) return; + const now = Date.now(); + coarseSignalHandledAtRef.current = now; + // A blur/visibility event releases the connection. Ignore the tiny + // event-delivery race during teardown; the next SUBSCRIBED true-edge + // performs the authoritative full recovery. + if (isDocumentHidden()) return; + org2CloudSyncEngine.invalidateOrgInbound(orgId); + bumpRemoteSessionsVersion(orgId); + bumpOrgCommentsSignal(orgId); + if ( + now - controlPlaneRefreshAtRef.current >= + CONTROL_PLANE_REFRESH_THROTTLE_MS + ) { + controlPlaneRefreshAtRef.current = now; + void refetchRef.current(); + void refreshEntitlementForOrg(orgId).then((floorChanged) => { + if (floorChanged) org2CloudSyncEngine.resumeOrg(orgId); + }); + } + }, [ + activeRealtimeOrgId, + bumpRemoteSessionsVersion, + bumpOrgCommentsSignal, + refreshEntitlementForOrg, + ]); + const scheduleCoarseSignalRefresh = useCallback(() => { + const now = Date.now(); + const elapsed = now - coarseSignalHandledAtRef.current; + if (elapsed >= COARSE_SIGNAL_THROTTLE_MS) { + runCoarseSignalRefresh(); + return; + } + if (coarseSignalTrailingTimerRef.current) return; + coarseSignalTrailingTimerRef.current = setTimeout(() => { + coarseSignalTrailingTimerRef.current = null; + runCoarseSignalRefresh(); + }, COARSE_SIGNAL_THROTTLE_MS - elapsed); + }, [runCoarseSignalRefresh]); + + // Recovery for missed signals on a (re)subscribed org scope. Legacy: the + // signal channel's SUBSCRIBED edge. 0005: the org broadcast channel's edge. + const runSignalEdgeRecovery = useCallback( + (orgId: string) => { + coarseSignalHandledAtRef.current = Date.now(); + controlPlaneRefreshAtRef.current = Date.now(); + if (isDocumentHidden()) return; + // A LONG gap forces complete listings so tombstone-free absences + // (revoked projects / retention shifts) are observed; a short gap or + // a rejoin storm keeps the delta cursors, which already merge + // deletedAt/LWW tombstones. + const decision = decideSubscribedEdgeRecovery({ + nowMs: Date.now(), + teardownAtMs: orgTeardownAtRef.current.get(orgId), + lastFullRecoveryAtMs: orgFullRecoveryAtRef.current.get(orgId), + }); + if (decision === "delta") { + org2CloudSyncEngine.invalidateOrgInbound(orgId); + bumpRemoteSessionsVersion(orgId); + bumpOrgCommentsSignal(orgId); + return; + } + orgFullRecoveryAtRef.current.set(orgId, Date.now()); + org2CloudSyncEngine.invalidateOrgInbound(orgId, { + full: true, + pushSessions: true, + }); + void refreshEntitlementForOrg(orgId).then((floorChanged) => { + if (floorChanged) org2CloudSyncEngine.resumeOrg(orgId); + }); + // Force a complete listing after a disconnected window while the + // last authorized snapshot stays visible. The fetch coordinator + // replaces it atomically only if the server truth changed. + bumpRemoteSessionsVersion(orgId, { full: true }); + bumpOrgCommentsSignal(orgId); + // The org-level bump above is TTL-gated; the session broadcast the + // released socket missed is exactly what the open thread needs, so + // force the active session's thread past the TTL. + bumpActiveSessionCommentsSignal(orgId); + }, + [ + bumpRemoteSessionsVersion, + bumpOrgCommentsSignal, + bumpActiveSessionCommentsSignal, + refreshEntitlementForOrg, + ] + ); + // --- Slice B: change-signal + roster subscriptions for the active org only. // Inactive orgs catch up immediately when selected; they must not keep // reconnecting channels or pulling data while the user works elsewhere. + // On 0005 backends both planes ride the org broadcast channel (Slice C) + // instead and this effect keeps only the teardown bookkeeping. useEffect(() => { const connection = connectionRef.current; if (!connection || !userId || !activeRealtimeOrgId) return undefined; const unsubscribes: Array<() => void> = []; const orgId = activeRealtimeOrgId; - const runCoarseSignalRefresh = () => { - const now = Date.now(); - coarseSignalHandledAtRef.current = now; - // A blur/visibility event releases the connection. Ignore the tiny - // event-delivery race during teardown; the next SUBSCRIBED true-edge - // below performs the authoritative full recovery. - if (isDocumentHidden()) return; - org2CloudSyncEngine.invalidateOrgInbound(orgId); - bumpRemoteSessionsVersion(orgId); - bumpOrgCommentsSignal(orgId); - if ( - now - controlPlaneRefreshAtRef.current >= - CONTROL_PLANE_REFRESH_THROTTLE_MS - ) { - controlPlaneRefreshAtRef.current = now; - void refetchRef.current(); - void refreshEntitlementForOrg(orgId).then((floorChanged) => { - if (floorChanged) org2CloudSyncEngine.resumeOrg(orgId); - }); - } - }; - const scheduleCoarseSignalRefresh = () => { - const now = Date.now(); - const elapsed = now - coarseSignalHandledAtRef.current; - if (elapsed >= COARSE_SIGNAL_THROTTLE_MS) { - runCoarseSignalRefresh(); - return; - } - if (coarseSignalTrailingTimerRef.current) return; - coarseSignalTrailingTimerRef.current = setTimeout(() => { - coarseSignalTrailingTimerRef.current = null; - runCoarseSignalRefresh(); - }, COARSE_SIGNAL_THROTTLE_MS - elapsed); - }; - unsubscribes.push( - connection.subscribe({ - table: CHANGE_SIGNALS_TABLE, - filter: `org_id=eq.${orgId}`, - onChange: () => { - // This row carries no plane/entity discriminator. Treat it only as - // a bounded durable event; plane-specific Presence broadcasts below - // keep normal session/comment/control changes immediate. - scheduleCoarseSignalRefresh(); - }, - onStatus: (subscribed) => { - // On (re)subscribe compensate for events missed while disconnected. - // A LONG gap forces complete listings so tombstone-free absences - // (revoked projects / retention shifts) are observed; a short gap or - // a rejoin storm keeps the delta cursors, which already merge - // deletedAt/LWW tombstones. - if (subscribed) { - coarseSignalHandledAtRef.current = Date.now(); - controlPlaneRefreshAtRef.current = Date.now(); - if (isDocumentHidden()) return; - const decision = decideSubscribedEdgeRecovery({ - nowMs: Date.now(), - teardownAtMs: orgTeardownAtRef.current.get(orgId), - lastFullRecoveryAtMs: orgFullRecoveryAtRef.current.get(orgId), - }); - if (decision === "delta") { - org2CloudSyncEngine.invalidateOrgInbound(orgId); - bumpRemoteSessionsVersion(orgId); - bumpOrgCommentsSignal(orgId); - return; - } - orgFullRecoveryAtRef.current.set(orgId, Date.now()); - org2CloudSyncEngine.invalidateOrgInbound(orgId, { - full: true, - pushSessions: true, - }); - void refreshEntitlementForOrg(orgId).then((floorChanged) => { - if (floorChanged) org2CloudSyncEngine.resumeOrg(orgId); - }); - // Force a complete listing after a disconnected window while the - // last authorized snapshot stays visible. The fetch coordinator - // replaces it atomically only if the server truth changed. - bumpRemoteSessionsVersion(orgId, { full: true }); - bumpOrgCommentsSignal(orgId); - // The org-level bump above is TTL-gated; the session broadcast the - // released socket missed is exactly what the open thread needs, so - // force the active session's thread past the TTL. - bumpActiveSessionCommentsSignal(orgId); - } - }, - }) - ); - // Org-wide roster: a TEAMMATE joining/leaving/changing role (Slice A - // only carries the signed-in user's OWN rows). Bumps the per-org - // version counter; CloudOrgPanelView keys its fetch on it so the - // members list updates live while the panel is open. - unsubscribes.push( - connection.subscribe({ - table: "org_memberships", - filter: `org_id=eq.${orgId}`, - onChange: () => { - bumpRosterVersion(orgId); - }, - onStatus: (subscribed) => { - setRosterRealtimeConnected((current) => - current[orgId] === subscribed - ? current - : { ...current, [orgId]: subscribed } - ); - // Compensate for roster events missed while disconnected. - if (subscribed) bumpRosterVersion(orgId); - }, - }) - ); - log.info(`realtime: subscribed inbound planes for active org ${orgId}`); + if (!broadcastSignals) { + unsubscribes.push( + connection.subscribe({ + table: CHANGE_SIGNALS_TABLE, + filter: `org_id=eq.${orgId}`, + onChange: () => { + // This row carries no plane/entity discriminator. Treat it only as + // a bounded durable event; plane-specific Presence broadcasts + // keep normal session/comment/control changes immediate. + scheduleCoarseSignalRefresh(); + }, + onStatus: (subscribed) => { + // On (re)subscribe compensate for events missed while + // disconnected. + if (subscribed) runSignalEdgeRecovery(orgId); + }, + }) + ); + // Org-wide roster: a TEAMMATE joining/leaving/changing role (Slice A + // only carries the signed-in user's OWN rows). Bumps the per-org + // version counter; CloudOrgPanelView keys its fetch on it so the + // members list updates live while the panel is open. + unsubscribes.push( + connection.subscribe({ + table: "org_memberships", + filter: `org_id=eq.${orgId}`, + onChange: () => { + bumpRosterVersion(orgId); + }, + onStatus: (subscribed) => { + setRosterRealtimeConnected((current) => + current[orgId] === subscribed + ? current + : { ...current, [orgId]: subscribed } + ); + // Compensate for roster events missed while disconnected. + if (subscribed) bumpRosterVersion(orgId); + }, + }) + ); + log.info(`realtime: subscribed inbound planes for active org ${orgId}`); + } return () => { for (const unsub of unsubscribes) unsub(); @@ -484,10 +537,10 @@ export function useOrg2CloudRealtime(): void { activeRealtimeOrgId, userId, endpointUrl, + broadcastSignals, bumpRosterVersion, - bumpRemoteSessionsVersion, - bumpOrgCommentsSignal, - refreshEntitlementForOrg, + scheduleCoarseSignalRefresh, + runSignalEdgeRecovery, ]); // --- Slice C: org-level presence for the actively-used org only. @@ -560,6 +613,22 @@ export function useOrg2CloudRealtime(): void { key: userId, payload: initialPayload, onBroadcast: (event, payload) => { + if (event === ORG_DB_CHANGED_EVENT) { + // Server-originated signal (0005 Broadcast-from-Database). Same + // downstream behavior as the legacy postgres_changes transports: + // roster events bump the per-org version, everything else runs + // the throttled coarse refresh (the kind discriminator is carried + // for future per-plane narrowing). + if (!broadcastSignals) return; + const kind = parseOrgDbChangeKind(payload); + if (!kind) return; + if (kind === "roster") { + bumpRosterVersion(orgId); + return; + } + scheduleCoarseSignalRefresh(); + return; + } if (event === ORG_CONTROL_CHANGED_EVENT) { const kind = parseOrgControlChangeKind(payload); if (kind && isDocumentHidden()) return; @@ -617,6 +686,20 @@ export function useOrg2CloudRealtime(): void { : { ...current, [orgId]: byUser } ); }, + onStatus: (subscribed) => { + // On 0005 backends this channel carries the change signals, so its + // edges own the roster-connected indicator and the missed-signal + // recovery that the dedicated legacy channels' edges owned. + if (!broadcastSignals) return; + setRosterRealtimeConnected((current) => + current[orgId] === subscribed + ? current + : { ...current, [orgId]: subscribed } + ); + if (!subscribed) return; + bumpRosterVersion(orgId); + runSignalEdgeRecovery(orgId); + }, }); handles.set(orgId, handle); payloadKeys.set(orgId, org2CloudPresencePayloadKey(initialPayload)); @@ -661,11 +744,15 @@ export function useOrg2CloudRealtime(): void { activeRealtimeOrgId, userId, endpointUrl, + broadcastSignals, buildPayload, setPresence, setOutboundPresence, setCommentsSignal, bumpRemoteSessionsVersion, + bumpRosterVersion, + scheduleCoarseSignalRefresh, + runSignalEdgeRecovery, ]); // Keep awareness attached to the session while this foreground lease owns From bdea98b07f856335f7aa1b848d354a2cff28b045 Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Thu, 23 Jul 2026 23:01:34 -0700 Subject: [PATCH 2/4] fix(cloud): roster broadcasts also run coarse recovery Pre-commit hook ran. Total eslint: 1, total circular: 0 --- src/features/Org2Cloud/useOrg2CloudRealtime.ts | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/src/features/Org2Cloud/useOrg2CloudRealtime.ts b/src/features/Org2Cloud/useOrg2CloudRealtime.ts index 6a4c3523e..d670b2431 100644 --- a/src/features/Org2Cloud/useOrg2CloudRealtime.ts +++ b/src/features/Org2Cloud/useOrg2CloudRealtime.ts @@ -615,17 +615,15 @@ export function useOrg2CloudRealtime(): void { onBroadcast: (event, payload) => { if (event === ORG_DB_CHANGED_EVENT) { // Server-originated signal (0005 Broadcast-from-Database). Same - // downstream behavior as the legacy postgres_changes transports: - // roster events bump the per-org version, everything else runs - // the throttled coarse refresh (the kind discriminator is carried - // for future per-plane narrowing). + // downstream behavior as the legacy postgres_changes transports. + // `roster` also runs the coarse refresh: the membership trigger's + // bump wins the per-org debounce window, so a member-floor RPC in + // the same transaction broadcasts as `roster` (0005 PART 4 "kind + // shadowing") and the policy recovery has to ride this kind too. if (!broadcastSignals) return; const kind = parseOrgDbChangeKind(payload); if (!kind) return; - if (kind === "roster") { - bumpRosterVersion(orgId); - return; - } + if (kind === "roster") bumpRosterVersion(orgId); scheduleCoarseSignalRefresh(); return; } From e97204162008813b74dceec9ef8e7bffccea44bb Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Thu, 23 Jul 2026 23:19:49 -0700 Subject: [PATCH 3/4] feat(cloud): storage-backed frozen segment transport Pre-commit hook ran. Total eslint: 1, total circular: 0 --- .../Org2Cloud/org2CloudBackendAdapter.test.ts | 109 +++++++++ .../Org2Cloud/org2CloudBackendAdapter.ts | 51 +++- .../Org2Cloud/org2CloudCapabilities.test.ts | 34 ++- .../Org2Cloud/org2CloudCapabilities.ts | 8 +- .../Org2Cloud/org2CloudStorageClient.test.ts | 129 +++++++++++ .../Org2Cloud/org2CloudStorageClient.ts | 98 ++++++++ .../Org2Cloud/org2CloudSyncClient.test.ts | 218 +++++++++++++++++- src/features/Org2Cloud/org2CloudSyncClient.ts | 192 +++++++++++++-- .../TeamCollaboration/sync/collabGzip.ts | 23 +- .../sync/segmentCodec.test.ts | 14 ++ .../TeamCollaboration/sync/segmentCodec.ts | 30 +++ 11 files changed, 862 insertions(+), 44 deletions(-) create mode 100644 src/features/Org2Cloud/org2CloudStorageClient.test.ts create mode 100644 src/features/Org2Cloud/org2CloudStorageClient.ts diff --git a/src/features/Org2Cloud/org2CloudBackendAdapter.test.ts b/src/features/Org2Cloud/org2CloudBackendAdapter.test.ts index 7b9cf2db9..d8d52c400 100644 --- a/src/features/Org2Cloud/org2CloudBackendAdapter.test.ts +++ b/src/features/Org2Cloud/org2CloudBackendAdapter.test.ts @@ -2,7 +2,9 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import type { SessionEvent } from "@src/engines/SessionCore/core/types"; +import { computeSegmentHash } from "../TeamCollaboration/sync/collabGzip"; import { + toFrozenSegmentStorage, toFrozenSegmentWire, toTailWire, } from "../TeamCollaboration/sync/segmentCodec"; @@ -10,6 +12,7 @@ import { buildCloudSessionFetchClient, cloudSessionIdFromRowId, } from "./org2CloudBackendAdapter"; +import { downloadReplayObject } from "./org2CloudStorageClient"; import type { CloudSessionEventsSnapshot } from "./org2CloudSyncClient"; import { Org2CloudSyncError, isOrg2SyncErrorCode } from "./org2CloudSyncClient"; @@ -18,8 +21,14 @@ vi.mock("./org2CloudSyncClient", async (importOriginal) => { return { ...actual, getSessionEvents: vi.fn() }; }); +vi.mock("./org2CloudStorageClient", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, downloadReplayObject: vi.fn() }; +}); + const { getSessionEvents } = await import("./org2CloudSyncClient"); const getSessionEventsMock = vi.mocked(getSessionEvents); +const downloadReplayObjectMock = vi.mocked(downloadReplayObject); function makeEvent(id: string): SessionEvent { return { @@ -53,6 +62,7 @@ async function cloudSnapshot(): Promise { describe("buildCloudSessionFetchClient", () => { beforeEach(() => { getSessionEventsMock.mockReset(); + downloadReplayObjectMock.mockReset(); }); it("maps the cloud response into the self-hosted snapshot shape", async () => { @@ -88,6 +98,105 @@ describe("buildCloudSessionFetchClient", () => { expect(tailSeg.events).toEqual(tail); }); + it("downloads storagePath segments and decodes their raw gzip bytes (mixed page)", async () => { + const stored = await toFrozenSegmentStorage({ seq: 2, events: frozen2 }); + const storagePath = `org-1/agentsession-abc/2/2-${stored.segmentHash}.gz`; + const tailWire = await toTailWire(tail); + getSessionEventsMock.mockResolvedValue({ + epoch: 2, + frozenSeq: 2, + tailHash: tailWire!.segmentHash, + count: 5, + segments: [ + await toFrozenSegmentWire({ seq: 1, events: frozen1 }), + { + seq: 2, + storagePath, + eventCount: 1, + segmentHash: stored.segmentHash, + }, + { ...tailWire!, seq: 0 }, + ], + }); + downloadReplayObjectMock.mockResolvedValue(stored.bytes); + const client = buildCloudSessionFetchClient("jwt-token"); + + const snapshot = await client.getSessionEventSegments({ + orgId: "org-1", + sessionRowId: "org-1:user-1:agentsession-abc", + }); + + expect(downloadReplayObjectMock).toHaveBeenCalledTimes(1); + expect(downloadReplayObjectMock).toHaveBeenCalledWith( + "jwt-token", + storagePath, + undefined, + undefined + ); + const [seg1, seg2, tailSeg] = snapshot.segments; + expect(seg1.events).toEqual(frozen1); + expect(seg2).toMatchObject({ seq: 2, isTail: false, eventCount: 1 }); + expect(seg2.events).toEqual(frozen2); + // Integrity contract downstream: the hash is over pre-gzip bytes, so the + // downloaded segment must still satisfy validateSegmentIntegrity. + expect(await computeSegmentHash(seg2.events)).toBe(seg2.segmentHash); + expect(tailSeg.events).toEqual(tail); + }); + + it("threads the pinned endpoint into storage downloads", async () => { + const stored = await toFrozenSegmentStorage({ seq: 1, events: frozen1 }); + const storagePath = `org-1/agentsession-abc/1/1-${stored.segmentHash}.gz`; + getSessionEventsMock.mockResolvedValue({ + epoch: 1, + frozenSeq: 1, + tailHash: null, + count: 2, + segments: [ + { seq: 1, storagePath, eventCount: 2, segmentHash: stored.segmentHash }, + ], + }); + downloadReplayObjectMock.mockResolvedValue(stored.bytes); + const endpoint = { + webOrigin: "https://app.custom.example.com", + supabaseUrl: "https://db.custom.example.com", + anonKey: "custom-anon", + isOfficial: false, + }; + const client = buildCloudSessionFetchClient("jwt-non-member", endpoint); + + await client.getSessionEventSegments({ + orgId: "org-1", + sessionRowId: "org-1:user-1:agentsession-abc", + shareToken: "t".repeat(64), + }); + + expect(downloadReplayObjectMock).toHaveBeenCalledWith( + "jwt-non-member", + storagePath, + endpoint, + undefined + ); + }); + + it("fails closed on a segment carrying neither payloadGz nor storagePath", async () => { + getSessionEventsMock.mockResolvedValue({ + epoch: 1, + frozenSeq: 1, + tailHash: null, + count: 1, + segments: [{ seq: 1, eventCount: 1, segmentHash: "h1" }], + }); + const client = buildCloudSessionFetchClient("jwt-token"); + + await expect( + client.getSessionEventSegments({ + orgId: "org-1", + sessionRowId: "org-1:user-1:agentsession-abc", + }) + ).rejects.toThrow(/neither payloadGz nor storagePath/); + expect(downloadReplayObjectMock).not.toHaveBeenCalled(); + }); + it("passes afterSeq to the server range read and drops any smuggled prefix", async () => { getSessionEventsMock.mockResolvedValue(await cloudSnapshot()); const client = buildCloudSessionFetchClient("jwt-token"); diff --git a/src/features/Org2Cloud/org2CloudBackendAdapter.ts b/src/features/Org2Cloud/org2CloudBackendAdapter.ts index f2ffc6c66..d058aeea0 100644 --- a/src/features/Org2Cloud/org2CloudBackendAdapter.ts +++ b/src/features/Org2Cloud/org2CloudBackendAdapter.ts @@ -10,7 +10,8 @@ * - cloud `seq = 0` is the mutable tail row → canonical `isTail: true`; * - cloud `payloadGz` (gzipped base64 event array) → decoded `events` via * the SHARED `decodeSegmentEvents` codec (byte-identical wire format — - * both backends push through `segmentCodec`); + * both backends push through `segmentCodec`); a 0006 `storagePath` + * segment downloads its raw gzip object from the replay bucket instead; * - cloud `{epoch, frozenSeq, tailHash, count}` map 1:1 to the snapshot * summary fields (`cloud_get_session_events` was built as a mirror of * `orgii_get_session_event_segments`); @@ -24,6 +25,8 @@ * server-side retention filter) propagates to the caller so the panel can * show an upgrade prompt instead of a generic failure. */ +import type { SessionEvent } from "@src/engines/SessionCore/core/types"; + import type { CollabSyncBackendClient, GetSessionEventSegmentsInput, @@ -31,12 +34,17 @@ import type { SessionEventSegmentsSnapshot, } from "../TeamCollaboration/sync/CollabSyncBackend"; import { - type SegmentWirePayload, decodeSegmentEvents, + decodeSegmentEventsFromBytes, mapSegmentsBounded, } from "../TeamCollaboration/sync/segmentCodec"; import type { CloudEndpoint } from "./config"; -import { getSessionEvents, streamSessionEvents } from "./org2CloudSyncClient"; +import { downloadReplayObject } from "./org2CloudStorageClient"; +import { + type CloudSegmentWire, + getSessionEvents, + streamSessionEvents, +} from "./org2CloudSyncClient"; /** The one capability replay/fork need from a backend. */ export type CloudSessionFetchClient = Pick< @@ -44,9 +52,33 @@ export type CloudSessionFetchClient = Pick< "getSessionEventSegments" | "streamSessionEventSegments" >; +async function decodeCloudSegmentEvents( + segment: CloudSegmentWire, + accessToken: string, + endpoint?: CloudEndpoint, + signal?: AbortSignal +): Promise { + if (segment.payloadGz != null) { + return decodeSegmentEvents(segment.payloadGz); + } + if (segment.storagePath != null) { + return decodeSegmentEventsFromBytes( + await downloadReplayObject( + accessToken, + segment.storagePath, + endpoint, + signal + ) + ); + } + throw new Error("cloud segment carries neither payloadGz nor storagePath"); +} + async function decodeCloudSegments( - segments: readonly SegmentWirePayload[], + segments: readonly CloudSegmentWire[], afterSeq: number, + accessToken: string, + endpoint?: CloudEndpoint, signal?: AbortSignal ): Promise { return mapSegmentsBounded( @@ -59,7 +91,12 @@ async function decodeCloudSegments( return { seq, isTail: seq === 0, - events: await decodeSegmentEvents(segment.payloadGz), + events: await decodeCloudSegmentEvents( + segment, + accessToken, + endpoint, + signal + ), eventCount: segment.eventCount, segmentHash: segment.segmentHash, }; @@ -123,6 +160,8 @@ export function buildCloudSessionFetchClient( const segments = await decodeCloudSegments( snapshot.segments, afterSeq, + accessToken, + endpoint, input.signal ); return { @@ -148,6 +187,8 @@ export function buildCloudSessionFetchClient( segments: await decodeCloudSegments( page.segments, afterSeq, + accessToken, + endpoint, input.signal ), }); diff --git a/src/features/Org2Cloud/org2CloudCapabilities.test.ts b/src/features/Org2Cloud/org2CloudCapabilities.test.ts index 665b6047f..b19f0c09b 100644 --- a/src/features/Org2Cloud/org2CloudCapabilities.test.ts +++ b/src/features/Org2Cloud/org2CloudCapabilities.test.ts @@ -25,32 +25,52 @@ describe("getCloudCapabilities", () => { rawMock.mockResolvedValueOnce({ broadcastSignals: true }); expect(await getCloudCapabilities("jwt-1")).toEqual({ broadcastSignals: true, + storageSegments: false, }); expect(await getCloudCapabilities("jwt-1")).toEqual({ broadcastSignals: true, + storageSegments: false, }); expect(rawMock).toHaveBeenCalledTimes(1); }); + it("parses the 0006 storageSegments flag", async () => { + rawMock.mockResolvedValueOnce({ + broadcastSignals: true, + storageSegments: true, + }); + expect(await getCloudCapabilities("jwt-1")).toEqual({ + broadcastSignals: true, + storageSegments: true, + }); + }); + it("answers legacy on failure without caching so the next probe retries", async () => { rawMock.mockResolvedValueOnce(null); expect(await getCloudCapabilities("jwt-1")).toEqual({ broadcastSignals: false, + storageSegments: false, }); rawMock.mockResolvedValueOnce({ broadcastSignals: true }); expect(await getCloudCapabilities("jwt-1")).toEqual({ broadcastSignals: true, + storageSegments: false, }); expect(rawMock).toHaveBeenCalledTimes(2); }); it("degrades a malformed flag to false and still caches the answer", async () => { - rawMock.mockResolvedValueOnce({ broadcastSignals: "yes" }); + rawMock.mockResolvedValueOnce({ + broadcastSignals: "yes", + storageSegments: "yes", + }); expect(await getCloudCapabilities("jwt-1")).toEqual({ broadcastSignals: false, + storageSegments: false, }); expect(await getCloudCapabilities("jwt-1")).toEqual({ broadcastSignals: false, + storageSegments: false, }); expect(rawMock).toHaveBeenCalledTimes(1); }); @@ -64,9 +84,15 @@ describe("getCloudCapabilities", () => { ); const first = getCloudCapabilities("jwt-1"); const second = getCloudCapabilities("jwt-1"); - release({ broadcastSignals: true }); - expect(await first).toEqual({ broadcastSignals: true }); - expect(await second).toEqual({ broadcastSignals: true }); + release({ broadcastSignals: true, storageSegments: true }); + expect(await first).toEqual({ + broadcastSignals: true, + storageSegments: true, + }); + expect(await second).toEqual({ + broadcastSignals: true, + storageSegments: true, + }); expect(rawMock).toHaveBeenCalledTimes(1); }); }); diff --git a/src/features/Org2Cloud/org2CloudCapabilities.ts b/src/features/Org2Cloud/org2CloudCapabilities.ts index 3c0ad0d8a..8ddf351d8 100644 --- a/src/features/Org2Cloud/org2CloudCapabilities.ts +++ b/src/features/Org2Cloud/org2CloudCapabilities.ts @@ -12,13 +12,18 @@ import { getCloudCapabilitiesRaw } from "./org2CloudClient"; const CloudCapabilitiesWireSchema = z.object({ broadcastSignals: z.boolean().nullish().catch(undefined), + storageSegments: z.boolean().nullish().catch(undefined), }); export interface CloudCapabilities { broadcastSignals: boolean; + storageSegments: boolean; } -const LEGACY_CAPABILITIES: CloudCapabilities = { broadcastSignals: false }; +const LEGACY_CAPABILITIES: CloudCapabilities = { + broadcastSignals: false, + storageSegments: false, +}; const capabilitiesByEndpoint = new Map(); const inFlightByEndpoint = new Map>(); @@ -42,6 +47,7 @@ export async function getCloudCapabilities( } const capabilities: CloudCapabilities = { broadcastSignals: parsed.data.broadcastSignals ?? false, + storageSegments: parsed.data.storageSegments ?? false, }; capabilitiesByEndpoint.set(endpointKey, capabilities); return capabilities; diff --git a/src/features/Org2Cloud/org2CloudStorageClient.test.ts b/src/features/Org2Cloud/org2CloudStorageClient.test.ts new file mode 100644 index 000000000..efe714095 --- /dev/null +++ b/src/features/Org2Cloud/org2CloudStorageClient.test.ts @@ -0,0 +1,129 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { + ORG2_CLOUD_OFFICIAL_ANON_KEY, + ORG2_CLOUD_OFFICIAL_SUPABASE_URL, +} from "./config"; +import { + Org2CloudStorageError, + buildReplayObjectPath, + downloadReplayObject, + uploadReplayObject, +} from "./org2CloudStorageClient"; + +const fetchMock = vi.fn(); + +function lastCall(): { url: string; init: RequestInit } { + const [url, init] = fetchMock.mock.calls.at(-1) as [string, RequestInit]; + return { url, init }; +} + +beforeEach(() => { + vi.stubGlobal("fetch", fetchMock); + fetchMock.mockResolvedValue(new Response(null, { status: 200 })); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + fetchMock.mockReset(); +}); + +describe("buildReplayObjectPath", () => { + it("builds the bucket-relative object key", () => { + expect(buildReplayObjectPath("org-1", "s-1", 3, 7, "abc123")).toBe( + "org-1/s-1/3/7-abc123.gz" + ); + }); +}); + +describe("uploadReplayObject", () => { + it("PUTs raw gzip bytes with JWT + apikey + upsert headers", async () => { + const bytes = new Uint8Array([31, 139, 8, 0]); + await uploadReplayObject("jwt-1", "org-1/s-1/3/7-abc.gz", bytes); + const { url, init } = lastCall(); + expect(url).toBe( + `${ORG2_CLOUD_OFFICIAL_SUPABASE_URL}/storage/v1/object/replay/org-1/s-1/3/7-abc.gz` + ); + expect(init.method).toBe("PUT"); + const headers = init.headers as Record; + expect(headers.apikey).toBe(ORG2_CLOUD_OFFICIAL_ANON_KEY); + expect(headers.authorization).toBe("Bearer jwt-1"); + expect(headers["content-type"]).toBe("application/gzip"); + expect(headers["x-upsert"]).toBe("true"); + expect(new Uint8Array(init.body as Uint8Array)).toEqual(bytes); + }); + + it("uses an explicit endpoint and percent-encodes path segments", async () => { + await uploadReplayObject( + "jwt-1", + "org-1/agentsession-a:b/1/1-hash.gz", + new Uint8Array([1]), + { + webOrigin: "https://app.custom.example.com", + supabaseUrl: "https://db.custom.example.com", + anonKey: "custom-anon", + isOfficial: false, + } + ); + const { url, init } = lastCall(); + expect(url).toBe( + "https://db.custom.example.com/storage/v1/object/replay/org-1/agentsession-a%3Ab/1/1-hash.gz" + ); + expect((init.headers as Record).apikey).toBe("custom-anon"); + }); + + it("throws a coded storage error on a non-ok response", async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 403 })); + const error = await uploadReplayObject( + "jwt-1", + "org-1/s-1/1/1-h.gz", + new Uint8Array([1]) + ).catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(Org2CloudStorageError); + expect((error as Org2CloudStorageError).status).toBe(403); + }); +}); + +describe("downloadReplayObject", () => { + it("GETs the object and returns its raw bytes", async () => { + const bytes = new Uint8Array([31, 139, 8, 0, 42]); + fetchMock.mockResolvedValueOnce( + new Response(new Uint8Array(bytes), { status: 200 }) + ); + const result = await downloadReplayObject("jwt-1", "org-1/s-1/3/7-abc.gz"); + const { url, init } = lastCall(); + expect(url).toBe( + `${ORG2_CLOUD_OFFICIAL_SUPABASE_URL}/storage/v1/object/replay/org-1/s-1/3/7-abc.gz` + ); + expect(init.method).toBe("GET"); + const headers = init.headers as Record; + expect(headers.apikey).toBe(ORG2_CLOUD_OFFICIAL_ANON_KEY); + expect(headers.authorization).toBe("Bearer jwt-1"); + expect(result).toEqual(bytes); + }); + + it("throws a coded storage error on a missing object", async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 404 })); + const error = await downloadReplayObject( + "jwt-1", + "org-1/s-1/1/1-h.gz" + ).catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(Org2CloudStorageError); + expect((error as Org2CloudStorageError).status).toBe(404); + }); + + it("passes request cancellation through to the transport", async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 200 })); + const controller = new AbortController(); + await downloadReplayObject( + "jwt-1", + "org-1/s-1/1/1-h.gz", + undefined, + controller.signal + ); + expect(fetchMock).toHaveBeenCalledWith( + expect.any(String), + expect.objectContaining({ signal: controller.signal }) + ); + }); +}); diff --git a/src/features/Org2Cloud/org2CloudStorageClient.ts b/src/features/Org2Cloud/org2CloudStorageClient.ts new file mode 100644 index 000000000..b40e2ea47 --- /dev/null +++ b/src/features/Org2Cloud/org2CloudStorageClient.ts @@ -0,0 +1,98 @@ +/** + * Supabase Storage REST helpers for the replay-segment offload (H5). + * + * Frozen segments upload/download as raw gzip objects in the private + * `replay` bucket — no base64 leg, no supabase-js. Access control is + * server-side storage RLS delegating to the session read/write ladder; + * the anon key plus the user JWT are the only credentials. Object paths + * are bucket-relative (`{orgId}/{sessionId}/{epoch}/{seq}-{hash}.gz`), + * immutable-by-name, and idempotent to re-upload (`x-upsert`). + */ +import { type CloudEndpoint, getCloudEndpoint } from "./config"; +import { fetchWithTransportRetry } from "./org2CloudFetchRetry"; + +const REPLAY_BUCKET = "replay"; + +/** Storage request failure carrying the HTTP status when one was received. */ +export class Org2CloudStorageError extends Error { + readonly status: number | null; + + constructor(message: string, status: number | null = null) { + super(message); + this.name = "Org2CloudStorageError"; + this.status = status; + } +} + +/** Bucket-relative object key of one frozen segment (no bucket prefix). */ +export function buildReplayObjectPath( + orgId: string, + sessionId: string, + epoch: number, + seq: number, + segmentHash: string +): string { + return `${orgId}/${sessionId}/${epoch}/${seq}-${segmentHash}.gz`; +} + +function objectUrl(path: string, endpoint: CloudEndpoint): string { + const encodedPath = path.split("/").map(encodeURIComponent).join("/"); + return `${endpoint.supabaseUrl}/storage/v1/object/${REPLAY_BUCKET}/${encodedPath}`; +} + +function objectHeaders( + accessToken: string, + endpoint: CloudEndpoint +): Record { + return { + apikey: endpoint.anonKey, + authorization: `Bearer ${accessToken}`, + }; +} + +export async function uploadReplayObject( + accessToken: string, + path: string, + bytes: Uint8Array, + endpoint: CloudEndpoint = getCloudEndpoint(), + signal?: AbortSignal +): Promise { + const response = await fetchWithTransportRetry(objectUrl(path, endpoint), { + method: "PUT", + headers: { + ...objectHeaders(accessToken, endpoint), + "content-type": "application/gzip", + "x-upsert": "true", + }, + // Copy into a fresh ArrayBuffer-backed view: the DOM typings reject + // Uint8Array as a BodyInit. + body: new Uint8Array(bytes), + signal, + }); + if (!response.ok) { + throw new Org2CloudStorageError( + `replay object upload failed with ${response.status}`, + response.status + ); + } +} + +export async function downloadReplayObject( + accessToken: string, + path: string, + endpoint: CloudEndpoint = getCloudEndpoint(), + signal?: AbortSignal +): Promise { + const response = await fetchWithTransportRetry(objectUrl(path, endpoint), { + method: "GET", + headers: objectHeaders(accessToken, endpoint), + signal, + }); + if (!response.ok) { + throw new Org2CloudStorageError( + `replay object download failed with ${response.status}`, + response.status + ); + } + return new Uint8Array(await response.arrayBuffer()); +} diff --git a/src/features/Org2Cloud/org2CloudSyncClient.test.ts b/src/features/Org2Cloud/org2CloudSyncClient.test.ts index dc0e7cc52..a03f61052 100644 --- a/src/features/Org2Cloud/org2CloudSyncClient.test.ts +++ b/src/features/Org2Cloud/org2CloudSyncClient.test.ts @@ -2,15 +2,21 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { SessionEvent } from "@src/engines/SessionCore/core/types"; -import { decodeSegmentEvents } from "../TeamCollaboration/sync/segmentCodec"; +import { computeSegmentHash } from "../TeamCollaboration/sync/collabGzip"; +import { + decodeSegmentEvents, + decodeSegmentEventsFromBytes, +} from "../TeamCollaboration/sync/segmentCodec"; import { ORG2_CLOUD_OFFICIAL_ANON_KEY, ORG2_CLOUD_OFFICIAL_SUPABASE_URL, ORG2_CLOUD_POSTGREST_SCHEMA, } from "./config"; +import { getCloudCapabilities } from "./org2CloudCapabilities"; import { Org2CloudSyncError, __SESSION_LISTING_INTERNALS, + __STORAGE_SEGMENTS_INTERNALS, appendSessionEvents, getOrgRepoScopes, getSessionEvents, @@ -21,6 +27,12 @@ import { upsertSessionMetadata, } from "./org2CloudSyncClient"; +vi.mock("./org2CloudCapabilities", () => ({ + getCloudCapabilities: vi.fn(), +})); + +const capabilitiesMock = vi.mocked(getCloudCapabilities); + const fetchMock = vi.fn(); function jsonResponse(body: unknown, status = 200): Response { @@ -46,11 +58,17 @@ function makeEvent(id: string): SessionEvent { beforeEach(() => { vi.stubGlobal("fetch", fetchMock); fetchMock.mockResolvedValue(jsonResponse(null)); + capabilitiesMock.mockResolvedValue({ + broadcastSignals: false, + storageSegments: false, + }); }); afterEach(() => { vi.unstubAllGlobals(); fetchMock.mockReset(); + capabilitiesMock.mockReset(); + __STORAGE_SEGMENTS_INTERNALS.resetStorageSupport(); }); describe("org2CloudSyncClient headers", () => { @@ -235,6 +253,160 @@ describe("cloud_rewrite_session_events", () => { }); }); +describe("storage segment offload (0006)", () => { + beforeEach(() => { + capabilitiesMock.mockResolvedValue({ + broadcastSignals: false, + storageSegments: true, + }); + }); + + function appendInput(frozen: SessionEvent[], tail: SessionEvent[] | null) { + return { + orgId: "org-1", + sessionId: "s-1", + expectedEpoch: 2, + expectedFrozenSeq: 5, + expectedTailHash: "hash-old-tail", + newFrozenSegments: frozen.length > 0 ? [{ seq: 6, events: frozen }] : [], + tail, + totalCount: 7, + }; + } + + it("uploads frozen segment objects and ships storagePath wire with an inline tail", async () => { + const frozen = [makeEvent("f1")]; + const tail = [makeEvent("t1")]; + await appendSessionEvents("jwt-1", appendInput(frozen, tail)); + + expect(fetchMock).toHaveBeenCalledTimes(2); + const hash = await computeSegmentHash(frozen); + const path = `org-1/s-1/2/6-${hash}.gz`; + const [uploadUrl, uploadInit] = fetchMock.mock.calls[0] as [ + string, + RequestInit, + ]; + expect(uploadUrl).toBe( + `${ORG2_CLOUD_OFFICIAL_SUPABASE_URL}/storage/v1/object/replay/${path}` + ); + expect(uploadInit.method).toBe("PUT"); + const uploadHeaders = uploadInit.headers as Record; + expect(uploadHeaders.authorization).toBe("Bearer jwt-1"); + expect(uploadHeaders["content-type"]).toBe("application/gzip"); + expect(uploadHeaders["x-upsert"]).toBe("true"); + expect( + await decodeSegmentEventsFromBytes(uploadInit.body as Uint8Array) + ).toEqual(frozen); + + expect(lastCall().url).toBe( + `${ORG2_CLOUD_OFFICIAL_SUPABASE_URL}/rest/v1/rpc/cloud_append_session_events` + ); + const body = lastBody(); + expect(body.new_frozen_segments).toEqual([ + { seq: 6, storagePath: path, eventCount: 1, segmentHash: hash }, + ]); + const tailWire = body.tail as Record; + expect(typeof tailWire.payloadGz).toBe("string"); + expect(tailWire).not.toHaveProperty("storagePath"); + }); + + it("rewrite keys the object paths by the new epoch", async () => { + const frozen = [makeEvent("f1")]; + await rewriteSessionEvents("jwt-1", { + orgId: "org-1", + sessionId: "s-1", + newEpoch: 4, + frozenSegments: [{ seq: 1, events: frozen }], + tail: null, + totalCount: 1, + }); + expect(fetchMock).toHaveBeenCalledTimes(2); + const hash = await computeSegmentHash(frozen); + const path = `org-1/s-1/4/1-${hash}.gz`; + expect((fetchMock.mock.calls[0] as [string, RequestInit])[0]).toBe( + `${ORG2_CLOUD_OFFICIAL_SUPABASE_URL}/storage/v1/object/replay/${path}` + ); + const body = lastBody(); + expect(body.new_epoch).toBe(4); + expect(body.frozen_segments).toEqual([ + { seq: 1, storagePath: path, eventCount: 1, segmentHash: hash }, + ]); + }); + + it("keeps the legacy inline wire when the capabilities probe says false", async () => { + capabilitiesMock.mockResolvedValue({ + broadcastSignals: false, + storageSegments: false, + }); + await appendSessionEvents("jwt-1", appendInput([makeEvent("f1")], null)); + expect(fetchMock).toHaveBeenCalledTimes(1); + const segments = lastBody().new_frozen_segments as Array< + Record + >; + expect(typeof segments[0].payloadGz).toBe("string"); + expect(segments[0]).not.toHaveProperty("storagePath"); + }); + + it("skips the probe and uploads entirely for a tail-only append", async () => { + await appendSessionEvents("jwt-1", appendInput([], [makeEvent("t1")])); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(capabilitiesMock).not.toHaveBeenCalled(); + expect(lastBody().new_frozen_segments).toEqual([]); + }); + + it("falls back to the inline wire on a missing-function rejection and remembers the endpoint", async () => { + fetchMock + .mockResolvedValueOnce(jsonResponse(null)) + .mockResolvedValueOnce( + jsonResponse( + { + code: "PGRST202", + message: + "Could not find the function org2_cloud.cloud_append_session_events in the schema cache", + }, + 404 + ) + ) + .mockResolvedValueOnce(jsonResponse(null)); + await appendSessionEvents("jwt-1", appendInput([makeEvent("f1")], null)); + expect(fetchMock).toHaveBeenCalledTimes(3); + const segments = lastBody().new_frozen_segments as Array< + Record + >; + expect(typeof segments[0].payloadGz).toBe("string"); + expect(segments[0]).not.toHaveProperty("storagePath"); + + fetchMock.mockResolvedValueOnce(jsonResponse(null)); + await appendSessionEvents("jwt-1", appendInput([makeEvent("f2")], null)); + expect(fetchMock).toHaveBeenCalledTimes(4); + expect(String(lastCall().url)).toContain("/rest/v1/rpc/"); + }); + + it("propagates ORG2_VALIDATION on the storage form without falling back", async () => { + fetchMock + .mockResolvedValueOnce(jsonResponse(null)) + .mockResolvedValueOnce(jsonResponse({ message: "ORG2_VALIDATION" }, 400)); + const error = await appendSessionEvents( + "jwt-1", + appendInput([makeEvent("f1")], null) + ).catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(Org2CloudSyncError); + expect((error as Org2CloudSyncError).message).toContain("ORG2_VALIDATION"); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("propagates an upload failure before any RPC is attempted", async () => { + fetchMock.mockResolvedValueOnce(new Response(null, { status: 403 })); + await expect( + appendSessionEvents("jwt-1", appendInput([makeEvent("f1")], null)) + ).rejects.toSatisfy( + (error: unknown) => + error instanceof Error && error.name === "Org2CloudStorageError" + ); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); +}); + describe("cloud_list_org_sessions", () => { it("parses the retention-windowed listing", async () => { fetchMock.mockResolvedValueOnce( @@ -467,6 +639,50 @@ describe("cloud_get_session_events", () => { expect(result).toMatchObject({ epoch: 1, count: 0, segments: [] }); }); + it("parses storagePath segments on the read wire", async () => { + fetchMock.mockResolvedValueOnce( + jsonResponse({ + epoch: 4, + frozenSeq: 1, + tailHash: "th", + count: 3, + nextAfterSeq: 1, + hasMore: false, + segments: [ + { + seq: 1, + storagePath: "org-1/s-1/4/1-h1.gz", + payloadGz: null, + eventCount: 2, + segmentHash: "h1", + }, + { seq: 0, payloadGz: "tail", eventCount: 1, segmentHash: "th" }, + ], + }) + ); + const result = await getSessionEvents("jwt-1", "org-1", "s-1"); + expect(result.segments[0].storagePath).toBe("org-1/s-1/4/1-h1.gz"); + expect(result.segments[0].payloadGz).toBeNull(); + expect(result.segments[1].payloadGz).toBe("tail"); + }); + + it("rejects a segment carrying neither payloadGz nor storagePath", async () => { + fetchMock.mockResolvedValueOnce( + jsonResponse({ + epoch: 1, + frozenSeq: 1, + tailHash: null, + count: 1, + nextAfterSeq: 1, + hasMore: false, + segments: [{ seq: 1, eventCount: 1, segmentHash: "h1" }], + }) + ); + await expect(getSessionEvents("jwt-1", "org-1", "s-1")).rejects.toThrow( + /neither payloadGz nor storagePath/ + ); + }); + it("maps ORG2_RETENTION_EXPIRED into a coded error", async () => { fetchMock.mockResolvedValueOnce( jsonResponse({ message: "ORG2_RETENTION_EXPIRED" }, 403) diff --git a/src/features/Org2Cloud/org2CloudSyncClient.ts b/src/features/Org2Cloud/org2CloudSyncClient.ts index 6af617114..80c892869 100644 --- a/src/features/Org2Cloud/org2CloudSyncClient.ts +++ b/src/features/Org2Cloud/org2CloudSyncClient.ts @@ -25,8 +25,8 @@ import type { import type { SessionEventsSegmentInput } from "../TeamCollaboration/sync/CollabSyncBackend"; import { - type SegmentWirePayload, mapSegmentsBounded, + toFrozenSegmentStorage, toFrozenSegmentWire, toTailWire, } from "../TeamCollaboration/sync/segmentCodec"; @@ -35,10 +35,15 @@ import { ORG2_CLOUD_POSTGREST_SCHEMA, getCloudEndpoint, } from "./config"; +import { getCloudCapabilities } from "./org2CloudCapabilities"; import { fetchWithTransportRetry, runCloudRequestWithTimeout, } from "./org2CloudFetchRetry"; +import { + buildReplayObjectPath, + uploadReplayObject, +} from "./org2CloudStorageClient"; // --------------------------------------------------------------------------- // Error model @@ -149,12 +154,20 @@ async function callSyncRpc( // Wire schemas // --------------------------------------------------------------------------- -const CloudSegmentWireSchema = z.object({ - seq: z.number(), - payloadGz: z.string(), - eventCount: z.number(), - segmentHash: z.string(), -}); +const CloudSegmentWireSchema = z + .object({ + seq: z.number(), + // 0006 storage offload: a frozen segment carries storagePath XOR the + // legacy inline payloadGz. The tail (seq 0) is always inline. + payloadGz: z.string().nullish(), + storagePath: z.string().nullish(), + eventCount: z.number(), + segmentHash: z.string(), + }) + .refine( + (segment) => segment.payloadGz != null || segment.storagePath != null, + { message: "segment carries neither payloadGz nor storagePath" } + ); const CloudSessionEventsSchema = z.object({ epoch: z.number().nullish().default(null), @@ -216,13 +229,22 @@ const CloudOrgSessionsSchema = z.object({ .catch(undefined), }); +/** Read-side segment wire: inline (`payloadGz`) or offloaded (`storagePath`). */ +export interface CloudSegmentWire { + seq?: number; + payloadGz?: string | null; + storagePath?: string | null; + eventCount: number; + segmentHash: string; +} + export interface CloudSessionEventsSnapshot { epoch: number | null; frozenSeq: number | null; tailHash: string | null; /** Total event count (frozen + tail); null ⇒ nothing published yet. */ count: number | null; - segments: SegmentWirePayload[]; + segments: CloudSegmentWire[]; } export type CloudSessionEventsSummary = Omit< @@ -242,7 +264,7 @@ const SESSION_LISTING_MAX_PAGES = 50; /** supabaseUrl set of backends that rejected the paged signature (pre-0005). */ const paginationUnsupportedEndpoints = new Set(); -function isPagedSignatureUnsupported(error: unknown): boolean { +function isRpcSignatureUnsupported(error: unknown): boolean { return ( error instanceof Org2CloudSyncError && error.status === 404 && @@ -254,6 +276,58 @@ export const __SESSION_LISTING_INTERNALS = { resetPaginationSupport: () => paginationUnsupportedEndpoints.clear(), }; +/** supabaseUrl set of backends that rejected the storage segment wire (pre-0006). */ +const storageSegmentsUnsupportedEndpoints = new Set(); + +export const __STORAGE_SEGMENTS_INTERNALS = { + resetStorageSupport: () => storageSegmentsUnsupportedEndpoints.clear(), +}; + +async function shouldUseStorageSegments( + accessToken: string, + endpoint: CloudEndpoint +): Promise { + if (storageSegmentsUnsupportedEndpoints.has(endpoint.supabaseUrl)) { + return false; + } + return (await getCloudCapabilities(accessToken)).storageSegments; +} + +interface CloudStorageSegmentWire { + seq: number; + storagePath: string; + eventCount: number; + segmentHash: string; +} + +/** Upload each frozen segment's raw gzip bytes, then describe it by path. */ +async function uploadFrozenSegmentsToStorage( + accessToken: string, + endpoint: CloudEndpoint, + orgId: string, + sessionId: string, + epoch: number, + segments: SessionEventsSegmentInput[] +): Promise { + return mapSegmentsBounded(segments, async (segment) => { + const encoded = await toFrozenSegmentStorage(segment); + const storagePath = buildReplayObjectPath( + orgId, + sessionId, + epoch, + encoded.seq, + encoded.segmentHash + ); + await uploadReplayObject(accessToken, storagePath, encoded.bytes, endpoint); + return { + seq: encoded.seq, + storagePath, + eventCount: encoded.eventCount, + segmentHash: encoded.segmentHash, + }; + }); +} + export type CloudOrgScopeState = z.output; // --------------------------------------------------------------------------- @@ -357,19 +431,53 @@ export async function rewriteSessionEvents( accessToken: string, input: CloudRewriteSessionEventsInput ): Promise { - await callSyncRpc("cloud_rewrite_session_events", accessToken, { + const endpoint = getCloudEndpoint(); + const baseBody = { p_org_id: input.orgId, p_session_id: input.sessionId, new_epoch: input.newEpoch, - // Bounded encode: `Promise.all` over every segment materializes all - // canonical/gzip/base64 buffers simultaneously and multiplies RSS. - frozen_segments: await mapSegmentsBounded( - input.frozenSegments, - toFrozenSegmentWire - ), tail: await toTailWire(input.tail), total_count: input.totalCount, - }); + }; + if ( + input.frozenSegments.length > 0 && + (await shouldUseStorageSegments(accessToken, endpoint)) + ) { + const frozenSegments = await uploadFrozenSegmentsToStorage( + accessToken, + endpoint, + input.orgId, + input.sessionId, + input.newEpoch, + input.frozenSegments + ); + try { + await callSyncRpc( + "cloud_rewrite_session_events", + accessToken, + { ...baseBody, frozen_segments: frozenSegments }, + endpoint + ); + return; + } catch (error) { + if (!isRpcSignatureUnsupported(error)) throw error; + storageSegmentsUnsupportedEndpoints.add(endpoint.supabaseUrl); + } + } + await callSyncRpc( + "cloud_rewrite_session_events", + accessToken, + { + ...baseBody, + // Bounded encode: `Promise.all` over every segment materializes all + // canonical/gzip/base64 buffers simultaneously and multiplies RSS. + frozen_segments: await mapSegmentsBounded( + input.frozenSegments, + toFrozenSegmentWire + ), + }, + endpoint + ); } export interface CloudAppendSessionEventsInput { @@ -389,19 +497,53 @@ export async function appendSessionEvents( accessToken: string, input: CloudAppendSessionEventsInput ): Promise { - await callSyncRpc("cloud_append_session_events", accessToken, { + const endpoint = getCloudEndpoint(); + const baseBody = { p_org_id: input.orgId, p_session_id: input.sessionId, expected_epoch: input.expectedEpoch, expected_frozen_seq: input.expectedFrozenSeq, expected_tail_hash: input.expectedTailHash, - new_frozen_segments: await mapSegmentsBounded( - input.newFrozenSegments, - toFrozenSegmentWire - ), tail: await toTailWire(input.tail), total_count: input.totalCount, - }); + }; + if ( + input.newFrozenSegments.length > 0 && + (await shouldUseStorageSegments(accessToken, endpoint)) + ) { + const newFrozenSegments = await uploadFrozenSegmentsToStorage( + accessToken, + endpoint, + input.orgId, + input.sessionId, + input.expectedEpoch, + input.newFrozenSegments + ); + try { + await callSyncRpc( + "cloud_append_session_events", + accessToken, + { ...baseBody, new_frozen_segments: newFrozenSegments }, + endpoint + ); + return; + } catch (error) { + if (!isRpcSignatureUnsupported(error)) throw error; + storageSegmentsUnsupportedEndpoints.add(endpoint.supabaseUrl); + } + } + await callSyncRpc( + "cloud_append_session_events", + accessToken, + { + ...baseBody, + new_frozen_segments: await mapSegmentsBounded( + input.newFrozenSegments, + toFrozenSegmentWire + ), + }, + endpoint + ); } /** Member: retention-windowed session listing for one cloud org. */ @@ -460,7 +602,7 @@ export async function listOrgSessions( 15_000 ); } catch (error) { - if (page === 0 && isPagedSignatureUnsupported(error)) { + if (page === 0 && isRpcSignatureUnsupported(error)) { paginationUnsupportedEndpoints.add(endpoint.supabaseUrl); parsed = await legacyCall(); break; @@ -545,7 +687,7 @@ export async function getSessionEvents( sessionId: string, options?: GetSessionEventsOptions ): Promise { - const segments: SegmentWirePayload[] = []; + const segments: CloudSegmentWire[] = []; const summary = await streamSessionEvents( accessToken, orgId, diff --git a/src/features/TeamCollaboration/sync/collabGzip.ts b/src/features/TeamCollaboration/sync/collabGzip.ts index a0617a51e..8aa8410f1 100644 --- a/src/features/TeamCollaboration/sync/collabGzip.ts +++ b/src/features/TeamCollaboration/sync/collabGzip.ts @@ -89,13 +89,14 @@ export async function computeSegmentHash(value: unknown): Promise { return computeSegmentHashFromBytes(segmentCanonicalBytes(value)); } +/** gzip over already-encoded canonical bytes (no base64 leg). */ +export async function gzipBytes(bytes: Uint8Array): Promise { + return pipeThroughStream(bytes, new CompressionStream("gzip")); +} + /** gzip → base64 over already-encoded canonical bytes. */ export async function gzipBytesToBase64(bytes: Uint8Array): Promise { - const compressed = await pipeThroughStream( - bytes, - new CompressionStream("gzip") - ); - return bytesToBase64(compressed); + return bytesToBase64(await gzipBytes(bytes)); } /** JSON → gzip → base64 (the client half of `payload_gz`). */ @@ -103,12 +104,18 @@ export async function gzipJsonToBase64(value: unknown): Promise { return gzipBytesToBase64(segmentCanonicalBytes(value)); } -/** base64 → gunzip → JSON (the client half of reading `payload_gz`). */ -export async function gunzipBase64ToJson(base64: string): Promise { - const compressed = base64ToBytes(base64); +/** gunzip → JSON over raw gzip bytes (storage-object downloads). */ +export async function gunzipBytesToJson( + compressed: Uint8Array +): Promise { const bytes = await pipeThroughStream( compressed, new DecompressionStream("gzip") ); return JSON.parse(new TextDecoder().decode(bytes)) as unknown; } + +/** base64 → gunzip → JSON (the client half of reading `payload_gz`). */ +export async function gunzipBase64ToJson(base64: string): Promise { + return gunzipBytesToJson(base64ToBytes(base64)); +} diff --git a/src/features/TeamCollaboration/sync/segmentCodec.test.ts b/src/features/TeamCollaboration/sync/segmentCodec.test.ts index 4c62d6797..28cae0bee 100644 --- a/src/features/TeamCollaboration/sync/segmentCodec.test.ts +++ b/src/features/TeamCollaboration/sync/segmentCodec.test.ts @@ -5,7 +5,9 @@ import type { SessionEvent } from "@src/engines/SessionCore/core/types"; import { computeSegmentHash } from "./collabGzip"; import { decodeSegmentEvents, + decodeSegmentEventsFromBytes, mapSegmentsBounded, + toFrozenSegmentStorage, toFrozenSegmentWire, toTailWire, } from "./segmentCodec"; @@ -29,6 +31,18 @@ describe("segmentCodec", () => { expect(await decodeSegmentEvents(wire.payloadGz)).toEqual(events); }); + it("builds a raw-gzip storage payload that round-trips and matches the inline hash", async () => { + const events = [makeEvent("e1"), makeEvent("e2")]; + const stored = await toFrozenSegmentStorage({ seq: 3, events }); + + expect(stored.seq).toBe(3); + expect(stored.eventCount).toBe(2); + expect(stored.segmentHash).toBe(await computeSegmentHash(events)); + expect(await decodeSegmentEventsFromBytes(stored.bytes)).toEqual(events); + const wire = await toFrozenSegmentWire({ seq: 3, events }); + expect(stored.segmentHash).toBe(wire.segmentHash); + }); + it("builds a tail wire payload without a seq", async () => { const events = [makeEvent("t1")]; const wire = await toTailWire(events); diff --git a/src/features/TeamCollaboration/sync/segmentCodec.ts b/src/features/TeamCollaboration/sync/segmentCodec.ts index 5c1cdc67e..877315c0b 100644 --- a/src/features/TeamCollaboration/sync/segmentCodec.ts +++ b/src/features/TeamCollaboration/sync/segmentCodec.ts @@ -10,6 +10,8 @@ import type { SessionEventsSegmentInput } from "./CollabSyncBackend"; import { computeSegmentHashFromBytes, gunzipBase64ToJson, + gunzipBytesToJson, + gzipBytes, gzipBytesToBase64, segmentCanonicalBytes, } from "./collabGzip"; @@ -22,6 +24,14 @@ export interface SegmentWirePayload { segmentHash: string; } +/** Raw-gzip encode of one frozen segment for a Storage object upload. */ +export interface SegmentStoragePayload { + seq: number; + bytes: Uint8Array; + eventCount: number; + segmentHash: string; +} + export async function toFrozenSegmentWire( segment: SessionEventsSegmentInput ): Promise { @@ -36,6 +46,18 @@ export async function toFrozenSegmentWire( }; } +export async function toFrozenSegmentStorage( + segment: SessionEventsSegmentInput +): Promise { + const bytes = segmentCanonicalBytes(segment.events); + return { + seq: segment.seq, + bytes: await gzipBytes(bytes), + eventCount: segment.events.length, + segmentHash: await computeSegmentHashFromBytes(bytes), + }; +} + export async function toTailWire( tail: SessionEvent[] | null ): Promise | null> { @@ -56,6 +78,14 @@ export async function decodeSegmentEvents( .parse(await gunzipBase64ToJson(payloadGz)); } +export async function decodeSegmentEventsFromBytes( + bytes: Uint8Array +): Promise { + return z + .array(z.custom()) + .parse(await gunzipBytesToJson(bytes)); +} + /** * Codec-side concurrency budget. Encoding/decoding every segment through * `Promise.all` materializes all canonical byte buffers, gzip streams and From d305a212552443c03ccaca4d72eaf645ad3df24b Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Fri, 24 Jul 2026 00:28:38 -0700 Subject: [PATCH 4/4] docs: descriptive names for the broadcast and storage design notes Pre-commit hook ran. Total eslint: 1, total circular: 0 --- .../broadcast-change-signals.md} | 4 ++-- .../replay-storage-offload.md} | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) rename docs/{h4-h5-design-2026-07-24/h4-broadcast.md => cloud-broadcast-and-storage-design-2026-07-24/broadcast-change-signals.md} (99%) rename docs/{h4-h5-design-2026-07-24/h5-storage-offload.md => cloud-broadcast-and-storage-design-2026-07-24/replay-storage-offload.md} (98%) diff --git a/docs/h4-h5-design-2026-07-24/h4-broadcast.md b/docs/cloud-broadcast-and-storage-design-2026-07-24/broadcast-change-signals.md similarity index 99% rename from docs/h4-h5-design-2026-07-24/h4-broadcast.md rename to docs/cloud-broadcast-and-storage-design-2026-07-24/broadcast-change-signals.md index 655d1a667..496185cd6 100644 --- a/docs/h4-h5-design-2026-07-24/h4-broadcast.md +++ b/docs/cloud-broadcast-and-storage-design-2026-07-24/broadcast-change-signals.md @@ -1,8 +1,8 @@ -# H4 — Change Signals via Broadcast-from-Database +# Change Signals via Broadcast-from-Database (audit item H4) Date: 2026-07-24. Basis: realtime signal-chain architecture map (agent survey of 0001/0003 server path + full client subscription topology). Companion: -`h5-storage-offload.md`. +`replay-storage-offload.md`. ## Why diff --git a/docs/h4-h5-design-2026-07-24/h5-storage-offload.md b/docs/cloud-broadcast-and-storage-design-2026-07-24/replay-storage-offload.md similarity index 98% rename from docs/h4-h5-design-2026-07-24/h5-storage-offload.md rename to docs/cloud-broadcast-and-storage-design-2026-07-24/replay-storage-offload.md index 60b4019f6..af5150c58 100644 --- a/docs/h4-h5-design-2026-07-24/h5-storage-offload.md +++ b/docs/cloud-broadcast-and-storage-design-2026-07-24/replay-storage-offload.md @@ -1,7 +1,7 @@ -# H5 — Replay Segment Payload Offload to Supabase Storage +# Replay Segment Payload Offload to Supabase Storage (audit item H5) Date: 2026-07-24. Basis: segment-lifecycle architecture map (agent survey of -0001/0002/0003 + client push/pull paths). Companion: `h4-broadcast.md`. +0001/0002/0003 + client push/pull paths). Companion: `broadcast-change-signals.md`. ## Decision summary