xiaoshu-app/scripts/sql/payroll-source-index.sql

-- A transactionally maintained projection index, validated against CURRENT source
-- dependencies at every read. It never defines row existence or financial authority.
-- Writes are best effort and never require a company-wide lock. Missing/stale rows
-- are rebuilt in the reader's snapshot; extra rows cannot become payable sources.
BEGIN;
CREATE TABLE IF NOT EXISTS "PayrollSourceIndex" (
 kind text NOT NULL,company text NOT NULL,"objectId" text NOT NULL,
 row_data jsonb NOT NULL,row_text text NOT NULL,input_signature text,
 generation text NOT NULL,selection jsonb NOT NULL DEFAULT '{}'::jsonb,PRIMARY KEY(kind,company,"objectId")
);
ALTER TABLE "PayrollSourceIndex" ADD COLUMN IF NOT EXISTS selection jsonb NOT NULL DEFAULT '{}'::jsonb;
CREATE TABLE IF NOT EXISTS "PayrollSourceIndexMeta" (
 singleton boolean PRIMARY KEY DEFAULT true CHECK(singleton),generation text NOT NULL,
 signature text NOT NULL,valid boolean NOT NULL DEFAULT false
);
REVOKE ALL ON "PayrollSourceIndex","PayrollSourceIndexMeta" FROM PUBLIC;
-- Only model 59 participates in canonical source selection; avoid scanning the
-- much larger combined CommonModel digest for models 54 and 60.
CREATE INDEX IF NOT EXISTS payroll_source_index_common59_digest
 ON "PayrollReadDigest" (company,"objectId") INCLUDE(hash) WHERE kind='common' AND model=59;

-- Conservative selectors reproduce the gateway transport rule. Unknown dates
-- remain candidates. Identity keys retain the existing SQL COALESCE text semantics.
CREATE OR REPLACE FUNCTION xs_payroll_source_index_metadata(data jsonb) RETURNS jsonb LANGUAGE sql STABLE AS $$
 SELECT jsonb_build_object('standard_key',data->>'standardLessonKey',
  'month_candidate_key',CASE WHEN data->>'lessonAt' ~ '^\d{4}-[0-9]{2}-[0-9]{2}T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9](\.[0-9]{1,3})?(Z|[+-][0-9]{2}:[0-9]{2})$'
    AND xs_payroll_time(data->>'lessonAt') IS NOT NULL THEN to_char(xs_payroll_time(data->>'lessonAt') AT TIME ZONE 'Asia/Shanghai','YYYY-MM') END,
  'scope_data',jsonb_build_array(data->'coachId',data->'jsmz',data->'storeId',data->'szmdid'),
  'coach_key',COALESCE(data->>'coachId',data->>'jsmz'),
  'appointment_key',COALESCE(data->>'appointmentGeneralId',data->>'yyds'));
$$;

CREATE OR REPLACE FUNCTION xs_payroll_source_index_signature() RETURNS text LANGUAGE sql STABLE AS $$
 SELECT md5(COALESCE((SELECT string_agg(c.oid||':'||c.relfilenode||':'||a.attnum||':'||a.attname||':'||a.atttypid||':'||a.atttypmod,'|' ORDER BY c.oid,a.attnum)
 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','LessonRecordFact','PayrollSourceIndex','PayrollSourceIndexMeta')),'')||'|'||
 COALESCE((SELECT string_agg(t.tgrelid||':'||t.tgname||':'||t.xmin||':'||t.ctid||':'||t.tgenabled::text||':'||pg_get_triggerdef(t.oid),'|' ORDER BY t.tgrelid,t.tgname)
 FROM pg_trigger t WHERE t.tgrelid IN ('"CommonModel"'::regclass,'"CourseAppointment"'::regclass,'"LessonRecord"'::regclass,'"LessonRecordFact"'::regclass,'"PayrollSourceIndexMeta"'::regclass) AND t.tgname IN ('xs_payroll_source_index_row','xs_payroll_source_index_truncate','xs_payroll_source_index_meta_guard')),'')||'|'||
 COALESCE((SELECT string_agg(pg_get_functiondef(p.oid),'|' ORDER BY p.oid) FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace
 WHERE n.nspname='public' AND p.prokind='f' AND (p.proname LIKE 'xs_payroll_source_index_%' OR p.proname IN ('xs_payroll_source_entries','xs_payroll_source_entries_scoped','xs_payroll_sources_indexed','xs_payroll_source_version_indexed','xs_payroll_time'))),''));
$$;
CREATE OR REPLACE FUNCTION xs_payroll_source_index_ready() RETURNS boolean LANGUAGE sql STABLE AS $$
 SELECT xs_payroll_authority_ready() AND xs_payroll_digest_ready()
 AND COALESCE((SELECT valid AND signature=xs_payroll_source_index_signature() FROM "PayrollSourceIndexMeta" WHERE singleton),false)
 AND (SELECT count(*) FROM pg_trigger WHERE tgrelid IN ('"CommonModel"'::regclass,'"CourseAppointment"'::regclass,'"LessonRecord"'::regclass,'"LessonRecordFact"'::regclass)
 AND tgname IN ('xs_payroll_source_index_row','xs_payroll_source_index_truncate') AND tgenabled='A')=8;
$$;
CREATE OR REPLACE FUNCTION xs_payroll_source_index_meta_guard() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN
 IF NEW.valid AND (TG_OP='INSERT' OR NEW IS DISTINCT FROM OLD) AND current_setting('xiaoshu.source_index_reseed',true) IS DISTINCT FROM pg_current_xact_id()::text THEN
  RAISE EXCEPTION 'source index requires a full reseed and new generation';
 END IF;
 RETURN NEW;
END $$;
DROP TRIGGER IF EXISTS xs_payroll_source_index_meta_guard ON "PayrollSourceIndexMeta";
CREATE TRIGGER xs_payroll_source_index_meta_guard BEFORE INSERT OR UPDATE ON "PayrollSourceIndexMeta" FOR EACH ROW EXECUTE FUNCTION xs_payroll_source_index_meta_guard();
ALTER TABLE "PayrollSourceIndexMeta" ENABLE ALWAYS TRIGGER xs_payroll_source_index_meta_guard;

-- Only a single target source is projected. Hashes and output are produced by the
-- same SELECT snapshot, never by a digest row which another trigger may update later.
CREATE OR REPLACE FUNCTION xs_payroll_source_index_target(co text,source_kind text,source_id text)
RETURNS TABLE(row_data jsonb,row_text text,input_signature text) LANGUAGE plpgsql VOLATILE SET jit=off AS $$
BEGIN
 IF source_kind='lesson' THEN
  RETURN QUERY WITH projected AS MATERIALIZED (SELECT
(to_jsonb(l)-'kzsj')||jsonb_build_object(
  'lessonObjectId',l."objectId",'legacyGeneralId',COALESCE(c."generalId",l."legacyGeneralId",0),
  'standardLessonKey',CASE WHEN COALESCE(c."generalId",l."legacyGeneralId",0)>0 THEN 'legacy:'||COALESCE(c."generalId",l."legacyGeneralId")::text ELSE 'lesson:'||l."objectId" END,
  'source','canonical','commonStatus',COALESCE(c."status",99),
  'lessonAt',COALESCE(l."lessonAt",xs_payroll_time(l."plpjsj"),xs_payroll_time(a."jssj"),l."sourceCreatedAt",c."createdAt"),
  'courseCategoryKey',COALESCE(NULLIF(l."courseCategoryKey",''),a."courseCategoryKey"),
  'courseCategoryName',COALESCE(NULLIF(l."courseCategoryName",''),a."courseCategoryName"),
  'teacherPaySnapshot',COALESCE(l."teacherPaySnapshot",a."teacherPaySnapshot"),
  'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(l."jffs",''),a."deliveryMode",a."jffs"),
  'teachingRuleId',a."teachingRuleId",'studentName',COALESCE(NULLIF(l."studentName",''),c."title"),
  'courseName',COALESCE(NULLIF(l."courseName",''),NULLIF(l."kcmc",''),c."subtitle")
 ) AS data,
 md5(jsonb_build_array(md5(l::text),CASE WHEN c."objectId" IS NOT NULL THEN md5(c::text) END,CASE WHEN a."objectId" IS NOT NULL THEN md5(a::text) END)::text) AS signature

 FROM "LessonRecord" l
 LEFT JOIN LATERAL (SELECT * FROM "CommonModel" c WHERE c."company"=co AND c."modelId"=59 AND c."itemId"=l."id" ORDER BY c."objectId" LIMIT 1) c ON true
 LEFT JOIN LATERAL (SELECT a.* FROM "CourseAppointment" a JOIN "CommonModel" ac ON ac."company"=co AND ac."modelId"=54 AND ac."itemId"=a."id" WHERE a."company"=co AND ac."generalId"::text=l."yyds"::text ORDER BY ac."objectId" LIMIT 1) a ON true
 WHERE l."company"=co AND l."objectId"=source_id) SELECT p.data,p.data::text,p.signature FROM projected p;
 ELSIF source_kind='fact' THEN
  RETURN QUERY WITH projected AS MATERIALIZED (SELECT
(to_jsonb(f)-'kzsj')||jsonb_build_object('lessonObjectId',f."objectId",'source','historical-fact','commonStatus',99,'courseCategoryKey',a."courseCategoryKey",'courseCategoryName',a."courseCategoryName",'teacherPaySnapshot',a."teacherPaySnapshot",'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(f."jffs",''),a."deliveryMode",a."jffs"),'teachingRuleId',a."teachingRuleId",
  'standardLessonKey',CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END) AS data,
 md5(jsonb_build_array(md5(f::text),CASE WHEN a."objectId" IS NOT NULL THEN md5(a::text) END)::text) AS signature

 FROM "LessonRecordFact" f LEFT JOIN LATERAL (SELECT a.* FROM "CourseAppointment" a JOIN "CommonModel" ac ON ac."company"=co AND ac."modelId"=54 AND ac."itemId"=a."id" WHERE a."company"=co AND ac."generalId"::text=f."yyds"::text ORDER BY ac."objectId" LIMIT 1) a ON true WHERE f."company"=co AND f."objectId"=source_id) SELECT p.data,p.data::text,p.signature FROM projected p;
 END IF;
END $$;

CREATE OR REPLACE FUNCTION xs_payroll_source_index_put(co text,source_kind text,source_id text) RETURNS text LANGUAGE plpgsql VOLATILE SET jit=off AS $$
DECLARE projected record; current_generation text;
BEGIN
 IF co IS NULL OR source_id IS NULL OR NOT pg_try_advisory_xact_lock(hashtext('payroll-source-index'),hashtext(co||':'||source_kind||':'||source_id)) THEN RETURN NULL;END IF;
 SELECT generation INTO current_generation FROM "PayrollSourceIndexMeta" WHERE singleton;
 SELECT * INTO projected FROM xs_payroll_source_index_target(co,source_kind,source_id);
 IF NOT FOUND THEN DELETE FROM "PayrollSourceIndex" WHERE company=co AND kind=source_kind AND "objectId"=source_id;RETURN NULL;END IF;
 INSERT INTO "PayrollSourceIndex"(kind,company,"objectId",row_data,row_text,input_signature,generation,selection) VALUES(source_kind,co,source_id,projected.row_data,projected.row_text,projected.input_signature,current_generation,xs_payroll_source_index_metadata(projected.row_data))
 ON CONFLICT(kind,company,"objectId") DO UPDATE SET row_data=EXCLUDED.row_data,row_text=EXCLUDED.row_text,input_signature=EXCLUDED.input_signature,generation=EXCLUDED.generation,selection=EXCLUDED.selection;
 RETURN projected.row_data->>'legacyGeneralId';
END $$;
-- Canonical changes alter fact visibility, not a fact's projection. The current
-- reader decides membership; keeping hidden fact entries is safe and avoids
-- rewriting historical facts for every lesson/comment update.
CREATE OR REPLACE FUNCTION xs_payroll_source_index_lesson(co text,source_id text) RETURNS void LANGUAGE plpgsql VOLATILE SET jit=off AS $$
BEGIN PERFORM xs_payroll_source_index_put(co,'lesson',source_id);END $$;
CREATE OR REPLACE FUNCTION xs_payroll_source_index_appointment(co text,appointment_key text) RETURNS void LANGUAGE plpgsql VOLATILE SET jit=off AS $$
DECLARE r record;
BEGIN
 FOR r IN SELECT 'lesson' AS kind,"objectId" FROM "LessonRecord" WHERE company=co AND yyds::text=appointment_key
 UNION ALL SELECT 'fact',"objectId" FROM "LessonRecordFact" WHERE company=co AND yyds::text=appointment_key ORDER BY kind,"objectId" LOOP
  PERFORM xs_payroll_source_index_put(co,r.kind,r."objectId");
 END LOOP;
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_source_index_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 item jsonb; r record; changes jsonb[]:=ARRAY[]::jsonb[]; seen text[]:=ARRAY[]::text[]; changed_key text;
BEGIN
 -- Review/model-60 synchronization can touch hundreds of thousands of rows.
 -- It cannot affect lesson projections: skip before catalog/meta reads or JSON.
 IF TG_TABLE_NAME='CommonModel' THEN
  IF TG_OP='INSERT' AND NOT COALESCE(NEW."modelId" IN (54,59),false) THEN RETURN NULL;END IF;
  IF TG_OP='DELETE' AND NOT COALESCE(OLD."modelId" IN (54,59),false) THEN RETURN NULL;END IF;
  IF TG_OP='UPDATE' AND NOT COALESCE(OLD."modelId" IN (54,59),false) AND NOT COALESCE(NEW."modelId" IN (54,59),false) THEN RETURN NULL;END IF;
 END IF;
 IF NOT EXISTS(SELECT 1 FROM "PayrollSourceIndexMeta" WHERE singleton AND valid) THEN RETURN NULL;END IF;
 IF TG_OP<>'INSERT' THEN changes:=array_append(changes,to_jsonb(OLD));END IF;
 IF TG_OP<>'DELETE' THEN changes:=array_append(changes,to_jsonb(NEW));END IF;
 FOREACH item IN ARRAY changes LOOP
  IF item->>'company' IS NULL THEN CONTINUE;END IF;
  changed_key:=CASE WHEN TG_TABLE_NAME IN ('LessonRecord','LessonRecordFact') THEN jsonb_build_array(item->>'company',item->>'objectId')::text
   WHEN TG_TABLE_NAME='CourseAppointment' THEN jsonb_build_array(item->>'company',item->>'id')::text
   WHEN (item->>'modelId')::numeric=59 THEN jsonb_build_array(item->>'company','59',item->>'itemId')::text
   ELSE jsonb_build_array(item->>'company',item->>'modelId',item->>'generalId')::text END;
  IF changed_key=ANY(seen) THEN CONTINUE;END IF;
  seen:=array_append(seen,changed_key);
  IF TG_TABLE_NAME='LessonRecord' THEN PERFORM xs_payroll_source_index_lesson(item->>'company',item->>'objectId');
  ELSIF TG_TABLE_NAME='LessonRecordFact' THEN PERFORM xs_payroll_source_index_put(item->>'company','fact',item->>'objectId');
  ELSIF TG_TABLE_NAME='CommonModel' AND (item->>'modelId')::numeric=59 THEN
   FOR r IN SELECT "objectId" FROM "LessonRecord" WHERE company=item->>'company' AND id=(item->>'itemId')::numeric ORDER BY "objectId" LOOP
    PERFORM xs_payroll_source_index_lesson(item->>'company',r."objectId");
   END LOOP;
  ELSIF TG_TABLE_NAME='CommonModel' AND (item->>'modelId')::numeric=54 THEN
   PERFORM xs_payroll_source_index_appointment(item->>'company',item->>'generalId');
  ELSIF TG_TABLE_NAME='CourseAppointment' THEN
   FOR r IN SELECT DISTINCT "generalId"::text AS key FROM "CommonModel" WHERE company=item->>'company' AND "modelId"=54 AND "itemId"=(item->>'id')::numeric ORDER BY key LOOP
    PERFORM xs_payroll_source_index_appointment(item->>'company',r.key);
   END LOOP;
  END IF;
 END LOOP;
 RETURN NULL;
 -- Projection upkeep must not prevent a real source write. The reader verifies
 -- every signature and recomputes missing/stale rows in its own current snapshot.
EXCEPTION WHEN OTHERS THEN RETURN NULL;
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_source_index_truncate() RETURNS trigger LANGUAGE plpgsql AS $$
BEGIN UPDATE "PayrollSourceIndexMeta" SET valid=false WHERE singleton;RETURN NULL;END $$;

CREATE OR REPLACE FUNCTION xs_payroll_source_index_scan(co text,reuse_index boolean,include_data boolean,mon text,coach_id text,appointment_ids text[],with_text boolean)
RETURNS TABLE(kind text,"objectId" text,row_data jsonb,row_text text,input_signature text,ordinal bigint,standard_key text,month_candidate_key text,scope_data jsonb)
LANGUAGE sql STABLE AS $$
 WITH common59 AS MATERIALIZED (
  SELECT DISTINCT ON (c."itemId") c.* FROM "CommonModel" c
  WHERE c.company=co AND c."modelId"=59
  ORDER BY c."itemId",c."objectId"
 ), appointment_map AS MATERIALIZED (
  -- Join before choosing: the earliest CommonModel can refer to a missing appointment.
  SELECT DISTINCT ON (ac."generalId"::text) ac."generalId"::text AS join_key,a AS appointment
  FROM "CommonModel" ac JOIN "CourseAppointment" a ON a.company=co AND a.id=ac."itemId"
  WHERE ac.company=co AND ac."modelId"=54
  ORDER BY ac."generalId"::text,ac."objectId"
 ), lessons AS MATERIALIZED (
  SELECT l AS lesson,row_number() OVER () AS ord FROM "LessonRecord" l WHERE l.company=co
 ), canonical_info AS MATERIALIZED (
  SELECT l.lesson,l.ord,c AS common,
   to_jsonb(COALESCE(c."generalId",(l.lesson)."legacyGeneralId",0))#>>'{}' AS legacy_text,
   CASE WHEN COALESCE(c."generalId",(l.lesson)."legacyGeneralId",0)>0
    THEN 'legacy:'||COALESCE(c."generalId",(l.lesson)."legacyGeneralId")::text
    ELSE 'lesson:'||(l.lesson)."objectId" END AS standard_key
  FROM lessons l LEFT JOIN common59 c ON c."itemId"=(l.lesson).id
 ), canonical_keys AS MATERIALIZED (
  SELECT legacy_text FROM canonical_info
 ), facts AS MATERIALIZED (
  SELECT f AS fact,row_number() OVER () AS ord,
   CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END AS standard_key
  FROM "LessonRecordFact" f WHERE f.company=co AND NOT EXISTS (
   SELECT 1 FROM canonical_keys k WHERE k.legacy_text=f."legacyGeneralId"::text AND COALESCE(f."legacyGeneralId",0)>0
  )
 ), ambiguous AS MATERIALIZED (
  SELECT EXISTS (SELECT 1 FROM (
   SELECT standard_key FROM canonical_info UNION ALL SELECT standard_key FROM facts
  ) k GROUP BY standard_key HAVING count(*)>1) AS value
 ), canonical_dependencies AS MATERIALIZED (
 SELECT x.*,am.appointment,
 md5(jsonb_build_array(COALESCE(ld.hash,md5((x.lesson)::text)),CASE WHEN (x.common)."objectId" IS NOT NULL THEN COALESCE(cd.hash,md5((x.common)::text)) END,CASE WHEN (am.appointment)."objectId" IS NOT NULL THEN COALESCE(ad.hash,md5((am.appointment)::text)) END)::text) AS signature
 FROM canonical_info x LEFT JOIN appointment_map am ON am.join_key=(x.lesson).yyds::text
 LEFT JOIN "PayrollReadDigest" ld ON ld.kind='lesson' AND ld.company=co AND ld."objectId"=(x.lesson)."objectId"
 LEFT JOIN "PayrollReadDigest" cd ON cd.kind='common' AND cd.model=59 AND cd.company=co AND cd."objectId"=(x.common)."objectId"
 LEFT JOIN "PayrollReadDigest" ad ON ad.kind='appointment' AND ad.company=co AND ad."objectId"=(am.appointment)."objectId"
 ), fact_dependencies AS MATERIALIZED (
 SELECT x.*,am.appointment,
 md5(jsonb_build_array(md5((x.fact)::text),CASE WHEN (am.appointment)."objectId" IS NOT NULL THEN COALESCE(ad.hash,md5((am.appointment)::text)) END)::text) AS signature
 FROM facts x LEFT JOIN appointment_map am ON am.join_key=(x.fact).yyds::text
 LEFT JOIN "PayrollReadDigest" ad ON ad.kind='appointment' AND ad.company=co AND ad."objectId"=(am.appointment)."objectId"
 ), projected AS (
 SELECT 'lesson'::text AS kind,l."objectId",0 AS source_group,x.ord,x.signature,
 CASE WHEN include_data AND (mon IS NULL OR m.selection->>'month_candidate_key' IS NULL OR m.selection->>'month_candidate_key'=mon) AND (coach_id IS NULL OR m.selection->>'coach_key' IS NULL OR m.selection->>'coach_key'=coach_id) AND (appointment_ids IS NULL OR m.selection->>'appointment_key' IS NULL OR m.selection->>'appointment_key'=ANY(appointment_ids)) THEN CASE WHEN i.input_signature=x.signature THEN i.row_data ELSE
(to_jsonb(l)-'kzsj')||jsonb_build_object(
  'lessonObjectId',l."objectId",'legacyGeneralId',COALESCE(c."generalId",l."legacyGeneralId",0),
  'standardLessonKey',CASE WHEN COALESCE(c."generalId",l."legacyGeneralId",0)>0 THEN 'legacy:'||COALESCE(c."generalId",l."legacyGeneralId")::text ELSE 'lesson:'||l."objectId" END,
  'source','canonical','commonStatus',COALESCE(c."status",99),
  'lessonAt',COALESCE(l."lessonAt",xs_payroll_time(l."plpjsj"),xs_payroll_time(a."jssj"),l."sourceCreatedAt",c."createdAt"),
  'courseCategoryKey',COALESCE(NULLIF(l."courseCategoryKey",''),a."courseCategoryKey"),
  'courseCategoryName',COALESCE(NULLIF(l."courseCategoryName",''),a."courseCategoryName"),
  'teacherPaySnapshot',COALESCE(l."teacherPaySnapshot",a."teacherPaySnapshot"),
  'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(l."jffs",''),a."deliveryMode",a."jffs"),
  'teachingRuleId',a."teachingRuleId",'studentName',COALESCE(NULLIF(l."studentName",''),c."title"),
  'courseName',COALESCE(NULLIF(l."courseName",''),NULLIF(l."kcmc",''),c."subtitle")
 ) END END AS data,
 CASE WHEN with_text THEN CASE WHEN i.input_signature=x.signature THEN i.row_text||'' ELSE ((to_jsonb(l)-'kzsj')||jsonb_build_object(
  'lessonObjectId',l."objectId",'legacyGeneralId',COALESCE(c."generalId",l."legacyGeneralId",0),
  'standardLessonKey',CASE WHEN COALESCE(c."generalId",l."legacyGeneralId",0)>0 THEN 'legacy:'||COALESCE(c."generalId",l."legacyGeneralId")::text ELSE 'lesson:'||l."objectId" END,
  'source','canonical','commonStatus',COALESCE(c."status",99),
  'lessonAt',COALESCE(l."lessonAt",xs_payroll_time(l."plpjsj"),xs_payroll_time(a."jssj"),l."sourceCreatedAt",c."createdAt"),
  'courseCategoryKey',COALESCE(NULLIF(l."courseCategoryKey",''),a."courseCategoryKey"),
  'courseCategoryName',COALESCE(NULLIF(l."courseCategoryName",''),a."courseCategoryName"),
  'teacherPaySnapshot',COALESCE(l."teacherPaySnapshot",a."teacherPaySnapshot"),
  'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(l."jffs",''),a."deliveryMode",a."jffs"),
  'teachingRuleId',a."teachingRuleId",'studentName',COALESCE(NULLIF(l."studentName",''),c."title"),
  'courseName',COALESCE(NULLIF(l."courseName",''),NULLIF(l."kcmc",''),c."subtitle")
 ))::text END END AS text_value,m.selection
 FROM canonical_dependencies x CROSS JOIN LATERAL (SELECT (x.lesson).*) l
 CROSS JOIN LATERAL (SELECT (x.common).*) c CROSS JOIN LATERAL (SELECT (x.appointment).*) a
 LEFT JOIN "PayrollSourceIndex" i ON reuse_index AND i.kind='lesson' AND i.company=co AND i."objectId"=l."objectId" AND i.generation=(SELECT generation FROM "PayrollSourceIndexMeta" WHERE singleton)
 CROSS JOIN LATERAL (SELECT CASE WHEN i.input_signature=x.signature THEN i.selection ELSE xs_payroll_source_index_metadata((to_jsonb(l)-'kzsj')||jsonb_build_object(
  'lessonObjectId',l."objectId",'legacyGeneralId',COALESCE(c."generalId",l."legacyGeneralId",0),
  'standardLessonKey',CASE WHEN COALESCE(c."generalId",l."legacyGeneralId",0)>0 THEN 'legacy:'||COALESCE(c."generalId",l."legacyGeneralId")::text ELSE 'lesson:'||l."objectId" END,
  'source','canonical','commonStatus',COALESCE(c."status",99),
  'lessonAt',COALESCE(l."lessonAt",xs_payroll_time(l."plpjsj"),xs_payroll_time(a."jssj"),l."sourceCreatedAt",c."createdAt"),
  'courseCategoryKey',COALESCE(NULLIF(l."courseCategoryKey",''),a."courseCategoryKey"),
  'courseCategoryName',COALESCE(NULLIF(l."courseCategoryName",''),a."courseCategoryName"),
  'teacherPaySnapshot',COALESCE(l."teacherPaySnapshot",a."teacherPaySnapshot"),
  'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(l."jffs",''),a."deliveryMode",a."jffs"),
  'teachingRuleId',a."teachingRuleId",'studentName',COALESCE(NULLIF(l."studentName",''),c."title"),
  'courseName',COALESCE(NULLIF(l."courseName",''),NULLIF(l."kcmc",''),c."subtitle")
 )) END AS selection OFFSET 0) m
 WHERE NOT (SELECT value FROM ambiguous)
 UNION ALL
 SELECT 'fact',f."objectId",1,x.ord,x.signature,
 CASE WHEN include_data AND (mon IS NULL OR m.selection->>'month_candidate_key' IS NULL OR m.selection->>'month_candidate_key'=mon) AND (coach_id IS NULL OR m.selection->>'coach_key' IS NULL OR m.selection->>'coach_key'=coach_id) AND (appointment_ids IS NULL OR m.selection->>'appointment_key' IS NULL OR m.selection->>'appointment_key'=ANY(appointment_ids)) THEN CASE WHEN i.input_signature=x.signature THEN i.row_data ELSE
(to_jsonb(f)-'kzsj')||jsonb_build_object('lessonObjectId',f."objectId",'source','historical-fact','commonStatus',99,'courseCategoryKey',a."courseCategoryKey",'courseCategoryName',a."courseCategoryName",'teacherPaySnapshot',a."teacherPaySnapshot",'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(f."jffs",''),a."deliveryMode",a."jffs"),'teachingRuleId',a."teachingRuleId",
  'standardLessonKey',CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END) END END,
 CASE WHEN with_text THEN CASE WHEN i.input_signature=x.signature THEN i.row_text||'' ELSE ((to_jsonb(f)-'kzsj')||jsonb_build_object('lessonObjectId',f."objectId",'source','historical-fact','commonStatus',99,'courseCategoryKey',a."courseCategoryKey",'courseCategoryName',a."courseCategoryName",'teacherPaySnapshot',a."teacherPaySnapshot",'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(f."jffs",''),a."deliveryMode",a."jffs"),'teachingRuleId',a."teachingRuleId",
  'standardLessonKey',CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END))::text END END,m.selection
 FROM fact_dependencies x CROSS JOIN LATERAL (SELECT (x.fact).*) f CROSS JOIN LATERAL (SELECT (x.appointment).*) a
 LEFT JOIN "PayrollSourceIndex" i ON reuse_index AND i.kind='fact' AND i.company=co AND i."objectId"=f."objectId" AND i.generation=(SELECT generation FROM "PayrollSourceIndexMeta" WHERE singleton)
 CROSS JOIN LATERAL (SELECT CASE WHEN i.input_signature=x.signature THEN i.selection ELSE xs_payroll_source_index_metadata((to_jsonb(f)-'kzsj')||jsonb_build_object('lessonObjectId',f."objectId",'source','historical-fact','commonStatus',99,'courseCategoryKey',a."courseCategoryKey",'courseCategoryName',a."courseCategoryName",'teacherPaySnapshot',a."teacherPaySnapshot",'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(f."jffs",''),a."deliveryMode",a."jffs"),'teachingRuleId',a."teachingRuleId",
  'standardLessonKey',CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END)) END AS selection OFFSET 0) m
 WHERE NOT (SELECT value FROM ambiguous)
 UNION ALL
 SELECT CASE WHEN s.row_data->>'source'='canonical' THEN 'lesson' ELSE 'fact' END,s.row_data->>'lessonObjectId',0,s.ord,NULL,CASE WHEN include_data THEN s.row_data END,CASE WHEN with_text THEN s.row_data::text END,xs_payroll_source_index_metadata(s.row_data)
 FROM ambiguous guard CROSS JOIN LATERAL xs_payroll_sources(co) WITH ORDINALITY AS s(row_data,ord) WHERE guard.value
 ) SELECT p.kind,p."objectId",p.data,p.text_value,p.signature,CASE WHEN include_data THEN CASE WHEN p.source_group=0 THEN p.ord ELSE (SELECT count(*) FROM lessons)+p.ord END END,p.selection->>'standard_key',p.selection->>'month_candidate_key',p.selection->'scope_data'
 FROM projected p;
$$;
CREATE OR REPLACE FUNCTION xs_payroll_source_index_entries(co text,reuse_index boolean,include_data boolean)
RETURNS TABLE(kind text,"objectId" text,row_data jsonb,row_text text,input_signature text,ordinal bigint)
LANGUAGE sql STABLE AS $$
 SELECT e.kind,e."objectId",e.row_data,e.row_text,e.input_signature,e.ordinal
 FROM xs_payroll_source_index_scan(co,reuse_index,include_data,NULL,NULL,NULL,true) e
 ORDER BY CASE WHEN include_data THEN e.ordinal END;
$$;
-- Compatibility path keeps complete ordered rows. The three-argument function
-- can inline at a version-only call, eliminating JSON and ordinal-sort branches.
CREATE OR REPLACE FUNCTION xs_payroll_source_index_entries(co text,reuse_index boolean DEFAULT true)
RETURNS TABLE(kind text,"objectId" text,row_data jsonb,row_text text,input_signature text,ordinal bigint)
LANGUAGE sql STABLE AS $$ SELECT * FROM xs_payroll_source_index_entries(co,reuse_index,true) $$;
CREATE OR REPLACE FUNCTION xs_payroll_source_entries(co text)
RETURNS TABLE(row_data jsonb,row_text text,ordinal bigint) LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
BEGIN
 IF NOT xs_payroll_source_index_ready() OR EXISTS(SELECT 1 FROM "CourseAppointment" WHERE company=co AND id IS NOT NULL GROUP BY id HAVING count(*)>1) THEN
  RETURN QUERY SELECT s.data,s.data::text,s.ord FROM xs_payroll_sources(co) WITH ORDINALITY s(data,ord);RETURN;
 END IF;
 RETURN QUERY SELECT e.row_data,e.row_text,e.ordinal FROM xs_payroll_source_index_entries(co,true,true) e;
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_sources_indexed(co text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$
 SELECT row_data FROM xs_payroll_source_entries(co) ORDER BY ordinal;
$$;
-- Every current source contributes narrow metadata and (optionally) exact text.
-- Only selected candidates fetch their full JSON. Callers apply their final exact
-- financial filters; unknown metadata and ambiguous source identities fail open
-- for transport inclusion, while authority/schema failures use original sources.
CREATE OR REPLACE FUNCTION xs_payroll_source_entries_scoped(co text,mon text DEFAULT NULL,coach_id text DEFAULT NULL,appointment_ids text[] DEFAULT NULL,with_text boolean DEFAULT true)
RETURNS TABLE(row_data jsonb,row_text text,ordinal bigint,standard_key text,month_candidate_key text,scope_data jsonb)
LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
BEGIN
 IF NOT xs_payroll_source_index_ready() OR EXISTS(SELECT 1 FROM "CourseAppointment" WHERE company=co AND id IS NOT NULL GROUP BY id HAVING count(*)>1) THEN
  RETURN QUERY SELECT s.data,CASE WHEN with_text THEN s.data::text END,s.ord,s.data->>'standardLessonKey',m.selection->>'month_candidate_key',m.selection->'scope_data'
  FROM xs_payroll_sources(co) WITH ORDINALITY s(data,ord)
  CROSS JOIN LATERAL (SELECT xs_payroll_source_index_metadata(s.data) AS selection) m;
  RETURN;
 END IF;
 RETURN QUERY SELECT e.row_data,e.row_text,e.ordinal,e.standard_key,e.month_candidate_key,e.scope_data
 FROM xs_payroll_source_index_scan(co,true,true,mon,coach_id,appointment_ids,with_text) e;
END $$;

CREATE OR REPLACE FUNCTION xs_payroll_source_version_indexed(co text) RETURNS text LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
BEGIN
 IF NOT xs_payroll_source_index_ready() OR EXISTS(SELECT 1 FROM "CourseAppointment" WHERE company=co AND id IS NOT NULL GROUP BY id HAVING count(*)>1) THEN RETURN xs_payroll_source_version(co);END IF;
 RETURN (
 SELECT md5(COALESCE((SELECT string_agg(row_text,'|' ORDER BY row_text) FROM xs_payroll_source_index_entries(co,true,false)),'')||
 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 $$;

CREATE OR REPLACE FUNCTION xs_payroll_source_index_rebuild() RETURNS void LANGUAGE plpgsql SET jit=off SET work_mem='64MB' AS $$
DECLARE new_generation text:=md5(clock_timestamp()::text||random()::text); co record; prior_setting text:=current_setting('xiaoshu.source_index_reseed',true);
BEGIN
 IF NOT xs_payroll_authority_ready() OR NOT xs_payroll_digest_ready() THEN RAISE EXCEPTION 'authoritative source and digest prerequisites required for source index rebuild';END IF;
 LOCK TABLE "CommonModel","CourseAppointment","LessonRecord","LessonRecordFact" IN SHARE ROW EXCLUSIVE MODE;
 PERFORM set_config('xiaoshu.source_index_reseed',pg_current_xact_id()::text,true);
 INSERT INTO "PayrollSourceIndexMeta" VALUES(true,new_generation,'',false) ON CONFLICT(singleton) DO UPDATE SET generation=EXCLUDED.generation,signature='',valid=false;
 DELETE FROM "PayrollSourceIndex";
 FOR co IN SELECT DISTINCT company FROM "LessonRecord" WHERE company IS NOT NULL UNION SELECT DISTINCT company FROM "LessonRecordFact" WHERE company IS NOT NULL LOOP
  INSERT INTO "PayrollSourceIndex"(kind,company,"objectId",row_data,row_text,input_signature,generation,selection) SELECT e.kind,co.company,e."objectId",e.row_data,e.row_text,e.input_signature,new_generation,xs_payroll_source_index_metadata(e.row_data) FROM xs_payroll_source_index_entries(co.company,false,true) e;
 END LOOP;
 UPDATE "PayrollSourceIndexMeta" SET signature=xs_payroll_source_index_signature(),valid=true WHERE singleton;
 PERFORM set_config('xiaoshu.source_index_reseed',COALESCE(prior_setting,''),true);
END $$;
DO $$ DECLARE tab text;BEGIN
 FOREACH tab IN ARRAY ARRAY['CommonModel','CourseAppointment','LessonRecord','LessonRecordFact'] LOOP
  EXECUTE format('LOCK TABLE %I IN SHARE ROW EXCLUSIVE MODE',tab);
  EXECUTE format('DROP TRIGGER IF EXISTS xs_payroll_source_index_row ON %I',tab);
  EXECUTE format('DROP TRIGGER IF EXISTS xs_payroll_source_index_truncate ON %I',tab);
  EXECUTE format('CREATE TRIGGER xs_payroll_source_index_row AFTER INSERT OR UPDATE OR DELETE ON %I FOR EACH ROW EXECUTE FUNCTION xs_payroll_source_index_row()',tab);
  EXECUTE format('CREATE TRIGGER xs_payroll_source_index_truncate AFTER TRUNCATE ON %I FOR EACH STATEMENT EXECUTE FUNCTION xs_payroll_source_index_truncate()',tab);
  EXECUTE format('ALTER TABLE %I ENABLE ALWAYS TRIGGER xs_payroll_source_index_row',tab);
  EXECUTE format('ALTER TABLE %I ENABLE ALWAYS TRIGGER xs_payroll_source_index_truncate',tab);
 END LOOP;
END $$;
SELECT xs_payroll_source_index_rebuild();
COMMIT;