diff --git a/docs/CREW_SYNC_V2_PHASES.md b/docs/CREW_SYNC_V2_PHASES.md index b09d113f..bf70506e 100644 --- a/docs/CREW_SYNC_V2_PHASES.md +++ b/docs/CREW_SYNC_V2_PHASES.md @@ -36,34 +36,23 @@ anything missed. | 2 | Broadcast infrastructure (DB triggers + RLS on `realtime.messages`) | ✅ Done — migration `20260725191358_crew_sync_broadcast_triggers.sql`, merged via PR #508 (2026-07-26), applied to prod **and** e2e project. Deploys dark per design — no client subscribes yet | | 3 | Client cutover (single private broadcast channel, behind a flag) | ✅ Done — merged via PR #508 (2026-07-26), `lib/dexie/context.tsx`. Ships dormant: `NEXT_PUBLIC_CREW_SYNC_V2` defaults off (unset in `.env.example`) | | 4 | Outbox retry backoff | ✅ Done — merged via PR #508 (2026-07-26), `lib/dexie/syncService.ts`'s `computeNextAttemptAt()` | -| 5 | Rollout, acceptance test, old-code deletion, convention + guardrail | 🟡 **In progress** — 5e (convention + `unit/guardrails/crew-sync-coverage.test.ts`) done 2026-07-29, `CREW_SYNCED_TABLES`/`LOCAL_ONLY_TABLES` added to `lib/dexie/schema.ts`. 5a-5d (Realtime quota check, flag flip, acceptance test, soak, deletion) still open — see section 5 below | +| 5 | Rollout, acceptance test, old-code deletion, convention + guardrail | 🟡 **In progress** — 5e (convention + `unit/guardrails/crew-sync-coverage.test.ts`) done 2026-07-29. 5b two-device acceptance test passed 2026-08-12; flag set in Vercel and client default inverted to on the same day, so **v2 is the serving path**. Still open: 5a Realtime quota check, 5c soak, 5d deletion of the v1 path | -> **2026-08-09 — flag flipped, redeploy still required.** `NEXT_PUBLIC_*` is -> inlined at BUILD time (see the note at `lib/dexie/context.tsx`'s -> `CREW_SYNC_V2` and `clientInlinedOnly: true` in `lib/env.ts`), so setting the -> var in Vercel does nothing to a bundle that was already built. v2 turns on at -> the next **rebuild**, not the next restart. Until then v1 is still serving. +> **2026-08-12 — v2 is the DEFAULT and is live.** `NEXT_PUBLIC_CREW_SYNC_V2` +> is set to `true` in Vercel and many builds have shipped since it was set, so +> v2 has been the serving path for some time — the 2026-08-09 note this +> replaces ("redeploy still required") was already stale when written against a +> branch that redeployed on every push. > -> Everything else on the DB side is verified live in production (2026-08-09): -> migration `20260725191358` in the ledger, `notify_crew_sync` present, -> 12 broadcast triggers attached, RLS policy `crew_receive_own_sync_broadcasts` -> on `realtime.messages`, and the trigger's `crew_members.user_id` join matches -> both the client's `crew:${userId}` topic and the policy's `auth.uid()`. -> `realtime.send()` was probed end to end and does land a row. +> The client default is now inverted (`!== 'false'`), so v2 serves even where +> the variable is absent. That closes a real hole: `NEXT_PUBLIC_*` is inlined +> at BUILD time, so an environment missing the variable silently served v1 with +> nothing in config to show it. Rollback is `NEXT_PUBLIC_CREW_SYNC_V2=false` +> plus a rebuild — still a build-time change, not a runtime toggle. > -> Still open before the soak: 5b (two-device acceptance test — never run), then -> 5c. Do NOT do 5d yet: the v1 `postgres_changes` code is the rollback path, and -> rolling back is another rebuild. -> -> Expect one deliberate behaviour change: v1 had a `postgres_changes` channel on -> `property_assets`; v2 has no trigger for it (`SAFETY_POLL_ONLY` in -> `unit/guardrails/crew-sync-coverage.test.ts`). Asset edits reach the device on -> screen-open or the safety poll rather than in ~2 s. -> -> Watch the Realtime DB pool (2 → 15, per 5a). Private-channel RLS -> authorization takes a connection on every join AND every reconnect, v2 is -> all-private-channel, and the reconnect jitter is 5–35 s — so a Realtime node -> restart puts the whole fleet through that pool inside 30 seconds. +> 5b (two-device acceptance test) passed 2026-08-12. Still open: 5a's Realtime +> quota check, the 5c soak, and 5d deletion of the v1 path — which stays in the +> tree, compiling and guardrail-covered, until then. ### Phase 1 artifacts you will build on (read these before touching code) diff --git a/lib/dexie/context.tsx b/lib/dexie/context.tsx index ea3869a2..ecb5b1d5 100644 --- a/lib/dexie/context.tsx +++ b/lib/dexie/context.tsx @@ -23,11 +23,27 @@ import { } from './sync/signals' // Crew Sync v2 (docs/CREW_SYNC_V2_PHASES.md Phase 3): broadcast signal + -// delta pull instead of postgres_changes. Ships dormant — the flag defaults -// off and the v1 path below stays the production behavior until Phase 5. -// NEXT_PUBLIC_ vars are inlined at build time, so this is a build-time -// constant, not a runtime toggle. -const CREW_SYNC_V2 = process.env.NEXT_PUBLIC_CREW_SYNC_V2 === 'true' +// delta pull instead of postgres_changes. +// +// DEFAULT-ON as of 2026-08-12, after the Phase 5b two-device acceptance test +// passed. `!== 'false'` rather than `=== 'true'`, and the direction is the +// whole point: absent used to mean v1, so any environment that did not carry +// the variable — a new preview, a rebuilt environment, a setting lost in a +// project migration — silently served crew the old sync path with nothing to +// indicate it. Since NEXT_PUBLIC_ vars are inlined at BUILD time, that state +// is invisible in config and only observable from behaviour. Now the fallback +// requires someone to ask for it explicitly. +// +// The v1 path below is deliberately still here and still compiling. It is the +// rollback: set NEXT_PUBLIC_CREW_SYNC_V2=false and rebuild, no code change. +// Phase 5d deletes it after a green soak week — commenting it out in the +// meantime would buy nothing (it is already unreachable when the flag is on) +// while costing type-checking, lint, guardrail coverage, and a one-variable +// rollback. +// +// Build-time constant, not a runtime toggle: changing the variable in Vercel +// does nothing to an already-built bundle. It takes effect on the next build. +const CREW_SYNC_V2 = process.env.NEXT_PUBLIC_CREW_SYNC_V2 !== 'false' /** How often the safety poll runs a full resync — the correctness backstop * for missed Realtime events/broadcasts and the only freshness path for diff --git a/lib/inngest/functions/ownerrez/incremental-sync.ts b/lib/inngest/functions/ownerrez/incremental-sync.ts index 1c9d7031..dcac4de6 100644 --- a/lib/inngest/functions/ownerrez/incremental-sync.ts +++ b/lib/inngest/functions/ownerrez/incremental-sync.ts @@ -67,6 +67,7 @@ import { partitionMappedBookingRows, selectOwnerRezBookingsToPostRevenue, } from '@/lib/integrations/providers/ownerrez' +import { upsertBookingsReturningIds } from './upsert-bookings' import type { MappedOwnerRezBookingRow } from '@/lib/integrations/providers/ownerrez' import { mergeIntegrationConnectionMetadata } from '@/lib/integrations/connection-metadata' import { unwrap, unwrapList, isRealQueryError } from '@/lib/supabase/unwrap' @@ -392,19 +393,14 @@ async function persistBookings( return { affectedPropertyIds: [], bookingsToPostRevenue: [], ownerBlocks: [] } } - const { data: upserted, error } = await supabase - .from('bookings') - .upsert(bookingRows, { onConflict: 'org_id,external_id,external_source' }) - .select('id, external_id') - - if (error) { - logger.error(`[OwnerRez:${conn.user_id}] bookings upsert: ${error.message}`) - throw new Error(error.message) - } - - const idByExternalId = Object.fromEntries( - (upserted ?? []).map((row) => [row.external_id, row.id as string]) - ) + // Chunked: an .upsert().select() returns at most max_rows = 1000 rows with + // no truncation signal, and a short id map silently drops those bookings + // from revenue posting. See upsert-bookings.ts. + // No try/catch: the helper's message already carries the connection label, + // and the enclosing step.run surfaces the throw. Catching only to re-log and + // re-throw would add a log-only catch block for no added context. + const idByExternalId = await upsertBookingsReturningIds( + supabase, bookingRows, `OwnerRez:${conn.user_id}`) return { affectedPropertyIds: Array.from(new Set(bookingRows.map((b) => b.property_id))), diff --git a/lib/inngest/functions/ownerrez/initial-sync.ts b/lib/inngest/functions/ownerrez/initial-sync.ts index 3001c8e4..536e9c2f 100644 --- a/lib/inngest/functions/ownerrez/initial-sync.ts +++ b/lib/inngest/functions/ownerrez/initial-sync.ts @@ -22,6 +22,7 @@ import { partitionMappedBookingRows, selectOwnerRezBookingsToPostRevenue, } from '@/lib/integrations/providers/ownerrez' +import { upsertBookingsReturningIds } from './upsert-bookings' import { logAuditEvent } from '@/lib/audit' import { applyMasterChecklistToProperty, @@ -559,15 +560,14 @@ export const ownerRezInitialSync = inngest.createFunction( ) } - const { data: upserted, error } = await supabase - .from('bookings') - .upsert(bookingRows, { onConflict: 'org_id,external_id,external_source' }) - .select('id, external_id') - - if (error) { - logger.error(`[OwnerRez:${user_id}] bookings upsert failed: ${error.message}`) - throw new Error(error.message) - } + // Chunked, and this is the sync where it matters most: it fetches + // every booking for every property in one call, so a portfolio with + // history clears max_rows = 1000 on its FIRST run — the one whose + // job is to get the historical owner ledger right. A truncated + // representation there silently omitted revenue for every booking + // past the first 1000. See upsert-bookings.ts. + const idByExternalId = await upsertBookingsReturningIds( + supabase, bookingRows, `OwnerRez:${user_id}`) logger.info(`[OwnerRez:${user_id}] Upserted ${bookingRows.length} bookings`) @@ -575,10 +575,6 @@ export const ownerRezInitialSync = inngest.createFunction( bookingRows.map((b) => b.property_id).filter((id): id is string => id !== null) )) - const idByExternalId = Object.fromEntries( - (upserted ?? []).map((row) => [row.external_id, row.id as string]) - ) - bookingsToPostRevenue = selectOwnerRezBookingsToPostRevenue(bookingRows, idByExternalId) } diff --git a/lib/inngest/functions/ownerrez/upsert-bookings.ts b/lib/inngest/functions/ownerrez/upsert-bookings.ts new file mode 100644 index 00000000..d7efae82 --- /dev/null +++ b/lib/inngest/functions/ownerrez/upsert-bookings.ts @@ -0,0 +1,90 @@ +import 'server-only' + +import type { SupabaseClient } from '@supabase/supabase-js' +import { SUPABASE_MAX_ROWS } from '@/lib/inngest/paginate' +import type { OwnerRezBookingRow } from '@/lib/integrations/providers/ownerrez' + +/** + * Upsert booking rows and return a COMPLETE external_id -> id map. + * + * THE BUG THIS EXISTS TO PREVENT, because it is invisible in every way that + * matters. Both OwnerRez syncs used to upsert the whole booking set in ONE + * call and build the id map from that call's returned rows — + * `.upsert(bookingRows, …).select('id, external_id')`, once, however many + * bookings there were. + * + * (Written as prose rather than as a code sample on purpose: a literal + * `const { … } = await supabase…` in this comment is matched by + * unit/guardrails/supabase-error-handling.test.ts, which scans raw source and + * cannot tell an illustration of a defect from the defect.) + * + * The WRITE is not capped — every row lands. The returned REPRESENTATION is: + * PostgREST caps it at max_rows = 1000 with a 200 and no truncation signal, + * exactly as it does for a read. So on a portfolio with more than 1000 + * bookings the map came back short, and + * selectOwnerRezBookingsToPostRevenue's closing + * + * .filter((b) => !!b.bookingId) + * + * silently dropped every booking whose id was missing from it. The bookings + * themselves existed and the calendar looked complete; only the + * owner_transactions REVENUE rows for them were never posted. No error, no + * log, nothing in the UI. An owner's P&L was simply short, by an amount that + * grew with the size of the portfolio. + * + * initial-sync is where this bites hardest: it fetches every booking for every + * property in one call, so a 50-property account with two years of history + * clears 1000 on its FIRST sync — the one run whose whole job is to get the + * historical ledger right. + * + * The fix is to chunk the upsert so no single response can be truncated, and + * merge the maps. Chunked at half of max_rows rather than at it: the cap + * applies to the returned representation, and leaving headroom means a future + * change that returns more columns, or a server configured with a lower + * max_rows, does not quietly re-introduce the same silent loss. + * + * A numeric chunk loop, not a per-row loop — structurally exempt from + * unit/guardrails/n-plus-one-loops.test.ts, and one round trip per 500 + * bookings rather than per booking. + */ +export const BOOKING_UPSERT_CHUNK = Math.floor(SUPABASE_MAX_ROWS / 2) + +export async function upsertBookingsReturningIds( + supabase: SupabaseClient, + rows: OwnerRezBookingRow[], + /** Connection label for the thrown message, e.g. `OwnerRez:`. */ + label: string, +): Promise> { + const idByExternalId: Record = {} + + for (let i = 0; i < rows.length; i += BOOKING_UPSERT_CHUNK) { + const chunk = rows.slice(i, i + BOOKING_UPSERT_CHUNK) + + const { data, error } = await supabase + .from('bookings') + .upsert(chunk, { onConflict: 'org_id,external_id,external_source' }) + .select('id, external_id') + + // Thrown, never swallowed: a partial failure here means a partial map, + // which is the same silent revenue gap this helper exists to close. The + // caller's step.run retries the whole batch — the upsert is idempotent on + // (org_id, external_id, external_source), so a replay is safe. + if (error) throw new Error(`[${label}] bookings upsert: ${error.message}`) + + // A short chunk means PostgREST truncated a response we deliberately + // sized to fit. Failing loudly beats returning a map that is quietly + // missing ids, which is precisely the failure mode being fixed. + if ((data ?? []).length < chunk.length) { + throw new Error( + `[${label}] bookings upsert returned ${(data ?? []).length} of ${chunk.length} rows — ` + + 'response truncated, refusing to post revenue against an incomplete id map' + ) + } + + for (const row of data ?? []) { + idByExternalId[row.external_id as string] = row.id as string + } + } + + return idByExternalId +} diff --git a/unit/guardrails/n-plus-one-loops.test.ts b/unit/guardrails/n-plus-one-loops.test.ts index 7acd0506..0195d861 100644 --- a/unit/guardrails/n-plus-one-loops.test.ts +++ b/unit/guardrails/n-plus-one-loops.test.ts @@ -133,10 +133,10 @@ const EXCEPTIONS: Record = { 'Per-section insert (parent-before-child, same reasoning as clone-actions.ts:122) — additionally guarded by a template-signature equality check just above that skips the whole delete-then-recreate rebuild when nothing changed.', 'lib/inngest/functions/cron/guest-pii-retention.ts:137': 'Per-secret delete_vault_secret RPC call — each is a distinct external Vault secret; structurally cannot be batched into one call any more than "one API call per distinct external resource" ever can. Bounded since the 2026-07-30 scalability pass: the loop now iterates one BOOKING_BATCH_SIZE page inside a per-batch step, not an org\'s entire un-anonymized booking history.', - 'lib/inngest/functions/ownerrez/initial-sync.ts:184': + 'lib/inngest/functions/ownerrez/initial-sync.ts:185': 'Per-property conditional field patch (bedrooms/bathrooms/square_footage) — each property\'s patch object contains different values, so it is not a uniform batched update. Pre-fetch of existing rows just above IS already batched via .in(\'external_id\', ids).', 'lib/guidebook/sync.ts:168': - 'Per-property conditional guidebook-config patch — same shape as ownerrez/initial-sync.ts:184 (differing patch per row); the read side just above is already batched via .in(\'property_id\', ids).', + 'Per-property conditional guidebook-config patch — same shape as ownerrez/initial-sync.ts:185 (differing patch per row); the read side just above is already batched via .in(\'property_id\', ids).', 'lib/properties/upsert-normalized.ts:172': 'Per-property conditional cleaning_cost backfill — same differing-patch-per-row shape as the two entries above.', 'lib/inngest/functions/turnover-events.ts:354': diff --git a/unit/inngest/ownerrez-upsert-bookings.test.ts b/unit/inngest/ownerrez-upsert-bookings.test.ts new file mode 100644 index 00000000..c9983e72 --- /dev/null +++ b/unit/inngest/ownerrez-upsert-bookings.test.ts @@ -0,0 +1,119 @@ +import { describe, it, expect, vi } from 'vitest' + +vi.mock('server-only', () => ({})) + +import { + upsertBookingsReturningIds, + BOOKING_UPSERT_CHUNK, +} from '@/lib/inngest/functions/ownerrez/upsert-bookings' +import { SUPABASE_MAX_ROWS } from '@/lib/inngest/paginate' +import { selectOwnerRezBookingsToPostRevenue } from '@/lib/integrations/providers/ownerrez' + +// ============================================================================ +// A single .upsert().select() returns at most max_rows = 1000 rows, with a 200 +// and no truncation signal — the WRITE is uncapped, only the returned +// representation is clipped. Both OwnerRez syncs built their external_id -> id +// map from that response, and selectOwnerRezBookingsToPostRevenue ends with +// +// .filter((b) => !!b.bookingId) +// +// so every booking missing from the clipped response was silently dropped from +// revenue posting. The bookings existed; only the owner's P&L was short. +// +// initial-sync fetches every booking for every property in one call, so a +// portfolio with history crosses 1000 on its FIRST run — the run whose whole +// job is to get the historical ledger right. +// ============================================================================ + +interface UpsertCall { size: number } + +/** + * A client whose .select() truncates at `cap`, exactly as PostgREST does. + * The write always "succeeds" for every row — that asymmetry is the bug. + */ +function makeSupabase(cap = SUPABASE_MAX_ROWS) { + const calls: UpsertCall[] = [] + const from = vi.fn(() => ({ + upsert: (rows: { external_id: string }[]) => ({ + select: () => { + calls.push({ size: rows.length }) + return Promise.resolve({ + data: rows.slice(0, cap).map((r) => ({ id: `id-${r.external_id}`, external_id: r.external_id })), + error: null, + }) + }, + }), + })) + // eslint-disable-next-line @typescript-eslint/no-explicit-any + return { client: { from } as any, calls } +} + +const rows = (n: number) => + // eslint-disable-next-line @typescript-eslint/no-explicit-any + Array.from({ length: n }, (_, i) => ({ external_id: `e${i}` })) as any + +describe('upsertBookingsReturningIds', () => { + it('returns an id for EVERY booking past the 1000-row response cap', async () => { + // 2,500 bookings is an ordinary two-year history for a 50-property account + // — CLAUDE.md's own stated target scale. + const { client } = makeSupabase() + const map = await upsertBookingsReturningIds(client, rows(2_500), 'OwnerRez:test') + + expect(Object.keys(map)).toHaveLength(2_500) + expect(map['e0']).toBe('id-e0') + expect(map['e2499']).toBe('id-e2499') // the row a single upsert lost + }) + + it('never sends a chunk that could itself be truncated', async () => { + const { client, calls } = makeSupabase() + await upsertBookingsReturningIds(client, rows(2_500), 'OwnerRez:test') + + expect(calls.length).toBeGreaterThan(1) + for (const c of calls) expect(c.size).toBeLessThanOrEqual(BOOKING_UPSERT_CHUNK) + // Headroom, not exactly at the cap: a future change returning more columns + // or a server with a lower max_rows must not silently re-break this. + expect(BOOKING_UPSERT_CHUNK).toBeLessThan(SUPABASE_MAX_ROWS) + }) + + it('THROWS on a short response rather than returning a partial map', async () => { + // The defining property. A partial map is indistinguishable from success + // downstream — it just posts less revenue. Failing loudly is the only way + // this surfaces at all. + const { client } = makeSupabase(10) + await expect(upsertBookingsReturningIds(client, rows(600), 'OwnerRez:test')) + .rejects.toThrow(/truncated/) + }) + + it('does one round trip for a small sync, not one per booking', async () => { + const { client, calls } = makeSupabase() + await upsertBookingsReturningIds(client, rows(40), 'OwnerRez:test') + expect(calls).toHaveLength(1) + }) + + it('is a no-op for zero bookings', async () => { + const { client, calls } = makeSupabase() + await expect(upsertBookingsReturningIds(client, rows(0), 'OwnerRez:test')).resolves.toEqual({}) + expect(calls).toHaveLength(0) + }) +}) + +describe('the revenue gap this closes', () => { + it('drops revenue for bookings missing from a truncated id map', async () => { + // Demonstrates the ORIGINAL failure end to end, so the reason for the + // chunking is testable rather than asserted in a comment. + const bookingRows = Array.from({ length: 1_200 }, (_, i) => ({ + external_id: `e${i}`, status: 'confirmed', stay_type: 'guest_stay', + property_id: 'p-1', actual_total_amount: 100, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + })) as any + + const truncatedMap = Object.fromEntries( + bookingRows.slice(0, SUPABASE_MAX_ROWS).map((r: { external_id: string }) => [r.external_id, `id-${r.external_id}`]), + ) + expect(selectOwnerRezBookingsToPostRevenue(bookingRows, truncatedMap)).toHaveLength(1_000) + + const { client } = makeSupabase() + const completeMap = await upsertBookingsReturningIds(client, bookingRows, 'OwnerRez:test') + expect(selectOwnerRezBookingsToPostRevenue(bookingRows, completeMap)).toHaveLength(1_200) + }) +})