-- Original source projection retained verbatim as the authoritative fallback.
CREATE OR REPLACE FUNCTION public.xs_payroll_sources_uncached(co text)
 RETURNS SETOF jsonb
 LANGUAGE sql
 STABLE
 SET jit TO 'off'
AS $function$
 WITH canonical AS (
 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 row_data
 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
 ), fallback AS (
 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 row_data
 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 NOT EXISTS(SELECT 1 FROM canonical c WHERE c.row_data->>'legacyGeneralId'=f."legacyGeneralId"::text AND COALESCE(f."legacyGeneralId",0)>0)
 ) SELECT row_data FROM canonical UNION ALL SELECT row_data FROM fallback
$function$;

CREATE OR REPLACE FUNCTION xs_payroll_sources_current(co text,saved text) RETURNS boolean LANGUAGE sql STABLE AS $$
 SELECT saved IS NOT NULL AND NOT EXISTS(SELECT 1 FROM "PayrollReadChangeScope" c WHERE c.company=co AND c.scope IN ('common:54','common:59','appointment','lesson') 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_source_stamp(co text) RETURNS text LANGUAGE plpgsql SET jit=off AS $$
DECLARE facts text; meta text; generation 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 generation FROM "PayrollReadDigestMeta";
 SELECT md5(COALESCE(string_agg(xmin::text||':'||ctid::text,',' ORDER BY ctid),'')) INTO facts FROM "LessonRecordFact" WHERE company=co;
 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','LessonRecordFact','PayrollReadChangeScope');
 RETURN md5(generation||'|'||facts||'|'||meta||'|'||pg_get_functiondef('xs_payroll_sources_uncached(text)'::regprocedure)||'|'||pg_get_functiondef('xs_payroll_time(text)'::regprocedure)||'|'||pg_get_functiondef('xs_payroll_sources_current(text,text)'::regprocedure)||'|'||pg_get_functiondef('xs_payroll_source_stamp(text)'::regprocedure)||'|'||current_setting('TimeZone')||'|'||current_setting('DateStyle')||'|'||current_setting('extra_float_digits'));
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_sources_cached(co text) RETURNS SETOF jsonb LANGUAGE plpgsql SET jit=off AS $$
DECLARE signature text; rows_json jsonb; may_publish boolean; read_snapshot text:=pg_current_snapshot()::text;
BEGIN
 IF NOT xs_payroll_authority_ready() THEN RETURN QUERY SELECT s FROM xs_payroll_sources(co) s;RETURN;END IF;
 signature:=xs_payroll_source_stamp(co);
 SELECT payload->'rows' INTO rows_json FROM "PayrollVerifiedReadCache" WHERE company=co AND key='canonical-sources-v1' AND stamp=signature AND xs_payroll_sources_current(co,payload->>'snapshot');
 IF rows_json IS NULL THEN
  may_publish:=pg_try_advisory_xact_lock(hashtext('xs-payroll-source-cache:'||co));
  SELECT COALESCE(jsonb_agg(s),'[]'::jsonb) INTO rows_json FROM xs_payroll_sources_uncached(co) s;
  IF may_publish AND signature=xs_payroll_source_stamp(co) AND xs_payroll_sources_current(co,read_snapshot) THEN
   INSERT INTO "PayrollVerifiedReadCache" VALUES(co,'canonical-sources-v1',signature,jsonb_build_object('snapshot',read_snapshot,'rows',rows_json),now()) ON CONFLICT(company,key) DO UPDATE SET stamp=EXCLUDED.stamp,payload=EXCLUDED.payload,"createdAt"=EXCLUDED."createdAt";
  END IF;
 END IF;
 RETURN QUERY SELECT value FROM jsonb_array_elements(rows_json);
END $$;
CREATE OR REPLACE FUNCTION public.xs_payroll_source_version_uncached_fast(co text)
 RETURNS text
 LANGUAGE sql
 VOLATILE
 SET jit TO 'off'
AS $function$
 SELECT md5(COALESCE((SELECT string_agg(s::text,'|' ORDER BY s::text) FROM xs_payroll_sources_cached(co) s),'')||
 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'),''))
$function$;

CREATE OR REPLACE FUNCTION xs_payroll_source_version_stamp(co text) RETURNS text LANGUAGE plpgsql SET jit=off AS $$
DECLARE signature text:=xs_payroll_source_stamp(co); tab text; v text; meta text;
BEGIN
 IF signature IS NULL THEN RETURN NULL;END IF;
 FOREACH tab IN ARRAY ARRAY['TeachingCompensationRule','CourseTeachingProfile','CoachPayRate','ReviewReward','ReviewRewardCorrection'] LOOP
  EXECUTE format('SELECT md5(COALESCE(string_agg(xmin::text||'':''||ctid::text,'','' ORDER BY ctid),'''')) FROM %I WHERE company=$1',tab) INTO v USING co;
  SELECT md5(COALESCE(string_agg(a.attnum||':'||a.attname||':'||a.atttypid||':'||a.atttypmod,'|' ORDER BY a.attnum),'')) INTO meta FROM pg_attribute a WHERE a.attrelid=to_regclass(format('public.%I',tab)) AND a.attnum>0 AND NOT a.attisdropped;
  signature:=signature||'|'||tab||':'||v||':'||meta;
 END LOOP;
 RETURN md5(signature||'|'||pg_get_viewdef('"ReviewRewardIncome"'::regclass,true)||'|'||pg_get_functiondef('xs_payroll_source_version_uncached_fast(text)'::regprocedure)||'|'||pg_get_functiondef('xs_payroll_source_version_stamp(text)'::regprocedure)||'|'||current_setting('IntervalStyle')||'|'||current_setting('bytea_output'));
END $$;
CREATE OR REPLACE FUNCTION xs_payroll_source_version_fast(co text) RETURNS text LANGUAGE plpgsql SET jit=off AS $$
DECLARE signature text; result text; may_publish boolean; read_snapshot text:=pg_current_snapshot()::text;
BEGIN
 IF NOT xs_payroll_authority_ready() THEN RETURN xs_payroll_source_version(co);END IF;
 signature:=xs_payroll_source_version_stamp(co);
 SELECT payload->>'version' INTO result FROM "PayrollVerifiedReadCache" WHERE company=co AND key='source-version-v2' AND stamp=signature AND xs_payroll_sources_current(co,payload->>'snapshot');
 IF result IS NOT NULL THEN RETURN result;END IF;
 may_publish:=pg_try_advisory_xact_lock(hashtext('xs-payroll-version-cache:'||co));
 result:=xs_payroll_source_version_uncached_fast(co);
 IF may_publish AND signature=xs_payroll_source_version_stamp(co) AND xs_payroll_sources_current(co,read_snapshot) THEN INSERT INTO "PayrollVerifiedReadCache" VALUES(co,'source-version-v2',signature,jsonb_build_object('snapshot',read_snapshot,'version',result),now()) ON CONFLICT(company,key) DO UPDATE SET stamp=EXCLUDED.stamp,payload=EXCLUDED.payload,"createdAt"=EXCLUDED."createdAt";END IF;
 RETURN result;
END $$;