// Server-only. The shared ingest core for Door epic Phase 4. Both transports // (the poll cron and the inbound webhook) parse/normalize events into // DoorIngestEvent[] and hand them here to resolve each card to a member and // idempotently insert audit rows. door_access_log has no client write policy, // so the service (admin) client is the only writer, exactly as for the // outbound actions. import type { createAdminClient } from '@/lib/supabase/admin' import type { TablesInsert } from '@/types/database' import { callDoor } from './executor' import { resolveDoorSecret } from './secret' import { encodeHeatSyncControl, applyTemplate, redactDoorSecrets } from '@/lib/door-logic' import { parseHeatSyncLog, cardMatchesEvent, type DoorIngestEvent, } from '@/lib/door-log-logic' import { captureException } from '@/lib/observability/capture' type Admin = ReturnType // The subset of door_connections the ingest paths read (service client). export type IngestConn = { id: string space_id: string adapter: string base_url: string pinned_host: string auth_param: string | null secret_ref: string | null verbs: Record | null is_enabled: boolean } function isUniqueViolation(message: string): boolean { return /duplicate key value|already exists|unique constraint/i.test(message) } function actionFor(result: DoorIngestEvent['result']): string { if (result === 'granted') return 'entry' if (result === 'denied') return 'entry_denied' return 'inbound' } // Resolve each event's card to a member and insert one redacted, deduped // door_access_log row per new event. Returns counts for the caller's response. // Dedup is two-layered: a pre-filter against existing (connection_id, // dedupe_key) cuts the common re-poll case to a single SELECT and zero writes, // and the partial-unique index makes a concurrent insert race a no-op (the // unique violation is swallowed per row). export async function ingestEvents( admin: Admin, spaceId: string, connectionId: string, events: DoorIngestEvent[], ): Promise<{ inserted: number; resolved: number; skipped: number }> { if (events.length === 0) return { inserted: 0, resolved: 0, skipped: 0 } const { data: cards } = await admin .from('member_cards') .select('id, card_uid, member_id') .eq('space_id', spaceId) .eq('is_active', true) const cardList = (cards ?? []) as { id: string; card_uid: string; member_id: string | null }[] const keys = events.map(e => e.dedupeKey) const { data: existing } = await admin .from('door_access_log') .select('dedupe_key') .eq('connection_id', connectionId) .in('dedupe_key', keys) const have = new Set((existing ?? []).map(r => r.dedupe_key as string)) let inserted = 0 let resolved = 0 let skipped = 0 for (const ev of events) { if (have.has(ev.dedupeKey)) { skipped++ continue } const card = cardList.find(c => cardMatchesEvent(c.card_uid, ev)) const targetMemberId = card?.member_id ?? null if (card) resolved++ const detail = redactDoorSecrets( `${ev.detail} (${card ? 'member matched' : 'no card match'})`.slice(0, 1000), null, null, ) const row: TablesInsert<'door_access_log'> = { space_id: spaceId, connection_id: connectionId, actor_member_id: null, target_member_id: targetMemberId, action: actionFor(ev.result), success: ev.result === 'granted', detail, dedupe_key: ev.dedupeKey, } // A webhook caller supplies occurred_at; accept only a parseable, // not-in-the-future time (60s skew) so a relay cannot pin spoofed rows to // the top of the operator's occurred_at-ordered log. Otherwise the row // defaults to ingest time. if (ev.occurredAt) { const t = Date.parse(ev.occurredAt) if (!Number.isNaN(t) && t <= Date.now() + 60_000) { row.occurred_at = new Date(t).toISOString() } } const { error } = await admin.from('door_access_log').insert(row) if (error) { // A concurrent ingest already inserted this exact event: not an error. if (!isUniqueViolation(error.message)) { // Anything else (e.g. a transient DB error) is surfaced via console so // the row is not silently lost; the next poll re-attempts it. console.error('[door-ingest] insert failed:', error.message) captureException(error, { surface: 'door/ingest', tags: { stage: 'insert' } }) } continue } inserted++ } return { inserted, resolved, skipped } } // Poll a single native-HeatSync connection's `?z` log and return parsed events. // Only native_heatsync is parsed (its wire format is characterized); generic // controllers ingest via the webhook. The full ring-buffer body is requested // (fullBody) so the parser sees every slot. export async function pollConnectionLog( admin: Admin, conn: IngestConn, ): Promise<{ ok: boolean; events: DoorIngestEvent[]; detail: string }> { const password = await resolveDoorSecret(admin, conn.space_id, conn.secret_ref) let query: string if (conn.adapter === 'native_heatsync') { const enc = encodeHeatSyncControl('log', password ?? '') if (!enc.ok) return { ok: false, events: [], detail: `encode failed: ${enc.reason}` } query = enc.query } else { return { ok: false, events: [], detail: 'poll supports native_heatsync only; use the webhook' } } const result = await callDoor({ url: conn.base_url + query, pinnedHost: conn.pinned_host, password, authParam: conn.auth_param, fullBody: true, }) if (!result.ok) { return { ok: false, events: [], detail: redactDoorSecrets(`poll failed: ${result.reason}`, password, conn.auth_param) } } const events = parseHeatSyncLog(result.snippet) return { ok: true, events, detail: `polled ${events.length} event(s)` } }