-- 053_wayleave_retained_flag.sql
-- retained becomes a flag on the claim, not a value in its source column.
-- (docs/2026-08-13-wayleave-coverage-approvals.md, Phase 5)
--
-- 042 modelled "kept after a removal was rejected" by rewriting the row: source became
-- 'retained' and source_ref became 'polygon:12', with the originals parked in
-- retained_from_source / retained_from_ref. That conflates two different facts - WHICH
-- SOURCE CLAIMS THIS PREMISE, and IS THAT SOURCE STILL PRODUCING IT - into one column, and
-- it has a concrete failure:
--
--   ERROR: duplicate key value violates unique constraint wayleave_coverage_uq
--   DETAIL: Key (wayleave_id, uprn, source, source_ref)=(1, 6, direct, 6) already exists.
--
-- Rewriting the row VACATES its natural key. Reject the removal of direct UPRN 6 and
-- (1,6,direct,'6') is free; add UPRN 6 again and a new row takes it; reverse the rejection
-- and the restore collides. The reverse UPDATE is the only statement in the decision
-- endpoint that writes a natural key with no conflict handling, and it cannot have any -
-- there is no sensible merge of two rows claiming the same premise from the same source.
--
-- A FLAG CANNOT COLLIDE. The row keeps its identity for its whole life; retained says the
-- source stopped producing it and someone decided to keep it anyway. Reverse becomes
-- `retained = false` with no key to restore, and the failure is gone by construction
-- rather than guarded against.
--
-- retained_from_source and retained_from_ref go with it: the row now holds its own source
-- and source_ref, so provenance never left. They existed only to undo the rewrite.
--
-- The four source views exclude retained rows, which is what stops them being diffed
-- against a source that no longer claims them - the same reason the old design moved them
-- out of 'polygon'. If the source starts producing the premise again it reappears as a
-- fresh pending addition, and approving it clears the flag, which is handled in
-- wayleave_approval_decide.php rather than here.

-- ── 1. Guard ────────────────────────────────────────────────────────────────

DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM information_schema.columns
                    WHERE table_schema = 'wayleave' AND table_name = 'wayleave_coverage'
                      AND column_name = 'withdrawn') THEN
        RAISE EXCEPTION '053: apply 052 first';
    END IF;
END $$;

-- ── 2. The flag, and the rows that need moving onto it ──────────────────────

ALTER TABLE wayleave.wayleave_coverage
    ADD COLUMN IF NOT EXISTS retained boolean NOT NULL DEFAULT false;

COMMENT ON COLUMN wayleave.wayleave_coverage.retained IS
    'The source stopped producing this premise and the removal was rejected, so it stays covered. A FLAG, not a source value: rewriting source/source_ref vacated the row natural key, so re-adding the same premise and then reversing the rejection collided on wayleave_coverage_uq. The row keeps its own source and source_ref for its whole life.';

DO $$
DECLARE
    v_conflicts text;
    v_moved     integer;
BEGIN
    -- Restoring a rewritten row can hit the very collision this migration exists to
    -- remove: another row may already hold the key it is going back to. Name them before
    -- touching anything - failing here with the list beats failing on a constraint.
    SELECT string_agg(r.uprn::text, ', ') INTO v_conflicts
      FROM wayleave.wayleave_coverage r
      JOIN wayleave.wayleave_coverage o
        ON o.wayleave_id = r.wayleave_id
       AND o.uprn        = r.uprn
       AND o.source      = r.retained_from_source
       AND o.source_ref  = r.retained_from_ref
       AND o.id         <> r.id
     WHERE r.source = 'retained';

    IF v_conflicts IS NOT NULL THEN
        RAISE EXCEPTION
            '053: premise(s) % are both retained and claimed again by the same source. Delete the newer claim, or reverse the rejection, before applying this.',
            v_conflicts;
    END IF;

    UPDATE wayleave.wayleave_coverage
       SET source               = COALESCE(retained_from_source, source),
           source_ref           = COALESCE(retained_from_ref, source_ref),
           retained             = true,
           retained_from_source = NULL,
           retained_from_ref    = NULL
     WHERE source = 'retained';
    GET DIAGNOSTICS v_moved = ROW_COUNT;
    RAISE NOTICE '053: % retained row(s) moved onto the flag', v_moved;
END $$;

-- ── 3. Views ────────────────────────────────────────────────────────────────
-- vw_wayleave_coverage and vw_uprns_retained both read the dropped columns, so both are
-- rebuilt. The four source views gain `AND NOT c.retained`.

-- EVERY DEPENDENT VIEW COMES DOWN BEFORE THE COLUMNS GO. vw_wayleave_coverage and
-- vw_uprns_retained both select retained_from_source / retained_from_ref, and a column a
-- view references cannot be dropped. Ordering the ALTERs first fails the whole migration
-- on a dependency error rather than on anything to do with its purpose.

DROP VIEW wayleave.vw_wayleave_coverage;
DROP VIEW wayleave.vw_uprns_retained;
DROP VIEW wayleave.vw_wayleave_pending;
DROP VIEW wayleave.vw_uprns_from_boundary;
DROP VIEW wayleave.vw_uprns_from_stocklists;
DROP VIEW wayleave.vw_uprns_from_titles;

-- 'retained' is no longer a source. Left in the allowed list it would be a value nothing
-- writes and every view has to keep excluding.
ALTER TABLE wayleave.wayleave_coverage
    DROP CONSTRAINT IF EXISTS wayleave_coverage_source_ck;

ALTER TABLE wayleave.wayleave_coverage
    ADD CONSTRAINT wayleave_coverage_source_ck
    CHECK (source IN ('polygon', 'direct', 'stocklist', 'title'));

ALTER TABLE wayleave.wayleave_coverage DROP COLUMN IF EXISTS retained_from_source;
ALTER TABLE wayleave.wayleave_coverage DROP COLUMN IF EXISTS retained_from_ref;

CREATE INDEX IF NOT EXISTS wayleave_coverage_retained_idx
    ON wayleave.wayleave_coverage (wayleave_id) WHERE retained;


CREATE VIEW wayleave.vw_wayleave_coverage AS
 SELECT c.wayleave_id,
    c.uprn,
    c.source,
    c.source_ref,
    c.retained,
    c.decided_user,
    c.decided_datetime,
    a.reference AS approval_reference,
    abp.address AS address_full,
    abp.geom
   FROM wayleave.wayleave_coverage c
     LEFT JOIN wayleave.wayleave_approvals a ON a.id = c.approval_id
     LEFT JOIN basedata.abp abp ON abp.uprn = c.uprn
  WHERE c.state = 'approved';

-- Its own layer because nothing produces these any more, so there is no source layer they
-- could sit on. They keep their original source and source_ref, which is now simply the
-- row's own - the answer to "where did this come from" no longer needs a second pair of
-- columns to survive the decision.
CREATE VIEW wayleave.vw_uprns_retained AS
 SELECT c.wayleave_id,
    c.uprn,
    c.source AS retained_from_source,
    c.source_ref AS retained_from_ref,
    'retained'::text AS src,
    c.state,
    false AS in_source,
    'retained'::text AS map_state,
    c.decided_user,
    c.decided_datetime,
    u.username AS decided_username,
    abp.address AS address_full,
    abp.geom
   FROM wayleave.wayleave_coverage c
     LEFT JOIN users.users u ON u.id = c.decided_user
     LEFT JOIN basedata.abp abp ON abp.uprn = c.uprn
  WHERE c.retained = true AND c.state = 'approved';

CREATE OR REPLACE VIEW wayleave.vw_uprns_from_direct AS
 SELECT c.wayleave_id,
    c.uprn,
    c.source_ref,
    'direct'::text AS src,
    c.state,
    (NOT c.withdrawn) AS in_source,
        CASE
            WHEN c.withdrawn AND c.state = 'approved' THEN 'removing'
            ELSE c.state::text
        END AS map_state,
    c.decided_user,
    c.decided_datetime,
    u.username AS decided_username,
    abp.address AS address_full,
    abp.geom
   FROM wayleave.wayleave_coverage c
     LEFT JOIN users.users u ON u.id = c.decided_user
     LEFT JOIN basedata.abp abp ON abp.uprn = c.uprn
  WHERE c.source = 'direct'
    AND c.state <> 'removed'
    AND NOT c.retained
    AND NOT (c.withdrawn AND c.state <> 'approved');

CREATE VIEW wayleave.vw_uprns_from_boundary AS
 SELECT p.wayleave_id, abp.uprn, p.id::text AS source_ref, 'polygon'::text AS src,
    COALESCE(c.state, 'pending'::character varying) AS state,
    true AS in_source,
    COALESCE(c.state, 'pending'::character varying)::text AS map_state,
    c.decided_user, c.decided_datetime, u.username AS decided_username,
    abp.address AS address_full, abp.geom
   FROM wayleave.wayleave_polygons p
     JOIN basedata.abp abp ON public.st_within(abp.geom, p.geom)
     LEFT JOIN wayleave.wayleave_coverage c
            ON c.wayleave_id = p.wayleave_id AND c.uprn = abp.uprn
           AND c.source = 'polygon' AND c.source_ref::text = p.id::text
           AND c.state <> 'removed' AND NOT c.retained
     LEFT JOIN users.users u ON u.id = c.decided_user
  WHERE p.is_deleted = false
UNION ALL
 SELECT c.wayleave_id, c.uprn, c.source_ref::text, 'polygon'::text, c.state, false,
    CASE WHEN c.state = 'approved' THEN 'removing' ELSE c.state::text END,
    c.decided_user, c.decided_datetime, u.username, abp.address, abp.geom
   FROM wayleave.wayleave_coverage c
     LEFT JOIN users.users u ON u.id = c.decided_user
     LEFT JOIN basedata.abp abp ON abp.uprn = c.uprn
  WHERE c.source = 'polygon' AND c.state <> 'removed' AND NOT c.retained
    AND NOT EXISTS (
        SELECT 1 FROM wayleave.wayleave_polygons p2
          JOIN basedata.abp a2 ON public.st_within(a2.geom, p2.geom)
         WHERE p2.is_deleted = false AND p2.wayleave_id = c.wayleave_id
           AND p2.id::text = c.source_ref::text AND a2.uprn = c.uprn);

CREATE VIEW wayleave.vw_uprns_from_stocklists AS
 SELECT ws.wayleave_id, sp.uprn, ws.stocklist_id::text AS source_ref, 'stocklist'::text AS src,
    COALESCE(c.state, 'pending'::character varying) AS state,
    true AS in_source,
    COALESCE(c.state, 'pending'::character varying)::text AS map_state,
    c.decided_user, c.decided_datetime, u.username AS decided_username,
    abp.address AS address_full, abp.geom
   FROM wayleave.wayleave_stocklists ws
     JOIN stocklists.stocklist_premises sp ON sp.stocklist_id = ws.stocklist_id
     LEFT JOIN wayleave.wayleave_coverage c
            ON c.wayleave_id = ws.wayleave_id AND c.uprn = sp.uprn
           AND c.source = 'stocklist' AND c.source_ref::text = ws.stocklist_id::text
           AND c.state <> 'removed' AND NOT c.retained
     LEFT JOIN users.users u ON u.id = c.decided_user
     LEFT JOIN basedata.abp abp ON abp.uprn = sp.uprn
  WHERE ws.is_deleted = false
UNION ALL
 SELECT c.wayleave_id, c.uprn, c.source_ref::text, 'stocklist'::text, c.state, false,
    CASE WHEN c.state = 'approved' THEN 'removing' ELSE c.state::text END,
    c.decided_user, c.decided_datetime, u.username, abp.address, abp.geom
   FROM wayleave.wayleave_coverage c
     LEFT JOIN users.users u ON u.id = c.decided_user
     LEFT JOIN basedata.abp abp ON abp.uprn = c.uprn
  WHERE c.source = 'stocklist' AND c.state <> 'removed' AND NOT c.retained
    AND NOT EXISTS (
        SELECT 1 FROM wayleave.wayleave_stocklists ws2
          JOIN stocklists.stocklist_premises sp2 ON sp2.stocklist_id = ws2.stocklist_id
         WHERE ws2.is_deleted = false AND ws2.wayleave_id = c.wayleave_id
           AND ws2.stocklist_id::text = c.source_ref::text AND sp2.uprn = c.uprn);

CREATE VIEW wayleave.vw_uprns_from_titles AS
 SELECT wl.wayleave_id, abp.uprn, wl.title_number::text AS source_ref, 'title'::text AS src,
    COALESCE(c.state, 'pending'::character varying) AS state,
    true AS in_source,
    COALESCE(c.state, 'pending'::character varying)::text AS map_state,
    c.decided_user, c.decided_datetime, u.username AS decided_username,
    abp.address AS address_full, abp.geom
   FROM wayleave.wayleave_landregistry wl
     JOIN basedata.abp abp ON abp.title_no = wl.title_number::text
     LEFT JOIN wayleave.wayleave_coverage c
            ON c.wayleave_id = wl.wayleave_id AND c.uprn = abp.uprn
           AND c.source = 'title' AND c.source_ref::text = wl.title_number::text
           AND c.state <> 'removed' AND NOT c.retained
     LEFT JOIN users.users u ON u.id = c.decided_user
  WHERE wl.is_deleted = false AND abp.uprn IS NOT NULL
UNION ALL
 SELECT c.wayleave_id, c.uprn, c.source_ref::text, 'title'::text, c.state, false,
    CASE WHEN c.state = 'approved' THEN 'removing' ELSE c.state::text END,
    c.decided_user, c.decided_datetime, u.username, abp.address, abp.geom
   FROM wayleave.wayleave_coverage c
     LEFT JOIN users.users u ON u.id = c.decided_user
     LEFT JOIN basedata.abp abp ON abp.uprn = c.uprn
  WHERE c.source = 'title' AND c.state <> 'removed' AND NOT c.retained
    AND NOT EXISTS (
        SELECT 1 FROM wayleave.wayleave_landregistry wl2
          JOIN basedata.abp a2 ON a2.title_no = wl2.title_number::text
         WHERE wl2.is_deleted = false AND wl2.wayleave_id = c.wayleave_id
           AND wl2.title_number::text = c.source_ref::text AND a2.uprn = c.uprn);

CREATE VIEW wayleave.vw_wayleave_pending AS
 WITH all_src AS (
         SELECT wayleave_id, uprn, src, source_ref, map_state FROM wayleave.vw_uprns_from_boundary
         UNION ALL
         SELECT wayleave_id, uprn, src, source_ref, map_state FROM wayleave.vw_uprns_from_direct
         UNION ALL
         SELECT wayleave_id, uprn, src, source_ref, map_state FROM wayleave.vw_uprns_from_stocklists
         UNION ALL
         SELECT wayleave_id, uprn, src, source_ref, map_state FROM wayleave.vw_uprns_from_titles
        ), marked AS (
         SELECT a.wayleave_id, a.uprn, a.src, a.source_ref, a.map_state,
                EXISTS (SELECT 1 FROM wayleave.wayleave_coverage c
                         WHERE c.wayleave_id = a.wayleave_id AND c.uprn = a.uprn
                           AND c.state = 'approved') AS covered_any,
                EXISTS (SELECT 1 FROM wayleave.wayleave_coverage c
                         WHERE c.wayleave_id = a.wayleave_id AND c.uprn = a.uprn
                           AND c.state = 'approved'
                           AND NOT (c.source = a.src AND c.source_ref = a.source_ref)) AS covered_other
           FROM all_src a
          WHERE a.map_state IN ('pending', 'removing')
        )
 SELECT m.wayleave_id, m.src, m.source_ref,
    count(*) FILTER (WHERE m.map_state = 'pending')                          AS adding_gross,
    count(*) FILTER (WHERE m.map_state = 'pending'  AND NOT m.covered_any)   AS adding_net,
    count(*) FILTER (WHERE m.map_state = 'removing')                         AS removing_gross,
    count(*) FILTER (WHERE m.map_state = 'removing' AND NOT m.covered_other) AS removing_net
   FROM marked m
  GROUP BY m.wayleave_id, m.src, m.source_ref;

COMMENT ON VIEW wayleave.vw_wayleave_pending IS
    'One row per outstanding source change. Pending is derived, not stored: only a direct UPRN writes a pending row, so counting wayleave_coverage would miss every polygon, stocklist and title change. Filter by wayleave_id - since 048 that predicate reaches the base tables.';

-- ── 4. Verify ───────────────────────────────────────────────────────────────

DO $$
DECLARE
    v_bad text;
BEGIN
    IF EXISTS (SELECT 1 FROM wayleave.wayleave_coverage WHERE source = 'retained') THEN
        RAISE EXCEPTION '053: rows still carry source = retained';
    END IF;

    SELECT string_agg(v, ', ') INTO v_bad
      FROM unnest(ARRAY['retained_from_source', 'retained_from_ref']) AS v
     WHERE EXISTS (SELECT 1 FROM information_schema.columns
                    WHERE table_schema = 'wayleave' AND table_name = 'wayleave_coverage'
                      AND column_name = v);
    IF v_bad IS NOT NULL THEN
        RAISE EXCEPTION '053: column(s) not dropped: %', v_bad;
    END IF;

    -- Every source view must exclude retained rows, or a premise kept after a rejected
    -- removal would be diffed against the source that stopped producing it and reappear
    -- as a pending removal on every single load - the loop 'retained' exists to break.
    SELECT string_agg(v, ', ') INTO v_bad
      FROM unnest(ARRAY['vw_uprns_from_boundary', 'vw_uprns_from_direct',
                        'vw_uprns_from_stocklists', 'vw_uprns_from_titles']) AS v
     WHERE pg_get_viewdef(('wayleave.' || v)::regclass) NOT LIKE '%retained%';
    IF v_bad IS NOT NULL THEN
        RAISE EXCEPTION '053: retained not excluded from: %', v_bad;
    END IF;

    SELECT string_agg(v, ', ') INTO v_bad
      FROM unnest(ARRAY['vw_uprns_from_boundary', 'vw_uprns_from_direct',
                        'vw_uprns_from_stocklists', 'vw_uprns_from_titles',
                        'vw_uprns_retained']) AS v
     WHERE NOT EXISTS (SELECT 1 FROM information_schema.columns
                        WHERE table_schema = 'wayleave' AND table_name = v
                          AND column_name = 'map_state');
    IF v_bad IS NOT NULL THEN
        RAISE EXCEPTION '053: map_state missing from: %', v_bad;
    END IF;

    PERFORM 1 FROM wayleave.vw_wayleave_coverage     LIMIT 1;
    PERFORM 1 FROM wayleave.vw_uprns_retained        LIMIT 1;
    PERFORM 1 FROM wayleave.vw_uprns_from_boundary   LIMIT 1;
    PERFORM 1 FROM wayleave.vw_uprns_from_direct     LIMIT 1;
    PERFORM 1 FROM wayleave.vw_uprns_from_stocklists LIMIT 1;
    PERFORM 1 FROM wayleave.vw_uprns_from_titles     LIMIT 1;
    PERFORM 1 FROM wayleave.vw_wayleave_pending      LIMIT 1;
END $$;
