-- Recompute authoritative source rows directly; no result cache or derived tables.
-- Bulk lookup maps eliminate one common/appointment lookup per lesson.
CREATE OR REPLACE FUNCTION xs_payroll_sources_bulk(co text)
RETURNS SETOF jsonb LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
BEGIN
 IF NOT xs_payroll_authority_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 FROM xs_payroll_sources(co) s;
  RETURN;
 END IF;
 RETURN QUERY
 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
 ), output AS (
 SELECT 0 AS source_group,x.ord,
 (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 canonical_info x
 CROSS JOIN LATERAL (SELECT (x.lesson).*) l
 CROSS JOIN LATERAL (SELECT (x.common).*) c
 LEFT JOIN appointment_map am ON am.join_key=l."yyds"::text
 LEFT JOIN LATERAL (SELECT (am.appointment).*) a ON true
 WHERE NOT (SELECT value FROM ambiguous)
 UNION ALL
 SELECT 1 AS source_group,x.ord,
 (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 facts x CROSS JOIN LATERAL (SELECT (x.fact).*) f
 LEFT JOIN appointment_map am ON am.join_key=f."yyds"::text
 LEFT JOIN LATERAL (SELECT (am.appointment).*) a ON true
 WHERE NOT (SELECT value FROM ambiguous)
 UNION ALL
 -- A changed join order must never choose a different duplicate representative.
 SELECT 0,s.ord,s.row_data FROM ambiguous guard
 CROSS JOIN LATERAL xs_payroll_sources(co) WITH ORDINALITY AS s(row_data,ord)
 WHERE guard.value
 ) SELECT output.row_data FROM output ORDER BY output.source_group,output.ord;
END $$;