-- R553 Phase 52 — Event ledger / CDC substrate.
-- Single append-only events stream. Every meaningful state change writes here.
-- Consumers: workflow_runner (subscribes by event_type), reflexion, timeline UI,
-- per-entity history, replay tooling, anomaly detection.
-- Per [[project_high_leverage_moves]] move #2 (Mike's strategic doc).

CREATE TABLE IF NOT EXISTS events (
  event_id        TEXT PRIMARY KEY,            -- ULID (client-generated for ordering)
  event_type      TEXT NOT NULL,               -- 'customer.updated', 'price.changed',
                                               -- 'ar.bucket_moved', 'email.parsed',
                                               -- 'hitl.approved', 'hitl.rejected',
                                               -- 'sync.tier_completed', 'workflow.run_completed', ...
  entity_type     TEXT,                        -- 'customer' | 'item' | 'vendor' | 'bid' | 'so' | 'po' | 'workflow_run' | 'spec' | ...
  entity_id       TEXT,                        -- NS id, slug, or composite key
  payload_json    TEXT NOT NULL,               -- domain-specific JSON; consumers schema-validate
  caused_by       TEXT,                        -- 'sync:tier=hot' | 'chat:role=admin' | 'workflow:bid_price_update' | 'user:mikelevine' | 'cron:nightly_health'
  workflow_run_id TEXT,                        -- FK-ish to workflow_run_log.run_id when part of a workflow
  source_system   TEXT NOT NULL DEFAULT 'platform', -- 'netsuite' | 'platform' | 'email' | 'dropbox' | 'manual'
  occurred_at     TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
  recorded_at     TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
                                               -- recorded_at > occurred_at when backfilled
  sequence_no     INTEGER                      -- optional per-entity monotonic counter, set by producer
);
CREATE INDEX IF NOT EXISTS idx_events_entity ON events(entity_type, entity_id, occurred_at DESC);
CREATE INDEX IF NOT EXISTS idx_events_type ON events(event_type, occurred_at DESC);
CREATE INDEX IF NOT EXISTS idx_events_workflow ON events(workflow_run_id);
CREATE INDEX IF NOT EXISTS idx_events_source ON events(source_system, occurred_at DESC);
CREATE INDEX IF NOT EXISTS idx_events_occurred ON events(occurred_at DESC);

-- Subscriber config: which workflow_types fire on which event_types.
-- The workflow_runner cron drains pending events and matches against this.
CREATE TABLE IF NOT EXISTS event_subscriptions (
  subscription_id INTEGER PRIMARY KEY AUTOINCREMENT,
  event_type_glob TEXT NOT NULL,               -- 'price.*' | 'customer.updated' | 'ar.bucket_moved'
  workflow_type   TEXT NOT NULL,               -- workflow_definitions.workflow_type
  input_mapper    TEXT,                        -- jsonpath-like recipe to translate payload→inputs ({"customer_id":"$.entity_id"})
  filter_expr     TEXT,                        -- optional SQL-WHERE-style filter on the event row
  enabled         INTEGER NOT NULL DEFAULT 1,
  created_at      TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
  notes           TEXT
);
CREATE INDEX IF NOT EXISTS idx_event_subs_glob ON event_subscriptions(event_type_glob) WHERE enabled = 1;

-- Drain cursor — last event processed by the subscription engine.
CREATE TABLE IF NOT EXISTS event_drain_state (
  drainer_id      TEXT PRIMARY KEY,            -- 'workflow_runner' | 'timeline_indexer' | 'anomaly_detector'
  last_event_id   TEXT,
  last_drained_at TEXT,
  events_processed_total INTEGER NOT NULL DEFAULT 0
);

INSERT OR IGNORE INTO migration_history (migration_id, description) VALUES
  ('123_events_ledger', 'R553 Phase 52 — events ledger + subscriptions + drain state (Mike high-leverage #2)');
