From c0ecf2fa3bebd46bdfd0169efd73b89653d1dfed Mon Sep 17 00:00:00 2001 From: Jakob Wennberg Date: Wed, 2 Sep 2026 18:18:35 +0200 Subject: [PATCH] feat(parties): merge with survivor choice and 30-day undo (#2175) Phase 1d. merge_parties soft-merges live parties into a survivor (merged_into + archived_at), unions alias keys and copies an org number the survivor lacks; facts, identities and role links stay where they are and readers resolve through canonical_party_id(). undo_party_merge restores the merged rows and the survivor snapshot within 30 days and logs a split decision; a second undo and other companies are refused. pg-real tests cover merge, undo, the window, chained merges and every rejection path. Co-authored-by: Jakob Wennberg <311770904+jakobwennberg-oss@users.noreply.github.com> Co-authored-by: Claude Fable 5.1 --- .../20260902210000_party_merge_undo.sql | 193 ++++++++++++++++++ tests/pg/party-merge-undo.pg.test.ts | 135 ++++++++++++ 2 files changed, 328 insertions(+) create mode 100644 supabase/migrations/20260902210000_party_merge_undo.sql create mode 100644 tests/pg/party-merge-undo.pg.test.ts diff --git a/supabase/migrations/20260902210000_party_merge_undo.sql b/supabase/migrations/20260902210000_party_merge_undo.sql new file mode 100644 index 00000000..b7dedd1d --- /dev/null +++ b/supabase/migrations/20260902210000_party_merge_undo.sql @@ -0,0 +1,193 @@ +-- Parties, phase 1d: merge with survivor choice, and undo within 30 days. +-- +-- A merge is soft: the merged party rows stay, with merged_into pointing at +-- the survivor and archived_at set. Facts, identities and role links keep +-- their party_id; readers resolve the canonical party through +-- canonical_party_id(). That is what makes undo a plain restore rather than +-- a reconstruction, which the July research found is where Attio, Pennylane +-- and Ramp lose data (merges irreversible, a paid undo market exists). +-- +-- What the survivor gains at merge time: the union of alias keys (so the +-- suggestion pipeline attaches future keys to the survivor) and an org +-- number if it had none. Both are snapshotted in the decision and restored +-- on undo. + +CREATE OR REPLACE FUNCTION public.canonical_party_id(p_party_id uuid) +RETURNS uuid +LANGUAGE sql +STABLE +SECURITY INVOKER +SET search_path TO 'public' +AS $$ + WITH RECURSIVE chain AS ( + SELECT p.id, p.merged_into, 1 AS depth + FROM public.parties p WHERE p.id = p_party_id + UNION ALL + SELECT p.id, p.merged_into, c.depth + 1 + FROM chain c JOIN public.parties p ON p.id = c.merged_into + WHERE c.merged_into IS NOT NULL AND c.depth < 16 + ) + SELECT id FROM chain ORDER BY depth DESC LIMIT 1; +$$; + +REVOKE ALL ON FUNCTION public.canonical_party_id(uuid) FROM PUBLIC, anon; +GRANT EXECUTE ON FUNCTION public.canonical_party_id(uuid) TO authenticated, service_role; +COMMENT ON FUNCTION public.canonical_party_id(uuid) IS + 'Follows merged_into to the surviving party (at most 16 hops). Returns the input id for a live party.'; + +-- ── Merge ─────────────────────────────────────────────────────────────────── +CREATE OR REPLACE FUNCTION public.merge_parties( + p_company_id uuid, + p_user_id uuid, + p_survivor uuid, + p_merged uuid[], + p_note text DEFAULT NULL +) +RETURNS uuid +LANGUAGE plpgsql +SECURITY INVOKER +SET search_path TO 'public' +AS $$ +DECLARE + v_survivor public.parties%ROWTYPE; + v_ids uuid[]; + v_before jsonb; + v_alias text[]; + v_org text; + v_decision uuid; +BEGIN + IF auth.uid() IS NOT NULL AND auth.uid() <> p_user_id THEN + RAISE EXCEPTION 'merge_parties: p_user_id must be the caller' USING ERRCODE = '42501'; + END IF; + + SELECT * INTO v_survivor FROM public.parties + WHERE id = p_survivor AND company_id = p_company_id AND merged_into IS NULL + FOR UPDATE; + IF v_survivor.id IS NULL THEN + RAISE EXCEPTION 'merge_parties: survivor is not a live party of this company' USING ERRCODE = '23503'; + END IF; + + -- Only live parties of this company, never the survivor, each once. + SELECT coalesce(array_agg(p.id ORDER BY p.created_at, p.id), '{}') INTO v_ids + FROM public.parties p + WHERE p.company_id = p_company_id AND p.id = ANY(p_merged) AND p.id <> p_survivor AND p.merged_into IS NULL; + IF coalesce(array_length(v_ids, 1), 0) <> (SELECT count(DISTINCT x) FROM unnest(p_merged) AS x WHERE x <> p_survivor) THEN + RAISE EXCEPTION 'merge_parties: every merged id must be a live party of this company' USING ERRCODE = '23503'; + END IF; + IF coalesce(array_length(v_ids, 1), 0) = 0 THEN + RAISE EXCEPTION 'merge_parties: nothing to merge' USING ERRCODE = '22023'; + END IF; + + PERFORM 1 FROM public.parties WHERE id = ANY(v_ids) FOR UPDATE; + + SELECT ARRAY(SELECT DISTINCT x FROM ( + SELECT unnest(v_survivor.alias_keys) AS x + UNION ALL + SELECT unnest(p.alias_keys) FROM public.parties p WHERE p.id = ANY(v_ids) + ) a), + coalesce(v_survivor.org_number, ( + SELECT p.org_number FROM public.parties p + WHERE p.id = ANY(v_ids) AND p.org_number IS NOT NULL + ORDER BY p.created_at, p.id LIMIT 1)) + INTO v_alias, v_org; + + v_before := jsonb_build_object( + 'survivor', jsonb_build_object('id', v_survivor.id, 'alias_keys', to_jsonb(v_survivor.alias_keys), 'org_number', v_survivor.org_number), + 'merged', (SELECT jsonb_agg(jsonb_build_object('id', p.id, 'display_name', p.display_name, 'status', p.status, 'archived_at', p.archived_at) ORDER BY p.created_at, p.id) + FROM public.parties p WHERE p.id = ANY(v_ids)) + ); + + UPDATE public.parties + SET merged_into = p_survivor, archived_at = coalesce(archived_at, now()) + WHERE id = ANY(v_ids); + + UPDATE public.parties + SET alias_keys = v_alias, org_number = v_org + WHERE id = p_survivor; + + INSERT INTO public.party_decisions (party_id, company_id, user_id, kind, before, after, note) + VALUES (p_survivor, p_company_id, p_user_id, 'merge', v_before, + jsonb_build_object('survivor', jsonb_build_object('id', p_survivor, 'alias_keys', to_jsonb(v_alias), 'org_number', v_org), 'merged', to_jsonb(v_ids)), + p_note) + RETURNING id INTO v_decision; + + RETURN v_decision; +END; +$$; + +REVOKE ALL ON FUNCTION public.merge_parties(uuid, uuid, uuid, uuid[], text) FROM PUBLIC, anon; +GRANT EXECUTE ON FUNCTION public.merge_parties(uuid, uuid, uuid, uuid[], text) TO authenticated, service_role; +COMMENT ON FUNCTION public.merge_parties(uuid, uuid, uuid, uuid[], text) IS + 'Soft-merges live parties into a survivor (merged_into + archived_at), unions alias keys, copies an org number the survivor lacks. Returns the merge decision id for undo_party_merge.'; + +-- ── Undo ──────────────────────────────────────────────────────────────────── +CREATE OR REPLACE FUNCTION public.undo_party_merge( + p_company_id uuid, + p_user_id uuid, + p_decision_id uuid +) +RETURNS integer +LANGUAGE plpgsql +SECURITY INVOKER +SET search_path TO 'public' +AS $$ +DECLARE + v_d public.party_decisions%ROWTYPE; + v_ids uuid[]; + v_restored integer; +BEGIN + IF auth.uid() IS NOT NULL AND auth.uid() <> p_user_id THEN + RAISE EXCEPTION 'undo_party_merge: p_user_id must be the caller' USING ERRCODE = '42501'; + END IF; + + SELECT * INTO v_d FROM public.party_decisions + WHERE id = p_decision_id AND company_id = p_company_id AND kind = 'merge' + FOR UPDATE; + IF v_d.id IS NULL THEN + RAISE EXCEPTION 'undo_party_merge: no merge decision % in this company', p_decision_id USING ERRCODE = '23503'; + END IF; + IF v_d.created_at < now() - interval '30 days' THEN + RAISE EXCEPTION 'undo_party_merge: the 30-day undo window has passed' USING ERRCODE = '22023'; + END IF; + IF EXISTS (SELECT 1 FROM public.party_decisions u + WHERE u.company_id = p_company_id AND u.kind = 'split' AND u.before->>'decision_id' = p_decision_id::text) THEN + RAISE EXCEPTION 'undo_party_merge: merge % is already undone', p_decision_id USING ERRCODE = '22023'; + END IF; + + v_ids := ARRAY(SELECT (x)::uuid FROM jsonb_array_elements_text(v_d.after->'merged') AS x); + + -- Survivor first: it may carry an org number copied from a merged row, and + -- that row cannot become live again while the survivor still holds it. + UPDATE public.parties + SET alias_keys = ARRAY(SELECT jsonb_array_elements_text(v_d.before->'survivor'->'alias_keys')), + org_number = nullif(v_d.before->'survivor'->>'org_number', '') + WHERE id = v_d.party_id AND company_id = p_company_id; + + -- Only rows still merged into this survivor come back; a party merged + -- onward since then belongs to a later decision. + WITH restored AS ( + UPDATE public.parties p + SET merged_into = NULL, + archived_at = (SELECT nullif(m->>'archived_at', '')::timestamptz + FROM jsonb_array_elements(v_d.before->'merged') m WHERE (m->>'id')::uuid = p.id) + WHERE p.id = ANY(v_ids) AND p.company_id = p_company_id AND p.merged_into = v_d.party_id + RETURNING p.id + ) + SELECT count(*) INTO v_restored FROM restored; + + INSERT INTO public.party_decisions (party_id, company_id, user_id, kind, before, after, note) + VALUES (v_d.party_id, p_company_id, p_user_id, 'split', + jsonb_build_object('decision_id', p_decision_id, 'merged', to_jsonb(v_ids)), + jsonb_build_object('restored', v_restored), + 'undo merge'); + + RETURN v_restored; +END; +$$; + +REVOKE ALL ON FUNCTION public.undo_party_merge(uuid, uuid, uuid) FROM PUBLIC, anon; +GRANT EXECUTE ON FUNCTION public.undo_party_merge(uuid, uuid, uuid) TO authenticated, service_role; +COMMENT ON FUNCTION public.undo_party_merge(uuid, uuid, uuid) IS + 'Restores the parties merged by one merge decision within 30 days: clears merged_into, restores the survivor''s alias keys and org number, logs a split decision. Returns the number of restored parties.'; + +NOTIFY pgrst, 'reload schema'; diff --git a/tests/pg/party-merge-undo.pg.test.ts b/tests/pg/party-merge-undo.pg.test.ts new file mode 100644 index 00000000..f1be06c0 --- /dev/null +++ b/tests/pg/party-merge-undo.pg.test.ts @@ -0,0 +1,135 @@ +import { describe, expect, it } from 'vitest' +import { getPool, withUserContext } from './setup' +import { seedCompany } from './fixtures' + +const ORG_A = '5564300142' +const ORG_B = '5560125790' + +async function party(companyId: string, userId: string, name: string, over: { org?: string; alias?: string[]; status?: string } = {}): Promise { + const { rows } = await getPool().query<{ id: string }>( + `INSERT INTO public.parties (company_id, user_id, display_name, org_number, alias_keys, status) + VALUES ($1, $2, $3, $4, $5::text[], $6) RETURNING id`, + [companyId, userId, name, over.org ?? null, over.alias ?? [], over.status ?? 'confirmed'], + ) + return rows[0]!.id +} + +async function merge(companyId: string, userId: string, survivor: string, merged: string[], note: string | null = null): Promise { + const { rows } = await getPool().query<{ id: string }>( + `SELECT public.merge_parties($1, $2, $3, $4::uuid[], $5) AS id`, + [companyId, userId, survivor, merged, note], + ) + return rows[0]!.id +} + +async function undo(companyId: string, userId: string, decisionId: string): Promise { + const { rows } = await getPool().query<{ n: number }>(`SELECT public.undo_party_merge($1, $2, $3) AS n`, [companyId, userId, decisionId]) + return rows[0]!.n +} + +async function state(id: string) { + const { rows } = await getPool().query<{ merged_into: string | null; archived: boolean; alias_keys: string[]; org_number: string | null; canonical: string }>( + `SELECT merged_into, archived_at IS NOT NULL AS archived, alias_keys, org_number, public.canonical_party_id(id) AS canonical + FROM public.parties WHERE id = $1`, + [id], + ) + return rows[0]! +} + +describe('merge_parties / undo_party_merge (pg)', () => { + it('soft-merges into the survivor, unions aliases, copies a missing org number, and logs the decision', async () => { + const c = await seedCompany() + const survivor = await party(c.companyId, c.userId, 'Beijer Byggmaterial AB', { alias: ['beijer byggmaterial'] }) + const dupA = await party(c.companyId, c.userId, 'BEIJER BYGG', { alias: ['beijer bygg'], org: ORG_A, status: 'suggested' }) + const dupB = await party(c.companyId, c.userId, 'Beijer', { alias: ['beijer', 'beijer byggmaterial'] }) + // A role link and an identity on a merged party stay where they are. + await getPool().query(`INSERT INTO public.suppliers (company_id, user_id, name, party_id) VALUES ($1, $2, 'Beijer', $3)`, [c.companyId, c.userId, dupA]) + await getPool().query( + `INSERT INTO public.party_identities (party_id, company_id, user_id, scheme, value, source) VALUES ($1, $2, $3, 'bankgiro', '53170900', 'document')`, + [dupA, c.companyId, c.userId], + ) + + const decision = await merge(c.companyId, c.userId, survivor, [dupA, dupB], 'same supplier') + + const s = await state(survivor) + expect(s.merged_into).toBeNull() + expect(s.archived).toBe(false) + expect([...s.alias_keys].sort()).toEqual(['beijer', 'beijer bygg', 'beijer byggmaterial']) + expect(s.org_number).toBe(ORG_A) + const a = await state(dupA) + expect(a.merged_into).toBe(survivor) + expect(a.archived).toBe(true) + expect(a.canonical).toBe(survivor) + expect(a.org_number).toBe(ORG_A) // kept on the merged row; the partial unique index ignores merged rows + const b = await state(dupB) + expect(b.merged_into).toBe(survivor) + expect(b.canonical).toBe(survivor) + + const d = await getPool().query<{ kind: string; party_id: string; note: string; merged: string[] }>( + `SELECT kind, party_id, note, ARRAY(SELECT jsonb_array_elements_text(after->'merged')) AS merged FROM public.party_decisions WHERE id = $1`, + [decision], + ) + expect(d.rows[0]).toEqual({ kind: 'merge', party_id: survivor, note: 'same supplier', merged: [dupA, dupB] }) + const links = await getPool().query<{ s: string; i: string }>( + `SELECT (SELECT party_id FROM public.suppliers WHERE company_id = $1) AS s, (SELECT party_id FROM public.party_identities WHERE company_id = $1) AS i`, + [c.companyId], + ) + expect(links.rows[0]).toEqual({ s: dupA, i: dupA }) + }) + + it('undo restores merged rows, the survivor snapshot, and logs a split; a second undo is refused', async () => { + const c = await seedCompany() + const survivor = await party(c.companyId, c.userId, 'Loopia AB', { alias: ['loopia'] }) + const dup = await party(c.companyId, c.userId, 'Loopia Webbhotell', { alias: ['loopia webbhotell'], org: ORG_B }) + const decision = await merge(c.companyId, c.userId, survivor, [dup]) + expect((await state(survivor)).org_number).toBe(ORG_B) + + expect(await undo(c.companyId, c.userId, decision)).toBe(1) + const s = await state(survivor) + expect(s.alias_keys).toEqual(['loopia']) + expect(s.org_number).toBeNull() + const d = await state(dup) + expect(d.merged_into).toBeNull() + expect(d.archived).toBe(false) + expect(d.canonical).toBe(dup) + const kinds = await getPool().query<{ kind: string }>(`SELECT kind FROM public.party_decisions WHERE company_id = $1 ORDER BY created_at`, [c.companyId]) + expect(kinds.rows.map((r) => r.kind)).toEqual(['merge', 'split']) + + await expect(undo(c.companyId, c.userId, decision)).rejects.toMatchObject({ code: '22023' }) + }) + + it('refuses undo after 30 days and for a merge of another company', async () => { + const c = await seedCompany() + const other = await seedCompany() + const survivor = await party(c.companyId, c.userId, 'A') + const dup = await party(c.companyId, c.userId, 'B') + const decision = await merge(c.companyId, c.userId, survivor, [dup]) + await expect(undo(other.companyId, other.userId, decision)).rejects.toMatchObject({ code: '23503' }) + await getPool().query(`UPDATE public.party_decisions SET created_at = now() - interval '31 days' WHERE id = $1`, [decision]) + await expect(undo(c.companyId, c.userId, decision)).rejects.toMatchObject({ code: '22023' }) + expect((await state(dup)).merged_into).toBe(survivor) + }) + + it('refuses a merged or foreign survivor, foreign or already-merged victims, and a spoofed user', async () => { + const mine = await seedCompany() + const theirs = await seedCompany() + const a = await party(mine.companyId, mine.userId, 'A') + const b = await party(mine.companyId, mine.userId, 'B') + const cId = await party(mine.companyId, mine.userId, 'C') + const foreign = await party(theirs.companyId, theirs.userId, 'F') + await merge(mine.companyId, mine.userId, a, [b]) + await expect(merge(mine.companyId, mine.userId, b, [cId])).rejects.toMatchObject({ code: '23503' }) + await expect(merge(mine.companyId, mine.userId, foreign, [cId])).rejects.toMatchObject({ code: '23503' }) + await expect(merge(mine.companyId, mine.userId, a, [foreign])).rejects.toMatchObject({ code: '23503' }) + await expect(merge(mine.companyId, mine.userId, a, [b])).rejects.toMatchObject({ code: '23503' }) + await expect(merge(mine.companyId, mine.userId, a, [a])).rejects.toMatchObject({ code: '22023' }) + await expect( + withUserContext(mine.userId, (client) => + client.query(`SELECT public.merge_parties($1, $2, $3, $4::uuid[], NULL)`, [mine.companyId, theirs.userId, a, [cId]]), + ), + ).rejects.toMatchObject({ code: '42501' }) + // Chained merge resolves to the final survivor. + await merge(mine.companyId, mine.userId, cId, [a]) + expect((await state(b)).canonical).toBe(cId) + }) +})