-- One fresh read snapshot; no result cache. Deploy after payroll-source-index and cutoff dependencies.
-- Call inside REPEATABLE READ READ ONLY because downstream legacy read helpers may be VOLATILE.
-- Capture the exact date parser used for conservative transport pruning. A later
-- parser change fails closed rather than assuming it still agrees with JavaScript.
DO $install$
DECLARE expected text:=md5(pg_get_functiondef('xs_payroll_time(text)'::regprocedure));
 conflicts_expected text:=md5(pg_get_functiondef('xs_payroll_cutoff_conflicts(text,text)'::regprocedure));
BEGIN
 IF to_regprocedure('xs_payroll_read_snapshot_authority_ready()') IS NOT NULL THEN RETURN; END IF;
 EXECUTE format('CREATE OR REPLACE FUNCTION xs_payroll_read_snapshot_authority_ready() RETURNS boolean LANGUAGE sql STABLE AS %L',
  'SELECT xs_payroll_authority_ready() AND md5(pg_get_functiondef(''xs_payroll_time(text)''::regprocedure))='||quote_literal(expected)||
  ' AND md5(pg_get_functiondef(''xs_payroll_cutoff_conflicts(text,text)''::regprocedure))='||quote_literal(conflicts_expected));
END $install$;

CREATE OR REPLACE FUNCTION public.xs_payroll_cutoff_from_version(co text, mon text, provided_version text)
 RETURNS jsonb
 LANGUAGE plpgsql STABLE
 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_read_snapshot_authority_ready() THEN RETURN NULL;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;
  matches:=xs_payroll_evidence_matches(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_bulk(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:=provided_version;
 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$;

CREATE OR REPLACE FUNCTION xs_payroll_read_snapshot(co text, mon text, for_dashboard boolean DEFAULT false, include_scope_rows boolean DEFAULT false)
RETURNS jsonb LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
DECLARE result jsonb; cutoff jsonb;
BEGIN
 IF NOT xs_payroll_read_snapshot_authority_ready() THEN RETURN NULL; END IF;
 IF mon !~ '^\d{4}-(0[1-9]|1[0-2])$' THEN RAISE EXCEPTION '工资月份无效'; END IF;
 WITH sources AS MATERIALIZED (
  SELECT row_data AS s,row_text,standard_key,scope_data,ordinal AS ord FROM xs_payroll_source_entries_scoped(co,mon,NULL,NULL,NOT for_dashboard)
 ), identity_state AS MATERIALIZED (
  SELECT EXISTS(SELECT 1 FROM sources GROUP BY standard_key HAVING count(*)>1) AS duplicates
 ), source_version AS MATERIALIZED (
  SELECT CASE WHEN for_dashboard THEN '' ELSE md5(
   COALESCE((SELECT string_agg(row_text,'|' ORDER BY row_text) FROM sources),'')||
   COALESCE((SELECT string_agg(to_jsonb(r)::text,'|' ORDER BY r."objectId") FROM "TeachingCompensationRule" r WHERE r."company"=co),'')||
   COALESCE((SELECT string_agg(to_jsonb(p)::text,'|' ORDER BY p."objectId") FROM "CourseTeachingProfile" p WHERE p."company"=co),'')||
   COALESCE((SELECT string_agg(to_jsonb(p)::text,'|' ORDER BY p."objectId") FROM "CoachPayRate" p WHERE p."company"=co),'')||
   COALESCE((SELECT string_agg(to_jsonb(r)::text,'|' ORDER BY r."objectId") FROM "ReviewRewardIncome" r WHERE r."company"=co AND r."status"='earned'),'')) END AS value
 ), candidates AS MATERIALIZED (
  SELECT s,ord FROM sources WHERE s IS NOT NULL
 ) SELECT jsonb_build_object(
  'rows',COALESCE((SELECT jsonb_agg(s ORDER BY ord) FROM candidates),'[]'::jsonb),
  'version',(SELECT value FROM source_version),
  'total',(SELECT count(*) FROM sources),
  'hasDuplicates',(SELECT duplicates FROM identity_state),
  'scopeRows',CASE WHEN include_scope_rows THEN COALESCE((SELECT jsonb_agg(scope_data ORDER BY ord) FROM sources),'[]'::jsonb) ELSE NULL END
 ) INTO result;
 IF NOT for_dashboard THEN
  cutoff:=xs_payroll_cutoff_from_version(co,mon,result->>'version');
  IF cutoff IS NULL THEN RETURN NULL; END IF;
  -- Full conflicts/approval already participated in the cutoff hash; only sync metadata crosses the wire.
  cutoff:=jsonb_build_object('approved',cutoff->'approved','monthClosed',cutoff->'monthClosed','version',cutoff->'version',
   'complete',cutoff->'complete','pending',cutoff->'pending','datasets',cutoff->'datasets',
   'cutoffAt',cutoff->'cutoffAt','sourceVersion',cutoff->'sourceVersion');
 END IF;
 RETURN result||jsonb_build_object('cutoff',cutoff);
END $$;

-- Select only identity/display fields before JSON encoding. The catalog check
-- preserves every supported alias even on schemas that omit optional columns.
CREATE OR REPLACE FUNCTION xs_payroll_read_users(co text) RETURNS SETOF jsonb
LANGUAGE plpgsql STABLE AS $$
DECLARE columns text;
BEGIN
 SELECT string_agg(quote_ident(attname),',' ORDER BY attnum) INTO columns FROM pg_attribute
 WHERE attrelid='"_User"'::regclass AND attnum>0 AND NOT attisdropped
 AND attname=ANY(ARRAY['objectId','legacyUserId','legacyUserData','legacyuserdata','realName','realname','nickname','nickName','nickname2','username','mobile','phone']);
 RETURN QUERY EXECUTE 'SELECT to_jsonb(u) FROM (SELECT '||columns||' FROM "_User" WHERE company=$1) u' USING co;
END $$;