← back to Commercialrealestate

scripts/db/migrations/20260731_broker_firm_dedup.sql

157 lines

-- ============================================================================
-- CRCP data cleanup — GATED / DRAFT (do NOT run without Steve's approval)
-- ----------------------------------------------------------------------------
-- Two $0, DB-only fixes surfaced during the mind-map build:
--   (1) FIRM CASE-MERGE   — collapse firms differing only by case (Compass/COMPASS,
--       eXp / Century-21 variants...): 47 groups, ~52 firm rows removed.
--   (2) BROKER DEDUP      — merge duplicate broker rows keyed on (normalized name,
--       TARGET firm). This covers BOTH the null-firm dups (the UNIQUE(name,firm_id)
--       NULL hole) AND the duplicates the firm-merge itself creates (e.g. a broker
--       who exists under two case-variants of the same firm) — which is what the
--       first dry-run caught as a UNIQUE(name,firm_id) violation.
--
-- ORDER MATTERS: brokers are deduped on their POST-merge target firm BEFORE the
-- firm_id repoint lands, so the repoint can never collide.
--
-- SAFETY:
--   * Fully transactional (BEGIN/COMMIT) — atomic, all-or-nothing.
--   * Backs up every mutated table to *_bak_20260731 (condo/sfr: affected rows only).
--   * Broker child rows (7 FK tables, all ON DELETE CASCADE) are MOVED to the
--     survivor conflict-safe (NOT EXISTS on each composite PK); the cascade drops
--     only the leftover colliding rows when the dup broker is deleted.
--   * Firm-referencing tables (broker, broker_firm_history, condo, sfr — plain FKs)
--     are repointed BEFORE the orphan firm is deleted.
--   * broker_firm_history current rows are rebuilt from the cleaned broker table.
--
-- HOW TO RUN (on approval):
--   DRY RUN : sed 's/^COMMIT;$/ROLLBACK;/' <this> | psql -v ON_ERROR_STOP=1 -h /tmp -d cre
--             inspect the report row, confirm no error, nothing persists.
--   REAL    : psql -v ON_ERROR_STOP=1 -h /tmp -d cre -f <this>   (then rebuild snapshot)
-- ============================================================================

BEGIN;

-- ── 0. Backups (rollback source) ────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS broker_bak_20260731               AS SELECT * FROM broker;
CREATE TABLE IF NOT EXISTS firm_bak_20260731                 AS SELECT * FROM firm;
CREATE TABLE IF NOT EXISTS broker_listing_bak_20260731       AS SELECT * FROM broker_listing;
CREATE TABLE IF NOT EXISTS broker_field_source_bak_20260731  AS SELECT * FROM broker_field_source;
CREATE TABLE IF NOT EXISTS broker_condo_bak_20260731         AS SELECT * FROM broker_condo;
CREATE TABLE IF NOT EXISTS broker_closed_listing_bak_20260731 AS SELECT * FROM broker_closed_listing;
CREATE TABLE IF NOT EXISTS broker_other_listing_bak_20260731 AS SELECT * FROM broker_other_listing;
CREATE TABLE IF NOT EXISTS broker_sfr_bak_20260731           AS SELECT * FROM broker_sfr;
CREATE TABLE IF NOT EXISTS broker_firm_history_bak_20260731  AS SELECT * FROM broker_firm_history;

-- ── 1. Canonical firm per lowercased name (case-merge groups only) ──────────
--   keep_id   = firm with the MOST brokers (tiebreak: fewest UPPERCASE chars, then id)
--   keep_name = the least-shouty variant (fewest uppercase chars)
CREATE TEMP TABLE firm_merge ON COMMIT DROP AS
WITH grp AS (
  SELECT f.id, f.name, lower(trim(f.name)) AS k,
         (SELECT count(*) FROM broker b WHERE b.firm_id = f.id) AS nb,
         length(regexp_replace(f.name, '[^A-Z]', '', 'g')) AS upper_ct  -- count of UPPERCASE letters (fewest = least shouty)
  FROM firm f),
dupk AS (SELECT k FROM grp GROUP BY k HAVING count(*) > 1),
ranked AS (
  SELECT g.*,
         first_value(g.id)   OVER (PARTITION BY g.k ORDER BY g.nb DESC, g.upper_ct ASC, g.id ASC) AS keep_id,
         first_value(g.name) OVER (PARTITION BY g.k ORDER BY g.upper_ct ASC, g.nb DESC, g.id ASC) AS keep_name
  FROM grp g JOIN dupk d ON d.k = g.k)
SELECT id AS old_id, keep_id, keep_name FROM ranked;

CREATE TABLE IF NOT EXISTS condo_bak_20260731 AS
  SELECT * FROM condo WHERE firm_id IN (SELECT old_id FROM firm_merge);
CREATE TABLE IF NOT EXISTS sfr_bak_20260731 AS
  SELECT * FROM sfr   WHERE firm_id IN (SELECT old_id FROM firm_merge);

-- ── 2. Each broker's TARGET firm after the case-merge ───────────────────────
CREATE TEMP TABLE broker_target ON COMMIT DROP AS
SELECT b.id AS broker_id, lower(trim(b.name)) AS namek,
       COALESCE(fm.keep_id, b.firm_id) AS target_firm
FROM broker b LEFT JOIN firm_merge fm ON fm.old_id = b.firm_id;

-- ── 3. Broker merge groups: same normalized name + same TARGET firm ─────────
--   (covers null-firm dups AND firm-merge-induced dups; survivor = min id)
CREATE TEMP TABLE broker_merge ON COMMIT DROP AS
WITH g AS (SELECT broker_id, namek, COALESCE(target_firm, -1) AS tf FROM broker_target),
     dup AS (SELECT namek, tf FROM g GROUP BY namek, tf HAVING count(*) > 1)
SELECT g.broker_id AS old_id,
       min(g.broker_id) OVER (PARTITION BY g.namek, g.tf) AS keep_id
FROM g JOIN dup d ON d.namek = g.namek AND d.tf = g.tf;

-- ── 4. Move dup brokers' child rows to the survivor (conflict-safe) ─────────
-- For composite-PK child tables, move EXACTLY ONE row per (survivor, key) — both
-- against the survivor's existing rows AND among the dup movers themselves (two
-- dups sharing a key would otherwise both move and collide). DISTINCT ON picks a
-- single winner by ctid; the losing duplicates are cascade-deleted with the dup.
WITH cand AS (
  SELECT DISTINCT ON (m.keep_id, t.listing_id) t.ctid AS cid, m.keep_id
  FROM broker_listing t JOIN broker_merge m ON t.broker_id = m.old_id AND m.old_id <> m.keep_id
  WHERE NOT EXISTS (SELECT 1 FROM broker_listing k WHERE k.broker_id = m.keep_id AND k.listing_id = t.listing_id)
  ORDER BY m.keep_id, t.listing_id, t.broker_id)
UPDATE broker_listing t SET broker_id = cand.keep_id FROM cand WHERE t.ctid = cand.cid;

WITH cand AS (
  SELECT DISTINCT ON (m.keep_id, t.field) t.ctid AS cid, m.keep_id
  FROM broker_field_source t JOIN broker_merge m ON t.broker_id = m.old_id AND m.old_id <> m.keep_id
  WHERE NOT EXISTS (SELECT 1 FROM broker_field_source k WHERE k.broker_id = m.keep_id AND k.field = t.field)
  ORDER BY m.keep_id, t.field, t.broker_id)
UPDATE broker_field_source t SET broker_id = cand.keep_id FROM cand WHERE t.ctid = cand.cid;

WITH cand AS (
  SELECT DISTINCT ON (m.keep_id, t.condo_id) t.ctid AS cid, m.keep_id
  FROM broker_condo t JOIN broker_merge m ON t.broker_id = m.old_id AND m.old_id <> m.keep_id
  WHERE NOT EXISTS (SELECT 1 FROM broker_condo k WHERE k.broker_id = m.keep_id AND k.condo_id = t.condo_id)
  ORDER BY m.keep_id, t.condo_id, t.broker_id)
UPDATE broker_condo t SET broker_id = cand.keep_id FROM cand WHERE t.ctid = cand.cid;

WITH cand AS (
  SELECT DISTINCT ON (m.keep_id, t.ext_id) t.ctid AS cid, m.keep_id
  FROM broker_other_listing t JOIN broker_merge m ON t.broker_id = m.old_id AND m.old_id <> m.keep_id
  WHERE NOT EXISTS (SELECT 1 FROM broker_other_listing k WHERE k.broker_id = m.keep_id AND k.ext_id = t.ext_id)
  ORDER BY m.keep_id, t.ext_id, t.broker_id)
UPDATE broker_other_listing t SET broker_id = cand.keep_id FROM cand WHERE t.ctid = cand.cid;

WITH cand AS (
  SELECT DISTINCT ON (m.keep_id, t.sfr_id) t.ctid AS cid, m.keep_id
  FROM broker_sfr t JOIN broker_merge m ON t.broker_id = m.old_id AND m.old_id <> m.keep_id
  WHERE NOT EXISTS (SELECT 1 FROM broker_sfr k WHERE k.broker_id = m.keep_id AND k.sfr_id = t.sfr_id)
  ORDER BY m.keep_id, t.sfr_id, t.broker_id)
UPDATE broker_sfr t SET broker_id = cand.keep_id FROM cand WHERE t.ctid = cand.cid;

UPDATE broker_closed_listing t SET broker_id = m.keep_id FROM broker_merge m
 WHERE t.broker_id = m.old_id AND m.old_id <> m.keep_id;   -- PK is (id): no collision

-- ── 5. Delete dup brokers (cascade clears leftover colliding child + bfh rows) ─
DELETE FROM broker b USING broker_merge m WHERE b.id = m.old_id AND m.old_id <> m.keep_id;

-- ── 6. Repoint firm_id everywhere (safe now — broker dups are gone) ─────────
UPDATE broker              b SET firm_id = fm.keep_id FROM firm_merge fm WHERE b.firm_id = fm.old_id;
UPDATE condo               c SET firm_id = fm.keep_id FROM firm_merge fm WHERE c.firm_id = fm.old_id;
UPDATE sfr                 s SET firm_id = fm.keep_id FROM firm_merge fm WHERE s.firm_id = fm.old_id;
UPDATE broker_firm_history h SET firm_id = fm.keep_id FROM firm_merge fm WHERE h.firm_id = fm.old_id;

-- ── 7. Delete orphan firm variants, rename survivor to least-shouty ─────────
DELETE FROM firm f USING firm_merge m WHERE f.id = m.old_id AND m.old_id <> m.keep_id;
UPDATE firm f SET name = m.keep_name FROM firm_merge m WHERE f.id = m.keep_id AND f.name <> m.keep_name;

-- ── 8. Rebuild broker_firm_history current rows from the cleaned broker table ─
DELETE FROM broker_firm_history WHERE is_current;
INSERT INTO broker_firm_history (broker_id, firm_id, firm_name, is_current, source)
SELECT b.id, b.firm_id, f.name, true, 'backfill'
FROM broker b JOIN firm f ON f.id = b.firm_id
WHERE b.firm_id IS NOT NULL
ON CONFLICT DO NOTHING;

-- ── Report (inspect BEFORE committing on the dry run) ───────────────────────
SELECT (SELECT count(*) FROM firm)                                 AS firms_after,
       (SELECT count(*) FROM broker)                               AS brokers_after,
       (SELECT count(*) FROM broker WHERE firm_id IS NULL)         AS null_firm_brokers_after,
       (SELECT count(*) FROM (SELECT 1 FROM firm GROUP BY lower(trim(name)) HAVING count(*)>1) t) AS firm_case_dups_left,
       (SELECT count(*) FROM broker_firm_history WHERE is_current) AS bfh_current_after;

COMMIT;

-- Rollback recipe: restore each table from its *_bak_20260731 (FK-order-aware),
-- or restore from the pre-change pg_dump.