-- Cache is accepted only after checking every dependency in the current DB snapshot.
-- Financial commands never use this preview cache.
CREATE OR REPLACE FUNCTION xs_payroll_preview_stamp(co text) RETURNS text LANGUAGE plpgsql SET jit=off SET work_mem='64MB' AS $$
DECLARE tab text; parts text:='v2'; v text; meta text;
BEGIN
 IF NOT xs_payroll_digest_ready() OR NOT xs_payroll_authority_ready() THEN RETURN NULL;END IF;
 SELECT md5(string_agg(kind||':'||signature,'|' ORDER BY kind)) INTO v FROM "PayrollReadDigestMeta";parts:=parts||'|'||v;
 FOREACH tab IN ARRAY ARRAY['LessonRecordFact','ReviewReward','ReviewRewardCorrection','TeachingCompensationRule','CourseTeachingProfile','CoachPayRate','TeachingCourseCategory','PayrollLine','PayrollBatch','LegacyDirectRun','LegacyDirectEvidence','LegacySyncConflict','PayrollCutoffDecision','PayrollCutoffApproval','_User','Node','CourseBinding'] LOOP
  -- xmin + physical tuple address detect updates even when updatedAt was not touched.
  -- Deletion, insertion, same-transaction rewrites, and rollback retain MVCC semantics.
  EXECUTE format('SELECT md5(COALESCE(string_agg(xmin::text||'':''||ctid::text,'','' ORDER BY ctid),'''')) FROM %I WHERE company=$1',tab) INTO v USING co;
  parts:=parts||'|'||tab||':'||v;
 END LOOP;
 SELECT md5(COALESCE(string_agg(c.oid||':'||c.relfilenode||':'||a.attnum||':'||a.attname||':'||a.atttypid||':'||a.atttypmod,'|' ORDER BY c.oid,a.attnum),'')) INTO meta
 FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace JOIN pg_attribute a ON a.attrelid=c.oid
 WHERE n.nspname='public' AND a.attnum>0 AND NOT a.attisdropped AND c.relname IN ('CommonModel','CourseAppointment','LessonRecord','MemoryPracticeRecord','LessonRecordFact','ReviewReward','ReviewRewardCorrection','TeachingCompensationRule','CourseTeachingProfile','CoachPayRate','TeachingCourseCategory','PayrollLine','PayrollBatch','LegacyDirectRun','LegacyDirectEvidence','LegacySyncConflict','PayrollCutoffDecision','PayrollCutoffApproval','_User','Node','CourseBinding','PayrollReadChange');
 parts:=parts||'|'||meta||'|'||COALESCE(pg_get_viewdef(to_regclass('"ReviewRewardIncome"'),true),'');
 SELECT md5(COALESCE(string_agg(pg_get_functiondef(p.oid),'|' ORDER BY p.oid),'')) INTO meta FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace WHERE n.nspname='public' AND p.proname LIKE 'xs_%' AND p.prokind='f';
 parts:=parts||'|'||meta||'|'||COALESCE((SELECT md5(code) FROM "Function" WHERE path='xiaoshu/ops/gateway-v3'),'')||'|'||current_setting('TimeZone')||'|'||current_setting('DateStyle')||'|'||current_setting('extra_float_digits');
 RETURN md5(parts);
END $$;
-- Top-level transaction IDs are checked against the build's original MVCC snapshot.
-- This handles commits arriving out of order; max(id) and timestamps are insufficient.
CREATE OR REPLACE FUNCTION xs_payroll_snapshot_current(co text, saved text) RETURNS boolean LANGUAGE sql STABLE AS $$
 SELECT saved IS NOT NULL AND NOT EXISTS(SELECT 1 FROM "PayrollReadChange" c WHERE c.company=co AND c.xid>=pg_snapshot_xmin(saved::pg_snapshot) AND NOT pg_visible_in_snapshot(c.xid,saved::pg_snapshot));
$$;
CREATE OR REPLACE FUNCTION xs_payroll_preview_cache_read(co text, cache_key text) RETURNS jsonb LANGUAGE plpgsql SET jit=off AS $$
DECLARE signature text; result jsonb; read_snapshot text:=pg_current_snapshot()::text;
BEGIN
 signature:=xs_payroll_preview_stamp(co);
 SELECT payload->'result' INTO result FROM "PayrollVerifiedReadCache" WHERE company=co AND key=cache_key AND stamp=signature AND "createdAt">now()-interval '1 day' AND xs_payroll_snapshot_current(co,payload->>'snapshot');
 RETURN jsonb_build_object('stamp',signature,'snapshot',read_snapshot,'payload',result);
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_preview_cache_write(co text,cache_key text,expected text,result jsonb) RETURNS boolean LANGUAGE plpgsql SET jit=off AS $$
BEGIN
 IF expected IS NULL OR expected IS DISTINCT FROM xs_payroll_preview_stamp(co) OR NOT xs_payroll_snapshot_current(co,result->>'snapshot') THEN RETURN false;END IF;
 INSERT INTO "PayrollVerifiedReadCache" VALUES(co,cache_key,expected,result,now()) ON CONFLICT(company,key) DO UPDATE SET stamp=EXCLUDED.stamp,payload=EXCLUDED.payload,"createdAt"=EXCLUDED."createdAt";
 RETURN true;
END $$;