-- Additive. Apply after payroll-integrity.sql and legacy-direct-sync.sql.
CREATE TABLE IF NOT EXISTS "PayrollCutoffDecision" (company text, month text, "conflictId" text, fingerprint text, actor text, reason text, "createdAt" timestamptz DEFAULT now(), PRIMARY KEY(company,month,"conflictId"));
CREATE TABLE IF NOT EXISTS "PayrollCutoffApproval" (company text, month text, version text, "runId" text, "cutoffAt" timestamptz, actor text, reason text, "createdAt" timestamptz DEFAULT now(), PRIMARY KEY(company,month));
REVOKE ALL ON "PayrollCutoffDecision","PayrollCutoffApproval" FROM PUBLIC;
CREATE OR REPLACE FUNCTION xs_payroll_cutoff_conflicts(co text, mon text) RETURNS SETOF jsonb LANGUAGE plpgsql AS $$
DECLARE r record; item jsonb; payload jsonb; fields jsonb; dt text; v_fingerprint text; cosmetic boolean;
BEGIN
 FOR r IN SELECT f.*,c."modelId",c."itemId",c.title,c.subtitle FROM "LegacySyncConflict" f LEFT JOIN "CommonModel" c ON c.company=f.company AND c."generalId"::text=f."legacyKey" AND c."modelId"::text=f.dataset WHERE f.company=co AND f.dataset IN ('54','59','60') AND NOT COALESCE(f.resolved,false) ORDER BY f."objectId" LOOP
  item:=NULL;
  IF r.dataset='54' THEN SELECT to_jsonb(a) INTO item FROM "CourseAppointment" a WHERE a.company=co AND a.id::text=r."itemId"::text;
  ELSIF r.dataset='59' THEN SELECT to_jsonb(a) INTO item FROM "LessonRecord" a WHERE a.company=co AND a.id::text=r."itemId"::text;
  ELSE SELECT to_jsonb(a) INTO item FROM "MemoryPracticeRecord" a WHERE a.company=co AND a.id::text=r."itemId"::text; END IF;
  dt:=CASE r.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;
  -- All older unresolved rewards can enter a later payroll; unknown dates always block.
  IF xs_payroll_time(dt) IS NOT NULL AND to_char(xs_payroll_time(dt) AT TIME ZONE 'Asia/Shanghai','YYYY-MM')>mon THEN CONTINUE; END IF;
  BEGIN payload:=r.message::jsonb; EXCEPTION WHEN others THEN payload:='{}'; END;
  fields:=COALESCE(payload->'fields','[]');
  IF jsonb_typeof(fields)<>'array' THEN fields:='[]'; END IF;
  cosmetic:=jsonb_array_length(fields)>0 AND NOT EXISTS(SELECT 1 FROM jsonb_array_elements_text(fields) f WHERE f NOT IN ('title','inputer'));
  v_fingerprint:=md5(jsonb_build_array(r.message,item,r.title,r.subtitle,(SELECT "runId" FROM "LegacyDirectRun" WHERE company=co AND status IN ('completed','completed-with-preserved-conflicts') ORDER BY "finishedAt" DESC LIMIT 1))::text);
  RETURN NEXT jsonb_build_object('id',r."objectId",'dataset',r.dataset,'legacyKey',r."legacyKey",'student',r.title,'course',r.subtitle,'date',dt,'fields',fields,'reason',payload->>'reason','current',item,'fingerprint',v_fingerprint,'cosmetic',cosmetic,'resolved',cosmetic OR EXISTS(SELECT 1 FROM "PayrollCutoffDecision" d WHERE d.company=co AND d.month=mon AND d."conflictId"=r."objectId" AND d.fingerprint=v_fingerprint));
 END LOOP;
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_cutoff_snapshot(co text, mon text) RETURNS jsonb LANGUAGE plpgsql AS $$
DECLARE run "LegacyDirectRun"; ev record; table_name text; matches integer; complete boolean:=true; conflicts jsonb; version text; approval "PayrollCutoffApproval"; data_version text; pending integer; datasets jsonb:='[]'; ready boolean;
BEGIN
 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('SELECT count(*)::int FROM (SELECT c."generalId" FROM "CommonModel" c JOIN %I a ON a.company=c.company AND a.id::text=c."itemId"::text WHERE c.company=$1 AND c."modelId"=$2 AND c."generalId"::text IN (SELECT jsonb_array_elements_text($3)) GROUP BY c."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;
 SELECT md5(COALESCE(string_agg(hash,'' ORDER BY kind,hash),'')) INTO data_version FROM (
 SELECT 'common' kind,md5(to_jsonb(c)::text) hash FROM "CommonModel" c WHERE company=co AND "modelId" IN (54,59,60)
 UNION ALL SELECT 'appointment',md5(to_jsonb(c)::text) FROM "CourseAppointment" c WHERE company=co
 UNION ALL SELECT 'lesson',md5(to_jsonb(c)::text) FROM "LessonRecord" c WHERE company=co
 UNION ALL SELECT 'review',md5(to_jsonb(c)::text) FROM "MemoryPracticeRecord" c WHERE company=co) current_rows;
 version:=md5(jsonb_build_array(data_version,xs_payroll_source_version(co),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',xs_payroll_source_version(co),'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 $$;
CREATE OR REPLACE FUNCTION xs_payroll_cutoff_confirm(co text,actor text,mon text,expected text,reason text,cid text DEFAULT '',fp text DEFAULT '') RETURNS jsonb LANGUAGE plpgsql AS $$
DECLARE snapshot jsonb; row jsonb;
BEGIN
 IF length(trim(reason))<2 THEN RAISE EXCEPTION '请填写核验依据'; END IF;
 PERFORM pg_advisory_xact_lock(hashtext('xs-payroll:'||co));
 PERFORM pg_advisory_xact_lock(hashtext('direct-sync:'||co));
 LOCK TABLE "LegacyDirectRun","LegacyDirectEvidence","LegacySyncConflict","CommonModel","CourseAppointment","LessonRecord","MemoryPracticeRecord","ReviewReward","TeachingCompensationRule","CourseTeachingProfile","CoachPayRate","LessonRecordFact","PayrollCutoffDecision","PayrollCutoffApproval" IN SHARE MODE;
 snapshot:=xs_payroll_cutoff_snapshot(co,mon);
 IF snapshot->>'version' IS DISTINCT FROM expected THEN RAISE EXCEPTION '来源已更新，请重新核验'; END IF;
 IF cid<>'' THEN
  SELECT c INTO row FROM jsonb_array_elements(snapshot->'conflicts') c WHERE c->>'id'=cid AND c->>'fingerprint'=fp;
  IF row IS NULL OR row->'current'='null'::jsonb THEN RAISE EXCEPTION '来源记录缺失或差异已变化，请先修复'; END IF;
  INSERT INTO "PayrollCutoffDecision" VALUES(co,mon,cid,fp,actor,reason,now()) ON CONFLICT(company,month,"conflictId") DO UPDATE SET fingerprint=EXCLUDED.fingerprint,actor=EXCLUDED.actor,reason=EXCLUDED.reason,"createdAt"=now();
 ELSE
  IF NOT (snapshot->>'complete')::boolean OR (snapshot->>'pending')::int>0 OR EXISTS(SELECT 1 FROM "LegacyDirectRun" WHERE company=co AND status='running') THEN RAISE EXCEPTION '同步凭据不完整、仍有差异或同步正在执行'; END IF;
  INSERT INTO "PayrollCutoffApproval" VALUES(co,mon,expected,snapshot->>'runId',(snapshot->>'cutoffAt')::timestamptz,actor,reason,now()) ON CONFLICT(company,month) DO UPDATE SET version=EXCLUDED.version,"runId"=EXCLUDED."runId","cutoffAt"=EXCLUDED."cutoffAt",actor=EXCLUDED.actor,reason=EXCLUDED.reason,"createdAt"=now();
 END IF;
 RETURN xs_payroll_cutoff_snapshot(co,mon);
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_assert_cutoff(co text,mon text,op text,expected text) RETURNS void LANGUAGE plpgsql AS $$
DECLARE s jsonb;
BEGIN
 LOCK TABLE "LegacyDirectRun","LegacyDirectEvidence","LegacySyncConflict","CommonModel","CourseAppointment","LessonRecord","MemoryPracticeRecord","ReviewReward","TeachingCompensationRule","CourseTeachingProfile","CoachPayRate","LessonRecordFact","PayrollCutoffDecision","PayrollCutoffApproval" IN SHARE MODE;
 s:=xs_payroll_cutoff_snapshot(co,mon);
 IF NOT (s->>'approved')::boolean OR s->>'version' IS DISTINCT FROM expected THEN RAISE EXCEPTION '工资截止来源确认已失效，请重新核验'; END IF;
 IF op='review' AND NOT (s->>'monthClosed')::boolean THEN RAISE EXCEPTION '本月尚未完成整月同步，只能保存工资草稿'; END IF;
END $$;