xiaoshu-app/scripts/sql/payroll-incremental-reads.sql

-- Derived, transactionally maintained fingerprints. No financial source is changed.
-- Install atomically after verifying original definitions; rollback restores them first.
CREATE TABLE IF NOT EXISTS "PayrollReadDigest" (
 kind text NOT NULL, company text NOT NULL, "objectId" text NOT NULL,
 model numeric, hash text NOT NULL, PRIMARY KEY(kind,company,"objectId")
);
CREATE INDEX IF NOT EXISTS payroll_read_digest_order ON "PayrollReadDigest" (company,kind COLLATE "C",hash COLLATE "C");
CREATE TABLE IF NOT EXISTS "PayrollReadAuthority" (name text PRIMARY KEY, hash text);
INSERT INTO "PayrollReadAuthority" SELECT name,md5(pg_get_functiondef(to_regprocedure(name))) FROM (VALUES ('xs_payroll_sources(text)'),('xs_payroll_source_version(text)'),('xs_payroll_cutoff_snapshot(text,text)')) v(name) ON CONFLICT(name) DO NOTHING;
CREATE TABLE IF NOT EXISTS "PayrollReadDigestMeta" (kind text PRIMARY KEY, signature text NOT NULL);
CREATE TABLE IF NOT EXISTS "PayrollReadChangeScope" (company text NOT NULL, scope text NOT NULL, xid xid8 NOT NULL, PRIMARY KEY(company,scope,xid));
CREATE TABLE IF NOT EXISTS "PayrollReadChange" (company text NOT NULL, xid xid8 NOT NULL, PRIMARY KEY(company,xid));
CREATE TABLE IF NOT EXISTS "PayrollVerifiedReadCache" (
 company text NOT NULL, key text NOT NULL, stamp text NOT NULL, payload jsonb NOT NULL,
 "createdAt" timestamptz NOT NULL DEFAULT now(), PRIMARY KEY(company,key)
);
REVOKE ALL ON "PayrollReadDigest","PayrollReadDigestMeta","PayrollVerifiedReadCache","PayrollReadChange","PayrollReadChangeScope","PayrollReadAuthority" FROM PUBLIC;

CREATE OR REPLACE FUNCTION xs_payroll_authority_ready() RETURNS boolean LANGUAGE sql STABLE AS $$
 SELECT count(*)=3 AND bool_and(hash IS NOT DISTINCT FROM md5(pg_get_functiondef(to_regprocedure(name)))) FROM "PayrollReadAuthority";
$$;
CREATE OR REPLACE FUNCTION xs_payroll_digest_signature(tab text) RETURNS text LANGUAGE sql STABLE AS $$
 SELECT md5(COALESCE((SELECT string_agg(a.attnum||':'||a.attname||':'||a.atttypid||':'||a.atttypmod,'|' ORDER BY a.attnum) FROM pg_attribute a WHERE a.attrelid=to_regclass(format('public.%I',tab)) AND a.attnum>0 AND NOT a.attisdropped),'')||'|'||
 COALESCE((SELECT string_agg(t.tgname||':'||t.xmin||':'||t.ctid||':'||t.tgenabled::text||':'||pg_get_triggerdef(t.oid),'|' ORDER BY t.tgname) FROM pg_trigger t WHERE t.tgrelid=to_regclass(format('public.%I',tab)) AND t.tgname IN ('xs_payroll_digest_row','xs_payroll_digest_truncate')),'')||'|'||
 COALESCE(pg_get_functiondef(to_regprocedure('xs_payroll_digest_row()')),'')||COALESCE(pg_get_functiondef(to_regprocedure('xs_payroll_digest_truncate()')),''));
$$;
CREATE OR REPLACE FUNCTION xs_payroll_digest_row() RETURNS trigger LANGUAGE plpgsql
 SET TimeZone='Etc/GMT-8' SET DateStyle='ISO, MDY' SET IntervalStyle='postgres' SET extra_float_digits=1 SET bytea_output='hex' AS $$
DECLARE k text:=TG_ARGV[0];
BEGIN
 IF TG_OP<>'INSERT' AND OLD.company IS NOT NULL AND (k<>'common' OR (to_jsonb(OLD)->>'modelId')::numeric IN (54,59,60)) THEN
  INSERT INTO "PayrollReadChange" VALUES(OLD.company,pg_current_xact_id()) ON CONFLICT DO NOTHING;
  INSERT INTO "PayrollReadChangeScope" VALUES(OLD.company,CASE WHEN k='common' THEN k||':'||((to_jsonb(OLD)->>'modelId')::numeric)::int::text ELSE k END,pg_current_xact_id()) ON CONFLICT DO NOTHING;
 END IF;
 IF TG_OP<>'DELETE' AND NEW.company IS NOT NULL AND (k<>'common' OR (to_jsonb(NEW)->>'modelId')::numeric IN (54,59,60)) THEN
  INSERT INTO "PayrollReadChange" VALUES(NEW.company,pg_current_xact_id()) ON CONFLICT DO NOTHING;
  INSERT INTO "PayrollReadChangeScope" VALUES(NEW.company,CASE WHEN k='common' THEN k||':'||((to_jsonb(NEW)->>'modelId')::numeric)::int::text ELSE k END,pg_current_xact_id()) ON CONFLICT DO NOTHING;
 END IF;
 IF TG_OP<>'INSERT' THEN DELETE FROM "PayrollReadDigest" WHERE kind=k AND company=OLD.company AND "objectId"=OLD."objectId"; END IF;
 IF TG_OP<>'DELETE' AND NEW.company IS NOT NULL AND (k<>'common' OR (to_jsonb(NEW)->>'modelId')::numeric IN (54,59,60)) THEN
  INSERT INTO "PayrollReadDigest"(kind,company,"objectId",model,hash) VALUES(k,NEW.company,NEW."objectId",CASE WHEN k='common' THEN (to_jsonb(NEW)->>'modelId')::numeric END,md5(NEW::text));
 END IF;
 RETURN NULL;
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_digest_truncate() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN INSERT INTO "PayrollReadChangeScope" SELECT DISTINCT company,CASE WHEN kind='common' THEN kind||':'||model::int::text ELSE kind END,pg_current_xact_id() FROM "PayrollReadDigest" WHERE kind=TG_ARGV[0] ON CONFLICT DO NOTHING;INSERT INTO "PayrollReadChange" SELECT DISTINCT company,pg_current_xact_id() FROM "PayrollReadDigest" WHERE kind=TG_ARGV[0] ON CONFLICT DO NOTHING;DELETE FROM "PayrollReadDigest" WHERE kind=TG_ARGV[0];RETURN NULL;END $$;

-- Matching environment and unchanged, ALWAYS-enabled triggers are prerequisites.
-- DDL (including disable/re-enable) changes catalog tuple versions and fails closed.
CREATE OR REPLACE FUNCTION xs_payroll_digest_ready() RETURNS boolean LANGUAGE sql STABLE AS $$
 SELECT current_setting('TimeZone')='Etc/GMT-8' AND current_setting('DateStyle')='ISO, MDY'
 AND current_setting('IntervalStyle')='postgres' AND current_setting('extra_float_digits')='1' AND current_setting('bytea_output')='hex'
 AND NOT EXISTS(SELECT 1 FROM (VALUES ('common','CommonModel'),('appointment','CourseAppointment'),('lesson','LessonRecord'),('review','MemoryPracticeRecord')) v(kind,tab)
 LEFT JOIN "PayrollReadDigestMeta" m ON m.kind=v.kind
 WHERE m.signature IS DISTINCT FROM xs_payroll_digest_signature(v.tab)
 OR (SELECT count(*) FROM pg_trigger t WHERE t.tgrelid=to_regclass(format('public.%I',v.tab)) AND t.tgname IN ('xs_payroll_digest_row','xs_payroll_digest_truncate') AND t.tgenabled='A')<>2);
$$;
CREATE OR REPLACE FUNCTION xs_payroll_data_version(co text) RETURNS text LANGUAGE plpgsql SET jit=off SET work_mem='64MB' AS $$
DECLARE v text;
BEGIN
 IF xs_payroll_digest_ready() THEN
  SELECT md5(COALESCE(string_agg(hash,'' ORDER BY kind COLLATE "C",hash COLLATE "C"),'')) INTO v FROM "PayrollReadDigest" WHERE company=co;
 ELSE
  SELECT md5(COALESCE(string_agg(hash,'' ORDER BY kind COLLATE "C",hash COLLATE "C"),'')) INTO v FROM (
   SELECT 'common' kind,md5(c::text) hash FROM "CommonModel" c WHERE company=co AND "modelId" IN (54,59,60)
   UNION ALL SELECT 'appointment',md5(c::text) FROM "CourseAppointment" c WHERE company=co
   UNION ALL SELECT 'lesson',md5(c::text) FROM "LessonRecord" c WHERE company=co
   UNION ALL SELECT 'review',md5(c::text) FROM "MemoryPracticeRecord" c WHERE company=co) s;
 END IF;
 RETURN v;
END $$;

-- Installation locks source writers until triggers and the initial digest agree.
DO $$ DECLARE r record;BEGIN
 FOR r IN SELECT * FROM (VALUES ('common','CommonModel'),('appointment','CourseAppointment'),('lesson','LessonRecord'),('review','MemoryPracticeRecord')) v(kind,tab) LOOP
  EXECUTE format('LOCK TABLE %I IN SHARE ROW EXCLUSIVE MODE',r.tab);
  EXECUTE format('DROP TRIGGER IF EXISTS xs_payroll_digest_row ON %I',r.tab);
  EXECUTE format('DROP TRIGGER IF EXISTS xs_payroll_digest_truncate ON %I',r.tab);
  EXECUTE format('CREATE TRIGGER xs_payroll_digest_row AFTER INSERT OR UPDATE OR DELETE ON %I FOR EACH ROW EXECUTE FUNCTION xs_payroll_digest_row(%L)',r.tab,r.kind);
  EXECUTE format('CREATE TRIGGER xs_payroll_digest_truncate AFTER TRUNCATE ON %I FOR EACH STATEMENT EXECUTE FUNCTION xs_payroll_digest_truncate(%L)',r.tab,r.kind);
  EXECUTE format('ALTER TABLE %I ENABLE ALWAYS TRIGGER xs_payroll_digest_row',r.tab);
  EXECUTE format('ALTER TABLE %I ENABLE ALWAYS TRIGGER xs_payroll_digest_truncate',r.tab);
  DELETE FROM "PayrollReadDigest" WHERE kind=r.kind;
  -- Cast the actual source row, exactly as in the previous cutoff hash.
  EXECUTE format('INSERT INTO "PayrollReadDigest"(kind,company,"objectId",model,hash) SELECT %L,c.company,c."objectId",%s,md5(c::text) FROM %I c WHERE company IS NOT NULL %s',r.kind,CASE WHEN r.kind='common' THEN 'c."modelId"' ELSE 'NULL::numeric' END,r.tab,CASE WHEN r.kind='common' THEN 'AND c."modelId" IN (54,59,60)' ELSE '' END);
  INSERT INTO "PayrollReadDigestMeta" VALUES(r.kind,xs_payroll_digest_signature(r.tab)) ON CONFLICT(kind) DO UPDATE SET signature=EXCLUDED.signature;
 END LOOP;
END $$;