1import { pgTable, text, integer, index, uniqueIndex, jsonb } from 'drizzle-orm/pg-core';2import { createdAt, ts, ratio } from './_common.js';34/**5 * Immutable raw capture (§108, §142). Rows are never updated except `processed_at`/`process_error`6 * bookkeeping columns. Payload is the parsed source payload (JSON from API, or extracted fields);7 * large HTML snapshots live in RI_DATA_DIR/raw/<connector>/<hash>.html referenced by snapshot_ref.8 */9export const rawRecords = pgTable(10 'raw_records',11 {12 id: text('id').primaryKey(),13 connectorId: text('connector_id').notNull(),14 sourceId: text('source_id').notNull(),15 runId: text('run_id'),16 engine: text('engine').notNull(),17 url: text('url').notNull(),18 externalId: text('external_id'),19 kind: text('kind').notNull(),20 fetchedAt: ts('fetched_at').notNull(),21 contentHash: text('content_hash').notNull(),22 httpStatus: integer('http_status'),23 payload: jsonb('payload').notNull(),24 snapshotRef: text('snapshot_ref'),25 parserVersion: text('parser_version').notNull(),26 connectorVersion: text('connector_version').notNull(),27 processedAt: ts('processed_at'),28 processError: text('process_error'),29 createdAt: createdAt(),30 },31 (t) => [32 uniqueIndex('raw_records_connector_hash_uq').on(t.connectorId, t.contentHash),33 index('raw_records_connector_fetched_idx').on(t.connectorId, t.fetchedAt),34 index('raw_records_unprocessed_idx').on(t.processedAt),35 index('raw_records_external_idx').on(t.connectorId, t.externalId),36 ],37);3839/** Staging between normalisation and canonical tables; keeps the match decision auditable (§112). */40export const normalizedRecords = pgTable(41 'normalized_records',42 {43 id: text('id').primaryKey(),44 rawRecordId: text('raw_record_id').notNull(),45 connectorId: text('connector_id').notNull(),46 sourceId: text('source_id').notNull(),47 kind: text('kind').notNull(),48 payload: jsonb('payload').notNull(),49 assetId: text('asset_id'),50 variantId: text('variant_id'),51 matchMethod: text('match_method'), // identifier | canonical_key | fuzzy | embedding | llm | manual52 matchConfidence: ratio('match_confidence'),53 status: text('status').notNull().default('pending'), // pending | matched | unmatched | rejected | applied54 rejectReason: text('reject_reason'),55 targetId: text('target_id'), // resulting sale/listing/observation id56 createdAt: createdAt(),57 processedAt: ts('processed_at'),58 },59 (t) => [60 index('normalized_records_status_idx').on(t.status, t.createdAt),61 index('normalized_records_asset_idx').on(t.assetId),62 uniqueIndex('normalized_records_raw_uq').on(t.rawRecordId),63 ],64);6566/** Pipeline/event log (§140). Queue itself is pg-boss; this is the durable domain event stream. */67export const events = pgTable(68 'events',69 {70 id: text('id').primaryKey(),71 type: text('type').notNull(), // page_discovered | page_crawled | listing_created | listing_updated | sale_detected | entity_matched | valuation_updated | index_updated72 entityType: text('entity_type'),73 entityId: text('entity_id'),74 payload: jsonb('payload').$type<Record<string, unknown>>().notNull().default({}),75 createdAt: createdAt(),76 },77 (t) => [index('events_type_created_idx').on(t.type, t.createdAt), index('events_entity_idx').on(t.entityType, t.entityId)],78);7980/** Audit log — never silently delete or alter data (§116). */81export const auditLog = pgTable(82 'audit_log',83 {84 id: text('id').primaryKey(),85 entityType: text('entity_type').notNull(),86 entityId: text('entity_id').notNull(),87 action: text('action').notNull(), // flagged | excluded | restored | merged | edited | rejected88 reason: text('reason').notNull(),89 actor: text('actor').notNull().default('system'),90 details: jsonb('details').$type<Record<string, unknown>>().notNull().default({}),91 createdAt: createdAt(),92 },93 (t) => [index('audit_entity_idx').on(t.entityType, t.entityId)],94);95