Claude/hostile code audit rbz3f8

docs/Integrations/ownerrez/api-markdown.md

@@ -14,6 +14,49 @@
 - Errors return JSON with a `messages` array of human-readable strings. Status codes: 400 = bad request / validation, 401 = authentication, 403 = permission, 404 = not found.
 - Prefer the current API (v2). Legacy versions remain documented below but should not be used for new integrations.
 
+## Verified contract details (2026-08-13)
+
+Pulled from `https://api.ownerrez.com/openapi/v2.json` because the orientation
+notes above have no per-endpoint parameter list, and guessing filled the gap
+wrongly once already. Re-verify against the spec before trusting any of this.
+
+### Paging — every list endpoint
+
+The response wrapper (`PageableEnumerableOf*`, used by all 22 list endpoints) is:
+
+```
+{ items: [...], limit: 100, offset: 0, next_page_url: "https://..." | null }
+```
+
+- `next_page_url` is the continuation mechanism. **Null means the collection is
+  finished** — including on a full page. Absent is a different answer from null.
+- There is **no `next_page_token` and no `total_count`.** `lib/integrations/types.ts`
+  declared both; neither string occurs anywhere in the spec. The pager read
+  `next_page_token`, got `undefined`, and stopped after one page — and since it
+  also sent no `limit`, that one page was OwnerRez's default of 20. One
+  production org's first sync imported exactly 20 bookings as a result.
+- `limit` (default 20, max 100) and `offset` are **declared on zero operations**
+  in the spec even though the response echoes both back. They work, but do not
+  assume a requested `limit` was honored — prefer `next_page_url`, and compare
+  short-page checks against the server-reported `page.limit`, not the one you asked for.
+
+### GET /v2/bookings query parameters
+
+| Param | Meaning |
+|---|---|
+| `property_ids` | comma-separated integers |
+| `from` | bookings that **depart on or after** this date (property timezone) |
+| `to` | bookings that **arrive on or before** this date (property timezone) |
+| `since_utc` | created or changed since (UTC) — a **modification-time cursor, not a stay-date filter** |
+| `status` | `Active` \| `Pending` \| `Canceled` |
+| `include_guest`, `include_charges`, `include_tags`, `include_fields`, `include_door_codes`, `include_cancellation_policy`, `include_agreements` | booleans |
+
+`from`/`to` together are an interval **overlap** filter, not containment: a stay
+straddling a window boundary is returned by both adjacent windows. That is what
+makes a progressive historical backfill possible (walk `to` backwards in fixed
+steps); the boundary duplicates are harmless because bookings upsert on the
+OwnerRez booking id.
+
 ## Documentation
 
 - [OpenAPI 3.0 spec](/openapi/v2.json): the full machine-readable contract for v2.

lib/inngest/functions/ownerrez/incremental-sync.ts

@@ -53,6 +53,11 @@ import type { GetStepTools }            from 'inngest'
 import { createServiceClient }          from '@/lib/supabase/server'
 import { fetchTurnoverCreatedEvents } from '@/lib/inngest/turnover-created-events'
 import { OwnerRezApiClient }  from '@/lib/integrations/providers/ownerrez-api'
+import {
+  planBackfillWindow,
+  advanceBackfill,
+  readBackfillState,
+} from '@/lib/integrations/providers/ownerrez-backfill'
 import { getRedis, upstashConfigured } from '@/lib/redis'
 import { RateLimitError, TokenRevokedError, translateSyncError } from '@/lib/integrations/types'
 import { logAuditEvent }                from '@/lib/audit'
@@ -996,10 +1001,172 @@ export const ownerRezConnectionSync = inngest.createFunction(
     await runPostSyncFanOut(step, { orgId, userId, affectedIds, logger })
 
     const synced = Boolean(syncResult && 'affectedPropertyIds' in syncResult)
-    return { connectionId, synced }
+
+    // ── Historical backfill: one older stay-date window per run ──────────────
+    //
+    // The initial sync deliberately takes only a recent window so a new PM is
+    // not kept waiting on years of history (see ownerrez-backfill.ts). This
+    // walks the rest backwards, one window per incremental run, until it
+    // reaches the horizon and stops for good.
+    //
+    // Gated on `synced`: if the live sync just skipped or failed, the
+    // connection is inactive, rate limited or degraded, and adding a second
+    // multi-page fetch on top of that helps nobody. It simply waits an hour.
+    const backfill = await runBackfillPhase(step, { connectionId, userId, orgId, synced, logger })
+
+    return { connectionId, synced, backfill }
   }
 )
 
+interface BackfillOutcome {
+  outcome:               string
+  bookingsToPostRevenue: SyncSuccess['bookingsToPostRevenue']
+}
+
+/**
+ * The backfill phase: claim one historical window, then post revenue for
+ * whatever it imported. Structured like runPostSyncFanOut — step tooling stays
+ * at the function's top level, never inside a step.run callback.
+ *
+ * Revenue for backfilled stays IS posted; historical owner P&L is the reason to
+ * import them at all. Turnovers deliberately are NOT — those are units of WORK,
+ * and generateTurnoversForProperty's own history floor is what stops a two-year
+ * import from manufacturing retroactive cleaning jobs. Backfilled properties
+ * therefore never enter runPostSyncFanOut.
+ */
+async function runBackfillPhase(
+  step: GetStepTools<typeof inngest>,
+  ctx:  { connectionId: string; userId: string; orgId: string | null
+          synced: boolean; logger: SyncLogger }
+): Promise<string> {
+  const backfill = await step.run('backfill-history', () => runHistoricalBackfill(ctx))
+
+  // Optional-chained for the same reason `syncResult &&` is, at the call site:
+  // a step's result is only as present as the step that produced it.
+  const revenue = backfill?.bookingsToPostRevenue ?? []
+
+  if (revenue.length > 0 && ctx.orgId) {
+    await step.sendEvent(
+      'post-backfill-revenue',
+      revenue.map((b) => ({
+        name: 'booking/confirmed' as const,
+        data: {
+          booking_id:          b.bookingId,
+          property_id:         b.propertyId,
+          org_id:              ctx.orgId as string,
+          source:              'ownerrez' as const,
+          actual_total_amount: b.actualTotalAmount,
+        },
+      }))
+    )
+  }
+
+  return backfill?.outcome ?? 'skipped'
+}
+
+const NO_BACKFILL = (outcome: string): BackfillOutcome =>
+  ({ outcome, bookingsToPostRevenue: [] })
+
+/**
+ * Claims and persists ONE historical stay-date window for a connection.
+ *
+ * Never throws. The backfill is strictly best-effort catch-up work running
+ * behind a live sync that already succeeded — letting a rate limit or a bad
+ * page here fail the whole run would turn "we did not fetch some 2024 bookings
+ * this hour" into "this connection's hourly sync is red", and Inngest would
+ * retry the entire function including the live sync that was already fine.
+ * Progress only advances on success, so a swallowed failure simply retries the
+ * same window next hour.
+ */
+async function runHistoricalBackfill(params: {
+  connectionId: string
+  userId:       string
+  orgId:        string | null
+  synced:       boolean
+  logger:       SyncLogger
+}): Promise<BackfillOutcome> {
+  const { connectionId, userId, orgId, synced, logger } = params
+
+  if (!synced || !orgId) return NO_BACKFILL('skipped_not_synced')
+
+  try {
+    const supabase = createServiceClient({ system: 'inngest:ownerrez-backfill' })
+
+    const connRes = await supabase
+      .from('integration_connections')
+      .select('id, user_id, org_id, external_user_id, metadata, status')
+      .eq('id', connectionId)
+      .maybeSingle()
+
+    const conn = unwrap(connRes, {
+      site:  'inngest.ownerrez-connection-sync.backfill-reload',
+      orgId: orgId,
+    })
+
+    // org_id first so the optional chain covers the null-connection case too:
+    // a truthy conn?.org_id already proves conn itself is non-null, which is
+    // what lets conn.status be read plainly on the next clause.
+    if (!conn?.org_id || conn.status !== 'active') {
+      return NO_BACKFILL('skipped_inactive')
+    }
+
+    const window = planBackfillWindow(readBackfillState(conn.metadata), new Date())
+    if (!window) return NO_BACKFILL('complete')
+
+    // property_ids is required alongside the date bounds for the same reason
+    // the live sync needs it: OwnerRez wants at least one scoping parameter,
+    // and this keeps the window to properties FieldStay actually knows about.
+    const propertyIds = await loadConnectedPropertyIds(supabase, conn.org_id)
+    if (!propertyIds.length) return NO_BACKFILL('skipped_no_properties')
+
+    const bookings = await new OwnerRezApiClient(userId).getBookings({
+      propertyIds,
+      from:         window.from,
+      to:           window.to,
+      includeGuest: true,
+    })
+
+    const activeConn: ActiveConnection = { ...conn, org_id: conn.org_id }
+    const persisted = await persistBookings(supabase, activeConn, bookings, logger)
+
+    // Property lookup failed. Do NOT advance the cursor — the same reasoning as
+    // the live sync's bail-out, plus advancing here would skip this window
+    // permanently since nothing revisits it.
+    if (!persisted) return NO_BACKFILL('persist_failed')
+
+    const advanced = advanceBackfill(window, new Date())
+    await mergeIntegrationConnectionMetadata({
+      userId,
+      providerId: PROVIDER,
+      patch: {
+        backfill_oldest_covered: advanced.oldestCovered,
+        backfill_complete:       advanced.complete,
+      },
+    })
+
+    logger.info(
+      `[OwnerRez:${userId}] backfilled ${window.from}..${window.to} — ` +
+      `${bookings.length} booking(s)${advanced.complete ? ', walk complete' : ''}`
+    )
+
+    return {
+      outcome:               advanced.complete ? 'window_done_complete' : 'window_done',
+      bookingsToPostRevenue: persisted.bookingsToPostRevenue,
+    }
+  } catch (err) {
+    logger.warn(
+      `[OwnerRez:${userId}] historical backfill failed, will retry next run: ` +
+      (err instanceof Error ? err.message : String(err))
+    )
+    reportError(err, {
+      site:  'inngest.ownerrez-connection-sync.backfill-history',
+      orgId: orgId ?? undefined,
+      extra: { connection_id: connectionId },
+    })
+    return NO_BACKFILL('failed')
+  }
+}
+
 /**
  * Fire a PM notification about a broken connection — throttled to once per
  * 4 hours per connection via an org_milestones timestamp. Shared by the

lib/inngest/functions/ownerrez/initial-sync.ts

@@ -23,6 +23,7 @@ import {
   selectOwnerRezBookingsToPostRevenue,
 } from '@/lib/integrations/providers/ownerrez'
 import { upsertBookingsReturningIds } from './upsert-bookings'
+import { initialHistoryFrom } from '@/lib/integrations/providers/ownerrez-backfill'
 import { logAuditEvent }        from '@/lib/audit'
 import {
   applyMasterChecklistToProperty,
@@ -52,6 +53,13 @@ import { fetchAllRows } from '@/lib/inngest/paginate'
 import { tryUnwrap, unwrapList } from '@/lib/supabase/unwrap'
 const PROVIDER = 'ownerrez'
 
+/**
+ * Bookings per booking/confirmed sendEvent step. Sized to keep one step's
+ * payload comfortably small while making the step count independent of
+ * portfolio size — see the batching note at the call site.
+ */
+const REVENUE_EVENT_CHUNK = 200
+
 async function writeSyncCount(
   user_id: string,
   field: 'properties_found' | 'bookings_found',
@@ -94,6 +102,27 @@ export const ownerRezInitialSync = inngest.createFunction(
     id:      'ownerrez-initial-sync',
     name:    'OwnerRez Initial Sync',
     retries: 3,
+    // Two caps, and they answer different questions.
+    //
+    // The unkeyed one is capacity: this is the heaviest OwnerRez consumer we
+    // have — it paginates every property, listing and booking for a brand-new
+    // account — and the 300-request/5-minute OwnerRez budget is shared by every
+    // tenant on the same deployment IP. Left uncapped, several signups landing
+    // together would spend that budget on each other and take the incremental
+    // syncs down with them. 3 matches ownerRezConnectionSync's cap for the
+    // same reason.
+    //
+    // The keyed one is correctness: never two initial syncs for one connection
+    // at once. This step chain seeds checklists, generates turnovers and seeds
+    // assets from amenities, and not all of that is safe to interleave with
+    // itself. A reconnect, a double-clicked Connect button, or a re-fired
+    // event would otherwise race. Keyed concurrency QUEUES the second run
+    // rather than dropping it, which is what a genuine reconnect wants —
+    // `idempotency` would silently discard it instead.
+    concurrency: [
+      { limit: 3 },
+      { limit: 1, key: 'event.data.user_id' },
+    ],
   },
   { event: 'integration/ownerrez.connected' as const },
   async ({ event, step, logger }) => {
@@ -515,6 +544,7 @@ export const ownerRezInitialSync = inngest.createFunction(
             cursor: new Date().toISOString(), count: 0,
             affectedPropertyIds: [] as string[],
             bookingsToPostRevenue: [] as { bookingId: string; propertyId: string; actualTotalAmount: number | null }[],
+            historyFrom: initialHistoryFrom(new Date()),
           }
         }
 
@@ -525,8 +555,28 @@ export const ownerRezInitialSync = inngest.createFunction(
 
         let bookings: OwnerRezBooking[]
 
+        // BOUNDED to a recent window, plus everything upcoming.
+        //
+        // This call used to pass no date bounds at all, i.e. "every booking
+        // this account has ever had". That was harmless only because the pager
+        // was broken and stopped at 20 records; with pagination fixed it became
+        // a request to page through a portfolio's entire history — thousands of
+        // rows against a request budget shared by every tenant — before the PM
+        // sees anything. Older history is walked backwards afterwards, one
+        // window per incremental sync (ownerrez-backfill.ts).
+        //
+        // `from` with no `to` is the important shape: `from` means "departs on
+        // or after", so every FUTURE booking is still included. Bounding the
+        // upper end would drop upcoming stays, which are the whole point of a
+        // first sync.
+        const historyFrom = initialHistoryFrom(new Date(fetchStartedAt))
+
         try {
-          bookings = await client.getBookings({ propertyIds: fetchPropsResult.ids, includeGuest: true })
+          bookings = await client.getBookings({
+            propertyIds:  fetchPropsResult.ids,
+            from:         historyFrom,
+            includeGuest: true,
+          })
         } catch (err) {
           if (err instanceof RateLimitError) {
             throw err
@@ -601,7 +651,7 @@ export const ownerRezInitialSync = inngest.createFunction(
           reportError(countErr, { site: 'inngest.ownerrez-initial-sync.fetch-bookings' })
         }
 
-        return { cursor: fetchStartedAt, count: bookings.length, affectedPropertyIds, bookingsToPostRevenue }  // MEDIUM-3: pre-fetch timestamp
+        return { cursor: fetchStartedAt, count: bookings.length, affectedPropertyIds, bookingsToPostRevenue, historyFrom }  // MEDIUM-3: pre-fetch timestamp
       })
 
       // ── Post booking revenue for newly-confirmed guest-stay bookings ───────
@@ -612,17 +662,30 @@ export const ownerRezInitialSync = inngest.createFunction(
       // 2026-07-15; booking-events.ts's handleBookingConfirmed still falls
       // back to the avg_nightly_rate estimate whenever this is null (e.g. a
       // booking whose charges genuinely didn't resolve to a positive total).
-      for (const b of fetchBookingsResult.bookingsToPostRevenue) {
-        await step.sendEvent(`post-booking-revenue-${b.bookingId}`, {
-          name: 'booking/confirmed' as const,
-          data: {
-            booking_id:          b.bookingId,
-            property_id:         b.propertyId,
-            org_id,
-            source:              'ownerrez' as const,
-            actual_total_amount: b.actualTotalAmount,
-          },
-        })
+      // BATCHED, one step per chunk rather than one step per booking.
+      //
+      // This was `for (const b of ...) await step.sendEvent(...)`, which makes
+      // a distinct Inngest step — with its own memoized state carried for the
+      // rest of the run — for every single booking. At 20 bookings that was
+      // invisible; bounded to a 90-day window it is hundreds, and unbounded it
+      // would have been thousands once pagination started working. step.sendEvent
+      // takes an array, so a chunk costs one step regardless of its size.
+      const revenueEvents = fetchBookingsResult.bookingsToPostRevenue.map((b) => ({
+        name: 'booking/confirmed' as const,
+        data: {
+          booking_id:          b.bookingId,
+          property_id:         b.propertyId,
+          org_id,
+          source:              'ownerrez' as const,
+          actual_total_amount: b.actualTotalAmount,
+        },
+      }))
+
+      for (let i = 0; i < revenueEvents.length; i += REVENUE_EVENT_CHUNK) {
+        const chunk = revenueEvents.slice(i, i + REVENUE_EVENT_CHUNK)
+        // Chunk index, not booking id: the step id must stay stable across
+        // retries, and it does because the source list is memoized upstream.
+        await step.sendEvent(`post-booking-revenue-${i / REVENUE_EVENT_CHUNK}`, chunk)
       }
 
       // ── Step 3: Update sync metadata ────────────────────────────────────────
@@ -639,6 +702,13 @@ export const ownerRezInitialSync = inngest.createFunction(
               last_sync_status: 'success',
               last_sync_error:  null,
               last_sync_count:  fetchBookingsResult.count,
+              // Seeds the historical backfill walk at the exact lower edge of
+              // the window this run actually fetched — NOT at "90 days before
+              // whenever the first backfill happens to run". Recomputing it
+              // later would open a gap the width of the delay between the two,
+              // and nothing ever revisits a skipped window.
+              backfill_oldest_covered: fetchBookingsResult.historyFrom,
+              backfill_complete:       false,
             },
           })
         } catch (err) {