-- Parse legacy conflict payloads with the authoritative function's exception semantics.
CREATE OR REPLACE FUNCTION xs_payroll_conflict_payload(value text) RETURNS jsonb LANGUAGE plpgsql IMMUTABLE AS $$
BEGIN RETURN value::jsonb;EXCEPTION WHEN others THEN RETURN '{}'::jsonb;END $$;

-- Fresh set-based conflict calculation; no result cache and no financial writes.
CREATE OR REPLACE FUNCTION xs_payroll_cutoff_conflicts_bulk(co text,mon text) RETURNS SETOF jsonb
LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
BEGIN
 IF NOT xs_payroll_read_snapshot_authority_ready() THEN RETURN QUERY SELECT s FROM xs_payroll_cutoff_conflicts(co,mon) s;RETURN;END IF;
 RETURN QUERY
 WITH latest AS MATERIALIZED (
  SELECT "runId" FROM "LegacyDirectRun" WHERE company=co AND status IN ('completed','completed-with-preserved-conflicts') ORDER BY "finishedAt" DESC LIMIT 1
 ), roots AS MATERIALIZED (
  SELECT f.*,c."modelId",c."itemId",c.title,c.subtitle,row_number() OVER(ORDER BY f."objectId") ord
  FROM "LegacySyncConflict" f LEFT JOIN "CommonModel" c ON c.company=f.company AND c."generalId"::text=f."legacyKey"
    AND c."modelId"=CASE f.dataset WHEN '54' THEN 54 WHEN '59' THEN 59 WHEN '60' THEN 60 END
  WHERE f.company=co AND f.dataset IN ('54','59','60') AND NOT COALESCE(f.resolved,false)
 ), joined AS MATERIALIZED (
  SELECT r.*,CASE r.dataset WHEN '54' THEN to_jsonb(a) WHEN '59' THEN to_jsonb(l) ELSE to_jsonb(m) END item,
   xs_payroll_conflict_payload(r.message) payload
  FROM roots r
  LEFT JOIN "CourseAppointment" a ON r.dataset='54' AND a.company=co AND a.id=r."itemId"
  LEFT JOIN "LessonRecord" l ON r.dataset='59' AND l.company=co AND l.id=r."itemId"
  LEFT JOIN "MemoryPracticeRecord" m ON r.dataset='60' AND m.company=co AND m.id=r."itemId"
 ), ambiguous AS MATERIALIZED (
  SELECT EXISTS(SELECT 1 FROM roots GROUP BY "objectId" HAVING count(*)>1)
   OR EXISTS(SELECT 1 FROM joined GROUP BY ord HAVING count(*)>1) value
 ), described AS MATERIALIZED (
  SELECT j.*,CASE dataset WHEN '54' THEN item->>'yysj' WHEN '59' THEN COALESCE(NULLIF(item->>'lessonAt',''),NULLIF(item->>'plpjsj','')) ELSE COALESCE(NULLIF(item->>'wcsj',''),NULLIF(item->>'kywrq','')) END dt,
   CASE WHEN jsonb_typeof(payload->'fields')='array' THEN payload->'fields' ELSE '[]'::jsonb END fields
  FROM joined j WHERE NOT (SELECT value FROM ambiguous)
 ), fingerprinted AS MATERIALIZED (
  SELECT d.*,md5(jsonb_build_array(message,item,title,subtitle,(SELECT "runId" FROM latest))::text) fingerprint,
   jsonb_array_length(fields)>0 AND NOT EXISTS(SELECT 1 FROM jsonb_array_elements_text(fields) f WHERE f NOT IN ('title','inputer')) cosmetic
  FROM described d
  WHERE xs_payroll_time(dt) IS NULL OR to_char(xs_payroll_time(dt) AT TIME ZONE 'Asia/Shanghai','YYYY-MM')<=mon
 ), results AS (SELECT r.ord,jsonb_build_object('id',r."objectId",'dataset',r.dataset,'legacyKey',r."legacyKey",'student',r.title,'course',r.subtitle,'date',r.dt,'fields',r.fields,'reason',r.payload->>'reason','current',r.item,'fingerprint',r.fingerprint,'cosmetic',r.cosmetic,
  'resolved',r.cosmetic OR EXISTS(SELECT 1 FROM "PayrollCutoffDecision" d WHERE d.company=co AND d.month=mon AND d."conflictId"=r."objectId" AND d.fingerprint=r.fingerprint))
 AS value FROM fingerprinted r
 UNION ALL SELECT c.ord,c.value FROM ambiguous guard
 CROSS JOIN LATERAL xs_payroll_cutoff_conflicts(co,mon) WITH ORDINALITY c(value,ord) WHERE guard.value
 ) SELECT value FROM results ORDER BY ord;
END $$;

-- Exact text membership before numeric duplicate counting, without the redundant
-- DISTINCT materialization or a per-ID nested lookup. Local settings affect only
-- this one fresh evidence query, not source readers or financial commands.
CREATE OR REPLACE FUNCTION xs_payroll_evidence_matches(co text,model numeric,ids jsonb) RETURNS integer
LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' SET enable_mergejoin=off SET enable_nestloop=off AS $$
DECLARE tab text; matches integer;
BEGIN
 tab:=CASE model WHEN 54 THEN 'CourseAppointment' WHEN 59 THEN 'LessonRecord' WHEN 60 THEN 'MemoryPracticeRecord' END;
 IF tab IS NULL THEN RAISE EXCEPTION 'Unsupported evidence model';END IF;
 EXECUTE format('WITH local_rows AS MATERIALIZED (SELECT c."generalId" FROM "CommonModel" c JOIN %I a ON a.company=c.company AND a.id=c."itemId" WHERE c.company=$1 AND c."modelId"=$2) SELECT count(*)::int FROM (SELECT l."generalId" FROM local_rows l WHERE EXISTS(SELECT 1 FROM jsonb_array_elements_text($3) w WHERE w=l."generalId"::text) GROUP BY l."generalId" HAVING count(*)=1) x',tab) INTO matches USING co,model,ids;
 RETURN matches;
END $$;