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 /** position of this record within the output of normalize(raw) — one raw row may yield several records */50 seq: integer('seq').notNull().default(0),51 assetId: text('asset_id'),52 variantId: text('variant_id'),53 matchMethod: text('match_method'), // identifier | canonical_key | fuzzy | embedding | llm | manual54 matchConfidence: ratio('match_confidence'),55 status: text('status').notNull().default('pending'), // pending | matched | unmatched | rejected | applied56 rejectReason: text('reject_reason'),57 targetId: text('target_id'), // resulting sale/listing/observation id58 createdAt: createdAt(),59 processedAt: ts('processed_at'),60 },61 (t) => [62 index('normalized_records_status_idx').on(t.status, t.createdAt),63 index('normalized_records_asset_idx').on(t.assetId),64 uniqueIndex('normalized_records_raw_uq').on(t.rawRecordId, t.kind, t.seq),65 ],66);6768/** Pipeline/event log (§140). Queue itself is pg-boss; this is the durable domain event stream. */69export const events = pgTable(70 'events',71 {72 id: text('id').primaryKey(),73 type: text('type').notNull(), // page_discovered | page_crawled | listing_created | listing_updated | sale_detected | entity_matched | valuation_updated | index_updated74 entityType: text('entity_type'),75 entityId: text('entity_id'),76 payload: jsonb('payload').$type<Record<string, unknown>>().notNull().default({}),77 createdAt: createdAt(),78 },79 (t) => [index('events_type_created_idx').on(t.type, t.createdAt), index('events_entity_idx').on(t.entityType, t.entityId)],80);8182/** Audit log — never silently delete or alter data (§116). */83export const auditLog = pgTable(84 'audit_log',85 {86 id: text('id').primaryKey(),87 entityType: text('entity_type').notNull(),88 entityId: text('entity_id').notNull(),89 action: text('action').notNull(), // flagged | excluded | restored | merged | edited | rejected90 reason: text('reason').notNull(),91 actor: text('actor').notNull().default('system'),92 details: jsonb('details').$type<Record<string, unknown>>().notNull().default({}),93 createdAt: createdAt(),94 },95 (t) => [index('audit_entity_idx').on(t.entityType, t.entityId)],96);97