xiaoshu-app/scripts/sql/legacy-review-sync.sql

-- Apply after legacy-direct-sync.sql and review-reconciliation.sql.
-- Source-owned versions are separate from human transfers; original receipts stay immutable.
BEGIN;
SET LOCAL lock_timeout='10s';
CREATE TABLE IF NOT EXISTS "LegacyReviewSyncState" (
 company text NOT NULL,"taskId" text NOT NULL,"assignmentVersion" numeric NOT NULL,
 fingerprint text NOT NULL,"pendingIncome" boolean NOT NULL DEFAULT false,
 "lastError" text NOT NULL DEFAULT '',"updatedAt" timestamptz NOT NULL DEFAULT now(),
 PRIMARY KEY(company,"taskId")
);
CREATE INDEX IF NOT EXISTS legacy_review_income_pending ON "LegacyReviewSyncState"(company,"taskId") WHERE "pendingIncome";
CREATE TABLE IF NOT EXISTS "LegacyReviewSyncAudit" (
 "objectId" text PRIMARY KEY,company text NOT NULL,"taskId" text NOT NULL,
 before_data jsonb NOT NULL,after_data jsonb NOT NULL,"createdAt" timestamptz NOT NULL DEFAULT now()
);
REVOKE ALL ON "LegacyReviewSyncState","LegacyReviewSyncAudit" FROM PUBLIC;

CREATE OR REPLACE FUNCTION xs_legacy_review_owned(co text, task_id text) RETURNS boolean LANGUAGE sql STABLE AS $$
 SELECT COALESCE((COALESCE(m."assignmentVersion",0)=0 OR EXISTS(
  SELECT 1 FROM "LegacyReviewSyncState" s WHERE s.company=m.company AND s."taskId"=m."objectId"
  AND s."assignmentVersion"=m."assignmentVersion" AND s.fingerprint=md5(to_jsonb(m)::text)))
  AND COALESCE(m."sourceKey",'') LIKE 'legacy:model:60:general:%'
  AND m.fxzt::text<>'1' AND NOT EXISTS(SELECT 1 FROM "ReviewReward" r WHERE r.company=m.company AND r."taskId"=m."objectId"),false)
 FROM "MemoryPracticeRecord" m WHERE m.company=co AND m."objectId"=task_id
$$;

CREATE OR REPLACE FUNCTION xs_legacy_review_track(co text, task_id text, expected_version numeric DEFAULT 0) RETURNS void LANGUAGE plpgsql AS $$
DECLARE m jsonb; c jsonb; pending boolean;
BEGIN
 SELECT to_jsonb(t) INTO m FROM "MemoryPracticeRecord" t WHERE t.company=co AND t."objectId"=task_id FOR UPDATE;
 SELECT to_jsonb(t) INTO c FROM "CommonModel" t WHERE t.company=co AND t."modelId"=60 AND t."itemId"=(m->>'id')::numeric;
 IF COALESCE((m->>'assignmentVersion')::numeric,0)<>expected_version THEN RETURN; END IF;
 IF m IS NULL OR c IS NULL OR m->>'sourceKey' NOT LIKE 'legacy:model:60:general:%' THEN RAISE EXCEPTION '旧复习来源不完整'; END IF;
 SELECT EXISTS(SELECT 1 FROM "ReviewRewardIncome" r WHERE r.company=co AND r."taskId"=task_id AND r."incomeMonth">='2026-09' AND r.status IN ('historical','exception')) INTO pending;
 INSERT INTO "LegacyReviewSyncState"(company,"taskId","assignmentVersion",fingerprint,"pendingIncome")
 VALUES(co,task_id,COALESCE((m->>'assignmentVersion')::numeric,0),md5(m::text),pending)
 ON CONFLICT(company,"taskId") DO UPDATE SET "assignmentVersion"=EXCLUDED."assignmentVersion",fingerprint=EXCLUDED.fingerprint,"pendingIncome"=EXCLUDED."pendingIncome","lastError"='',"updatedAt"=now();
 INSERT INTO "LegacyDirectBaseline"(company,model,gid,fingerprint) VALUES(co,60,(c->>'generalId')::numeric,md5(jsonb_build_array(c,m)::text))
 ON CONFLICT(company,model,gid) DO UPDATE SET fingerprint=EXCLUDED.fingerprint,"updatedAt"=now();
END $$;

-- Bootstrap only records whose complete business fields still match the live old source.
-- Once a baseline exists, an unexplained local modification can never be re-adopted.
CREATE OR REPLACE FUNCTION xs_legacy_review_observe(co text, source_rows jsonb) RETURNS integer LANGUAGE plpgsql AS $$
DECLARE row_data record; n integer:=0;
BEGIN
 IF jsonb_typeof(source_rows)<>'array' OR jsonb_array_length(source_rows)>1000 THEN RAISE EXCEPTION '无效旧复习来源页'; END IF;
 FOR row_data IN SELECT m."objectId" FROM jsonb_array_elements(source_rows) s
  JOIN "CommonModel" c ON c.company=co AND c."modelId"=60 AND c."generalId"=(s->>'GeneralID')::numeric
  JOIN "MemoryPracticeRecord" m ON m.company=c.company AND m.id=c."itemId"
  LEFT JOIN "LegacyDirectBaseline" b ON b.company=c.company AND b.model=60 AND b.gid=c."generalId"
  WHERE m.fxzt='0' AND COALESCE(m."assignmentVersion",0)=0 AND m."sourceKey" LIKE 'legacy:model:60:general:%'
  AND NOT COALESCE(c."teachingManaged",false) AND COALESCE(c.status,99)<>-2
  AND NOT EXISTS(SELECT 1 FROM "ReviewReward" r WHERE r.company=m.company AND r."taskId"=m."objectId")
  AND NOT EXISTS(SELECT 1 FROM "LegacyReviewSyncState" t WHERE t.company=m.company AND t."taskId"=m."objectId")
  AND (b.fingerprint IS NULL OR b.fingerprint=md5(jsonb_build_array(to_jsonb(c),to_jsonb(m))::text))
  AND jsonb_build_array(m.yhid::text,m.plid::text,m.xxjlid::text,m."orderId"::text,m.kcid::text,COALESCE(m.fxzt,''),COALESCE(m.wcsj,''),COALESCE(m.kywrq,''),COALESCE(m.kywsj,''))=
      jsonb_build_array(s->>'yhid',s->>'plid',s->>'xxjlid',s->>'orderId',s->>'kcid',COALESCE(s->>'fxzt',''),COALESCE(s->>'wcsj',''),COALESCE(s->>'kywrq',''),COALESCE(s->>'kywsj',''))
  FOR UPDATE OF m,c
 LOOP PERFORM xs_legacy_review_track(co,row_data."objectId");n:=n+1;END LOOP;
 RETURN n;
END $$;
REVOKE ALL ON FUNCTION xs_legacy_review_observe(text,jsonb) FROM PUBLIC;

CREATE OR REPLACE FUNCTION xs_legacy_review_patch(co text, task_id text, values_json jsonb, expected jsonb) RETURNS jsonb LANGUAGE plpgsql AS $$
DECLARE m "MemoryPracticeRecord"; before_row jsonb; after_row jsonb; c jsonb; incoming jsonb; f text; baseline text;
 prior text:=COALESCE(current_setting('xiaoshu.review_transfer',true),'');
BEGIN
 SELECT * INTO m FROM "MemoryPracticeRecord" WHERE company=co AND "objectId"=task_id FOR UPDATE;
 IF NOT FOUND THEN RETURN jsonb_build_object('applied',false,'reason','复习任务不存在'); END IF;
 before_row:=to_jsonb(m);
 IF before_row IS DISTINCT FROM expected OR NOT COALESCE(xs_legacy_review_owned(co,task_id),false) THEN RETURN jsonb_build_object('applied',false,'reason','保留新系统转交、完成或计薪凭据'); END IF;
 IF jsonb_typeof(values_json)<>'object' OR EXISTS(SELECT 1 FROM jsonb_object_keys(values_json) k WHERE k NOT IN ('plid','fxzt','wcsj')) THEN RETURN jsonb_build_object('applied',false,'reason','保留复习来源身份和计划'); END IF;
 SELECT to_jsonb(t) INTO c FROM "CommonModel" t WHERE t.company=co AND t."modelId"=60 AND t."itemId"=m.id FOR UPDATE;
 IF c IS NULL OR COALESCE((c->>'teachingManaged')::boolean,false) OR c->>'status'='-2' THEN RETURN jsonb_build_object('applied',false,'reason','来源主表受保护'); END IF;
 SELECT fingerprint INTO baseline FROM "LegacyDirectBaseline" WHERE company=co AND model=60 AND gid=(c->>'generalId')::numeric;
 IF (baseline IS NOT NULL AND baseline<>md5(jsonb_build_array(c,before_row)::text)) OR
 (baseline IS NULL AND (m."createdAt" IS NULL OR c->>'createdAt' IS NULL OR m."createdAt" IS DISTINCT FROM m."updatedAt" OR c->>'createdAt' IS DISTINCT FROM c->>'updatedAt')) THEN
  RETURN jsonb_build_object('applied',false,'reason','新系统已有修改或来源基线不可确认');
 END IF;
 incoming:=before_row||values_json;
 IF incoming->>'fxzt' NOT IN ('0','1') OR NULLIF(incoming->>'plid','') IS NULL THEN RAISE EXCEPTION '无效复习来源'; END IF;
 IF incoming->>'plid' IS DISTINCT FROM before_row->>'plid' THEN PERFORM xs_review_target(co,incoming->>'plid'); END IF;
 IF incoming->>'fxzt'='1' AND xiaoshu_safe_shanghai_timestamp(incoming->>'wcsj') IS NULL THEN RETURN jsonb_build_object('applied',false,'reason','缺少旧系统完成时间'); END IF;
 PERFORM set_config('xiaoshu.review_transfer','legacy-sync',true);
 UPDATE "MemoryPracticeRecord" SET plid=(incoming->>'plid')::numeric,fxzt=incoming->>'fxzt',wcsj=COALESCE(incoming->>'wcsj',''),
 "assignmentVersion"=COALESCE("assignmentVersion",0)+CASE WHEN plid::text IS DISTINCT FROM incoming->>'plid' THEN 1 ELSE 0 END,"updatedAt"=clock_timestamp()
 WHERE company=co AND "objectId"=task_id RETURNING to_jsonb("MemoryPracticeRecord".*) INTO after_row;
 PERFORM set_config('xiaoshu.review_transfer',prior,true);
 INSERT INTO "LegacyReviewSyncAudit"("objectId",company,"taskId",before_data,after_data)
 VALUES(md5(co||task_id||before_row::text||after_row::text),co,task_id,before_row,after_row) ON CONFLICT DO NOTHING;
 PERFORM xs_legacy_review_track(co,task_id,(after_row->>'assignmentVersion')::numeric);
 RETURN jsonb_build_object('applied',true);
EXCEPTION WHEN OTHERS THEN PERFORM set_config('xiaoshu.review_transfer',prior,true);RAISE;
END $$;

CREATE OR REPLACE FUNCTION xs_legacy_review_income(co text, actor text, batch_limit integer DEFAULT 100, after_task text DEFAULT '') RETURNS jsonb LANGUAGE plpgsql AS $$
DECLARE item record; m "MemoryPracticeRecord"; r "ReviewReward"; best record; expired record; outcome text; err text; n integer:=0; earned integer:=0; excluded integer:=0; amount numeric:=0; pending integer; errors jsonb:='[]'; cursor_id text:=after_task; more boolean; saved_n integer; saved_earned integer; saved_excluded integer; saved_amount numeric;
BEGIN
 PERFORM xs_review_manager(co,actor);
 IF batch_limit NOT BETWEEN 1 AND 500 THEN RAISE EXCEPTION '无效复习核定批量'; END IF;
 LOCK TABLE "MemoryPracticeRecord","ReviewReward" IN SHARE ROW EXCLUSIVE MODE;
 PERFORM pg_advisory_xact_lock(hashtext('xs-payroll:'||co));
 FOR item IN SELECT s.* FROM "LegacyReviewSyncState" s WHERE s.company=co AND s."pendingIncome" AND s."taskId">after_task ORDER BY s."taskId" LIMIT batch_limit FOR UPDATE LOOP
  cursor_id:=item."taskId";saved_n:=n;saved_earned:=earned;saved_excluded:=excluded;saved_amount:=amount;
  BEGIN
   SELECT * INTO m FROM "MemoryPracticeRecord" WHERE company=co AND "objectId"=item."taskId" FOR UPDATE;
   SELECT * INTO r FROM "ReviewReward" WHERE company=co AND "taskId"=item."taskId" FOR UPDATE;
   IF m."objectId" IS NULL OR r."objectId" IS NULL OR item.fingerprint<>md5(to_jsonb(m)::text) OR COALESCE(m."assignmentVersion",0)<>item."assignmentVersion" THEN RAISE EXCEPTION '来源同步后任务已改变'; END IF;
   IF EXISTS(SELECT 1 FROM "ReviewRewardIncome" WHERE company=co AND "objectId"=r."objectId" AND status IN ('earned','ineligible')) THEN
    UPDATE "LegacyReviewSyncState" SET "pendingIncome"=false,"lastError"='',"updatedAt"=now() WHERE company=co AND "taskId"=item."taskId";n:=n+1;CONTINUE;
   END IF;
   IF r."incomeMonth"<'2026-09' OR r."completedAt">now() OR r."completionRole" IS DISTINCT FROM 'historical_unknown' OR NULLIF(r."completedBy",'') IS NOT NULL THEN RAISE EXCEPTION '不符合已确认的历史核定范围'; END IF;
   IF EXISTS(SELECT 1 FROM "_User" u WHERE u.company=co AND u."legacyUserId"=r."coachId" AND concat_ws('',u."realName",u.nickname,u.username) LIKE '%自习%') THEN RAISE EXCEPTION '自习账号需单独核对，不能自动计薪'; END IF;
   -- An overdue normal copy must not disqualify an on-time duplicate copy merely
   -- because its raw receipt is historical while the valid copy is exception.
   IF xiaoshu_review_eligibility(r."plannedDate",r."completedAt")='eligible' THEN
    FOR expired IN SELECT v."objectId",v."taskId" FROM "ReviewRewardIncome" v
     JOIN "ReviewReward" raw ON raw."objectId"=v."objectId" AND raw.company=v.company
     JOIN "MemoryPracticeRecord" t ON t.company=v.company AND t."objectId"=v."taskId"
     WHERE v.company=co AND t.yhid=m.yhid AND t.xxjlid=m.xxjlid AND left(t.kywrq::text,10)=left(m.kywrq::text,10)
     AND v.status IN ('historical','exception') AND xiaoshu_review_eligibility(v."plannedDate",v."completedAt")='overdue'
     AND raw."completionRole"='historical_unknown' AND NULLIF(raw."completedBy",'') IS NULL AND v."completedAt"<=now()
     AND t."sourceKey" LIKE 'legacy:model:60:general:%'
    LOOP
     PERFORM xs_review_resolve(co,actor,jsonb_build_array(jsonb_build_object('rewardId',expired."objectId",'reasonCode','overdue')),'旧来源重复复习核对：先排除已核实逾期副本，保留计划当天完成副本的计薪资格。',true);
     UPDATE "LegacyReviewSyncState" SET "pendingIncome"=false,"lastError"='',"updatedAt"=now() WHERE company=co AND "taskId"=expired."taskId";
     excluded:=excluded+1;n:=n+1;
    END LOOP;
   END IF;
   SELECT v.* INTO best FROM "ReviewRewardIncome" v JOIN "MemoryPracticeRecord" t ON t.company=v.company AND t."objectId"=v."taskId"
    WHERE v.company=co AND t.yhid=m.yhid AND t.xxjlid=m.xxjlid AND left(t.kywrq::text,10)=left(m.kywrq::text,10) AND v.status IN ('earned','historical','exception')
    ORDER BY (v.status='earned') DESC,(v.status='historical') DESC,v."completedAt",v."objectId" LIMIT 1;
   IF xiaoshu_review_eligibility(r."plannedDate",r."completedAt")='eligible' AND best.status<>'earned' AND best."objectId"<>r."objectId" AND (
    best."completedAt" IS NULL OR best."completedAt">now() OR xiaoshu_review_eligibility(best."plannedDate",best."completedAt")<>'eligible'
    OR xs_review_duration(co,best."taskId") IS NULL
    OR NOT EXISTS(SELECT 1 FROM "ReviewReward" raw WHERE raw.company=co AND raw."objectId"=best."objectId" AND raw."completionRole"='historical_unknown' AND NULLIF(raw."completedBy",'') IS NULL AND raw.exception IN ('','来源复习任务重复'))
    OR EXISTS(SELECT 1 FROM "PayrollLine" l WHERE l.company=co AND l."reviewRewardId"=best."objectId" AND NOT COALESCE(l.excluded,false)))
   THEN RAISE EXCEPTION '重复候选的完成日期或计薪依据需核对，不能剔除有效当天完成'; END IF;
   IF xiaoshu_review_eligibility(r."plannedDate",r."completedAt")='overdue' THEN outcome:='overdue';
   ELSIF best."objectId"<>r."objectId" THEN outcome:='duplicate_review';
   ELSE outcome:='legacy_assignment_policy'; END IF;
   PERFORM xs_review_resolve(co,actor,jsonb_build_array(jsonb_build_object('rewardId',r."objectId",'reasonCode',outcome,'canonicalRewardId',best."objectId")),
    '旧系统来源自动核对：按已确认历史绑定老师、计划当天完成和来源课时时长口径；保留实际完成身份未知，排除逾期和重复。',true);
   SELECT * INTO best FROM "ReviewRewardIncome" WHERE company=co AND "objectId"=r."objectId";
   IF best.status='earned' THEN earned:=earned+1;amount:=amount+best.amount;ELSE excluded:=excluded+1;END IF;
   UPDATE "LegacyReviewSyncState" SET "pendingIncome"=false,"lastError"='',"updatedAt"=now() WHERE company=co AND "taskId"=item."taskId";
   n:=n+1;
  EXCEPTION WHEN OTHERS THEN
   GET STACKED DIAGNOSTICS err=MESSAGE_TEXT;
   n:=saved_n;earned:=saved_earned;excluded:=saved_excluded;amount:=saved_amount;
   UPDATE "LegacyReviewSyncState" SET "lastError"=err,"updatedAt"=clock_timestamp() WHERE company=co AND "taskId"=item."taskId";
   IF jsonb_array_length(errors)<25 THEN errors:=errors||jsonb_build_array(jsonb_build_object('taskId',item."taskId",'reason',err));END IF;
  END;
 END LOOP;
 SELECT count(*)::int INTO pending FROM "LegacyReviewSyncState" WHERE company=co AND "pendingIncome";
 SELECT EXISTS(SELECT 1 FROM "LegacyReviewSyncState" WHERE company=co AND "pendingIncome" AND "taskId">cursor_id) INTO more;
 RETURN jsonb_build_object('processed',n,'earned',earned,'excluded',excluded,'amount',amount,'pending',pending,'errors',errors,'afterTaskId',cursor_id,'hasMore',more);
END $$;
REVOKE ALL ON FUNCTION xs_legacy_review_owned(text,text),xs_legacy_review_track(text,text,numeric),xs_legacy_review_patch(text,text,jsonb,jsonb),xs_legacy_review_income(text,text,integer,text) FROM PUBLIC;
CREATE OR REPLACE FUNCTION xs_legacy_review_common_insert() RETURNS trigger LANGUAGE plpgsql AS $$
DECLARE task_id text;
BEGIN
 SELECT "objectId" INTO task_id FROM "MemoryPracticeRecord" WHERE company=NEW.company AND id=NEW."itemId" AND "sourceKey" LIKE 'legacy:model:60:general:%' AND COALESCE("assignmentVersion",0)=0;
 IF task_id IS NOT NULL THEN PERFORM xs_legacy_review_track(NEW.company,task_id); END IF;
 RETURN NEW;
END $$;
DROP TRIGGER IF EXISTS legacy_review_source_insert ON "CommonModel";
CREATE TRIGGER legacy_review_source_insert AFTER INSERT ON "CommonModel" FOR EACH ROW WHEN (NEW."modelId"=60) EXECUTE FUNCTION xs_legacy_review_common_insert();
REVOKE ALL ON FUNCTION xs_legacy_review_common_insert() FROM PUBLIC;
-- The two audited historical repairs came from the old source, not a human transfer.
-- Preserve their version numbers and audit rows, and only adopt an unchanged repair version.
DO $$ DECLARE a record;
BEGIN
 IF to_regclass('"ReviewTransferCommand"') IS NOT NULL AND to_regclass('"ReviewTransferAudit"') IS NOT NULL THEN
  FOR a IN SELECT m.company,m."objectId",m."assignmentVersion" FROM "MemoryPracticeRecord" m
   JOIN "ReviewTransferAudit" t ON t.company=m.company AND t."taskId"=m."objectId" AND t."newVersion"=m."assignmentVersion" AND t."newCoachId"=m.plid::text
   JOIN "ReviewTransferCommand" c ON c.company=t.company AND c."objectId"=t."commandId"
   WHERE c.key IN ('repair-zeng-1761-review-2026-11-02','repair-legacy-review-transfers-2026-11-02')
   AND m."sourceKey" LIKE 'legacy:model:60:general:%'
   AND NOT EXISTS(SELECT 1 FROM "LegacyReviewSyncState" s WHERE s.company=m.company AND s."taskId"=m."objectId")
  LOOP PERFORM xs_legacy_review_track(a.company,a."objectId",a."assignmentVersion"::numeric);END LOOP;
 END IF;
END $$;
COMMIT;