`db/migrations/014_listing_lifecycle.sql`

74 lines
-- TK-10155: listing-lifecycle tracking + withdrawal-snapshot preservation.
--
-- Two append-only tables + one "latest state" view. Built on the same discipline as
-- broker_of_record_history (migration 013): NOTHING here mutates existing listing/catalog
-- data — new tables/view only, all CREATE ... IF NOT EXISTS so the migration is idempotent
-- and reversible. No BEGIN/COMMIT here — migrate.ts wraps each file in a transaction.
--
-- GOAL (Steve): track every broker listing across its whole lifecycle — new, active, stale,
-- withdrawn, closed, reappeared — and, critically, PRESERVE A FULL SNAPSHOT of all property
-- info the moment a listing becomes withdrawn (disappears from the source feed), so the data
-- survives after the listing vanishes from the vendor.

-- ─────────────────────────────────────────────────────────────────────────────
-- listing_lifecycle_event — append-only, one row per DETECTED state transition.
-- The recorder inserts a row only when the derived state DIFFERS from the latest
-- observation for that listing (or when there is no prior event), so the table is
-- the change log of every lifecycle update.
-- ─────────────────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS listing_lifecycle_event (
  id           BIGSERIAL PRIMARY KEY,
  listing_id   BIGINT REFERENCES listing(id),      -- nullable: a listing may vanish/be deleted
  source       TEXT,
  source_id    TEXT,
  address_key  TEXT,                                -- normalized "address, city" (shared normalizer)
  state        TEXT NOT NULL,                       -- new | active | stale | withdrawn | closed | reappeared
  prev_state   TEXT,
  price        NUMERIC,
  reason       TEXT,
  observed_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_lle_listing_observed  ON listing_lifecycle_event (listing_id, observed_at DESC);
CREATE INDEX IF NOT EXISTS idx_lle_addrkey_observed  ON listing_lifecycle_event (address_key, observed_at DESC);
CREATE INDEX IF NOT EXISTS idx_lle_state             ON listing_lifecycle_event (state);
CREATE INDEX IF NOT EXISTS idx_lle_source_sid_obs    ON listing_lifecycle_event (source, source_id, observed_at DESC);

-- ─────────────────────────────────────────────────────────────────────────────
-- listing_snapshot — append-only, the FULL property info preserved at a point in time.
-- The snapshot JSONB carries the ENTIRE listing row + joined broker_name + firm_name +
-- region/county + every field available, so a pulled listing's data is still browsable
-- after it vanishes from the vendor feed.
-- ─────────────────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS listing_snapshot (
  id              BIGSERIAL PRIMARY KEY,
  listing_id      BIGINT REFERENCES listing(id),   -- nullable, same reasoning as above
  source          TEXT,
  source_id       TEXT,
  address_key     TEXT,
  snapshot        JSONB NOT NULL,                   -- full listing row + broker/firm/region names
  captured_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
  captured_reason TEXT                              -- withdrawn | closed | stale | new | periodic | manual
);

CREATE INDEX IF NOT EXISTS idx_lsnap_source_sid_cap ON listing_snapshot (source, source_id, captured_at DESC);
CREATE INDEX IF NOT EXISTS idx_lsnap_addrkey_cap    ON listing_snapshot (address_key, captured_at DESC);
CREATE INDEX IF NOT EXISTS idx_lsnap_reason         ON listing_snapshot (captured_reason);

-- ─────────────────────────────────────────────────────────────────────────────
-- listing_current_state — the latest lifecycle state per listing (DISTINCT ON the
-- newest event per source/source_id). LEFT-JOINed by the listings API so every row
-- can carry its lifecycle_state + lifecycle_since.
-- ─────────────────────────────────────────────────────────────────────────────
CREATE OR REPLACE VIEW listing_current_state AS
SELECT DISTINCT ON (source, source_id)
  source,
  source_id,
  listing_id,
  address_key,
  state,
  reason,
  observed_at
FROM listing_lifecycle_event
ORDER BY source, source_id, observed_at DESC;