-- Same cutoff semantics and hashes; bulk membership join and incremental row digest.
CREATE OR REPLACE FUNCTION public.xs_payroll_cutoff_snapshot_fast(co text, mon text)
 RETURNS jsonb
 LANGUAGE plpgsql
 SET jit TO 'off'
 SET work_mem TO '64MB'
 SET max_parallel_workers_per_gather TO '2'
AS $function$
DECLARE run "LegacyDirectRun"; ev record; table_name text; matches integer; complete boolean:=true; conflicts jsonb; version text; approval "PayrollCutoffApproval"; data_version text; source_version text; pending integer; datasets jsonb:='[]'; ready boolean;
BEGIN
 IF NOT xs_payroll_authority_ready() THEN RETURN xs_payroll_cutoff_snapshot(co,mon);END IF;
 IF mon !~ '^\d{4}-(0[1-9]|1[0-2])$' THEN RAISE EXCEPTION '工资月份无效'; END IF;
 SELECT * INTO run FROM "LegacyDirectRun" WHERE company=co AND status IN ('completed','completed-with-preserved-conflicts') ORDER BY "finishedAt" DESC LIMIT 1;
 IF NOT FOUND THEN complete:=false; END IF;
 IF (SELECT count(*) FROM "LegacyDirectEvidence" WHERE company=co AND "runId"=run."runId" AND model IN (54,59,60))<>3 THEN complete:=false; END IF;
 FOR ev IN SELECT * FROM "LegacyDirectEvidence" WHERE company=co AND "runId"=run."runId" AND model IN (54,59,60) LOOP
  table_name:=CASE ev.model WHEN 54 THEN 'CourseAppointment' WHEN 59 THEN 'LessonRecord' ELSE 'MemoryPracticeRecord' END;
  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), wanted AS MATERIALIZED (SELECT DISTINCT jsonb_array_elements_text($3) AS id) SELECT count(*)::int FROM (SELECT l."generalId" FROM local_rows l JOIN wanted w ON w.id=l."generalId"::text GROUP BY l."generalId" HAVING count(*)=1) x',table_name) INTO matches USING co,ev.model,ev.ids;
  complete:=complete AND matches=ev."sourceCount" AND ev."sourceChecksum"=ev."targetChecksum" AND jsonb_array_length(ev.ids)=ev."sourceCount" AND ev."sourceChecksum"=md5((SELECT COALESCE(string_agg(v,E'\n' ORDER BY v::numeric),'') FROM jsonb_array_elements_text(ev.ids) v));
  datasets:=datasets||jsonb_build_array(jsonb_build_object('model',ev.model,'source',ev."sourceCount",'matched',matches,'checksum',ev."sourceChecksum"));
 END LOOP;
 SELECT COALESCE(jsonb_agg(c ORDER BY c->>'id'),'[]') INTO conflicts FROM xs_payroll_cutoff_conflicts(co,mon) c;
 SELECT count(*) INTO pending FROM jsonb_array_elements(conflicts) c WHERE NOT (c->>'resolved')::boolean;
 -- Native row serialization covers every column without constructing a JSON object per row.
 data_version:=xs_payroll_data_version(co);
 source_version:=xs_payroll_source_version_fast(co);
 version:=md5(jsonb_build_array(data_version,source_version,run."runId",run."finishedAt",datasets,conflicts)::text);
 SELECT * INTO approval FROM "PayrollCutoffApproval" WHERE company=co AND month=mon;
 ready:=complete AND pending=0 AND approval.version=version AND NOT EXISTS(SELECT 1 FROM "LegacyDirectRun" WHERE company=co AND status='running');
 RETURN jsonb_build_object('month',mon,'version',version,'sourceVersion',source_version,'runId',run."runId",'cutoffAt',run."finishedAt",'complete',complete,'pending',pending,'conflicts',conflicts,'datasets',datasets,'approved',COALESCE(ready,false),'monthClosed',COALESCE(run."finishedAt">=((mon||'-01')::date+interval '1 month') AT TIME ZONE 'Asia/Shanghai',false),'approval',to_jsonb(approval));
END $function$;