xiaoshu-app/scripts/merge-reading-users.mjs

#!/usr/bin/env node

import { randomBytes } from 'node:crypto';
import { buildPlan, isReadingImport, isCanonical, text } from './lib/reading-account-policy.mjs';

const APP_ID = process.env.XIAOSHU_PARSE_APP_ID || '7pIbDBJmKx_main';
const MASTER_KEY = process.env.XIAOSHU_MASTER_KEY || '';
const PARSE_URL = (process.env.XIAOSHU_PARSE_URL || 'https://server.xiaoshu.pro/parse').replace(/\/$/, '');
const FUNCTION_URL = PARSE_URL.replace(/\/parse$/, '/api/functions');
const COMPANY_ID = process.env.XIAOSHU_COMPANY_ID || '7pIbDBJmKx';
const commit = process.argv.includes('--commit');

if (!MASTER_KEY) throw new Error('缺少 XIAOSHU_MASTER_KEY');

const headers = {
  'X-Parse-Application-Id': APP_ID,
  'X-Parse-Master-Key': MASTER_KEY,
  'Content-Type': 'application/json',
};
const company = { __type: 'Pointer', className: 'Company', objectId: COMPANY_ID };
const sleep = (milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds));

async function parse(path, init = {}, retries = 4) {
  try {
    const response = await fetch(`${PARSE_URL}${path}`, {
      ...init,
      headers: { ...headers, ...(init.headers || {}) },
    });
    const payload = await response.json().catch(() => ({}));
    if (!response.ok || payload.error) {
      const error = new Error(typeof payload.error === 'string' ? payload.error : JSON.stringify(payload.error || { status: response.status }));
      error.status = response.status;
      throw error;
    }
    return payload;
  } catch (error) {
    if (!retries || (error.status && error.status < 500 && error.status !== 429)) throw error;
    await sleep((5 - retries) * 800);
    return parse(path, init, retries - 1);
  }
}

async function allRows(path, where, keys) {
  const rows = [];
  for (let skip = 0; ; skip += 1000) {
    const query = new URLSearchParams({
      where: JSON.stringify(where),
      limit: '1000',
      skip: String(skip),
      order: 'objectId',
      keys,
    });
    const page = (await parse(`${path}?${query}`)).results || [];
    rows.push(...page);
    if (page.length < 1000) break;
  }
  return rows;
}

async function ownerStats() {
  const result = {};
  for (const className of ['SurveyItem', 'SurveyLog']) {
    const rows = await allRows(`/classes/${className}`, { user: { $exists: true } }, 'objectId,user');
    const counts = new Map();
    for (const row of rows) {
      const id = text(row.user?.objectId || row.user).replace(/^_User\$/, '');
      if (id) counts.set(id, (counts.get(id) || 0) + 1);
    }
    result[className] = { records: rows.length, owners: counts.size, counts };
  }
  return result;
}

let temporaryUserId = '';
let temporaryFunctionId = '';

async function executePlan(plan) {
  const suffix = `${Date.now()}_${randomBytes(4).toString('hex')}`;
  const username = `reading_merge_${suffix}`;
  const password = `${randomBytes(24).toString('base64url')}Aa9!`;
  const path = `xiaoshu/system/merge-reading-users-${suffix}`;
  const created = await parse('/users', {
    method: 'POST',
    body: JSON.stringify({
      username, password, company, isAdmin: true, role: 'admin',
      roles: ['admin', 'super-admin'], adminRoleKey: 'super-admin',
      realName: '阅读账号合并临时管理员', testCreatedBy: 'merge-reading-users',
    }),
  });
  temporaryUserId = created.objectId;
  const token = created.sessionToken || (await parse('/login', {
    method: 'POST', body: JSON.stringify({ username, password }),
  })).sessionToken;

  const embeddedMerge = JSON.stringify(plan.merge).replace(/</g, '\\u003c');
  const code = String.raw`
async function handler(request,response){
  try{
    const current=request.user||(typeof user!=='undefined'?user:null);if(!current)return response.status(401).json({success:false,message:'需要超级管理员会话'});await current.fetch({useMasterKey:true});
    const roles=Array.isArray(current.get('roles'))?current.get('roles').map(String):[];if(current.get('adminRoleKey')!=='super-admin'&&!roles.includes('super-admin'))return response.status(403).json({success:false,message:'仅超级管理员可执行'});
    const companyId=${JSON.stringify(COMPANY_ID)},merge=${embeddedMerge};
    await Psql.none('ALTER TABLE "SurveyLog" ALTER COLUMN "user" TYPE text USING CAST("user" AS text)');
    const merged=merge.length?await Psql.one('WITH input AS MATERIALIZED (SELECT x.source,x.target FROM jsonb_to_recordset($2::jsonb) AS x(source text,target text)),valid AS MATERIALIZED (SELECT i.source,i.target FROM input i JOIN "_User" s ON s."objectId"=i.source AND s."company"=$1 JOIN "_User" t ON t."objectId"=i.target AND t."company"=$1 WHERE COALESCE(s."isDeleted",FALSE)=FALSE AND COALESCE(t."isDeleted",FALSE)=FALSE),items AS (UPDATE "SurveyItem" r SET "user"=v.target,"updatedAt"=NOW() FROM valid v WHERE CAST(r."user" AS text) IN (v.source,\'_User$\'||v.source) RETURNING r."objectId"),logs AS (UPDATE "SurveyLog" r SET "user"=v.target,"updatedAt"=NOW() FROM valid v WHERE CAST(r."user" AS text) IN (v.source,\'_User$\'||v.source) RETURNING r."objectId"),profiles AS (UPDATE "_User" t SET "level"=COALESCE(NULLIF(t."level",\'\'),NULLIF(s."level",\'\')),"updatedAt"=NOW() FROM valid v JOIN "_User" s ON s."objectId"=v.source WHERE t."objectId"=v.target RETURNING t."objectId"),sessions AS (DELETE FROM "_Session" x USING valid v WHERE CAST(x."user" AS text) IN (v.source,\'_User$\'||v.source) RETURNING x."objectId"),sources AS (UPDATE "_User" s SET "isDisabled"=TRUE,"isDeleted"=TRUE,"identityType"=\'merged\',"sourceKey"=\'reading-merged:user:\'||s."objectId","legacyUserData"=COALESCE(s."legacyUserData",\'{}\'::jsonb)||jsonb_build_object(\'ReadingMergedIntoObjectId\',v.target,\'ReadingMergedAt\',NOW()),"updatedAt"=NOW() FROM valid v WHERE s."objectId"=v.source RETURNING s."objectId") SELECT (SELECT COUNT(*)::int FROM sources) users,(SELECT COUNT(*)::int FROM items) items,(SELECT COUNT(*)::int FROM logs) logs,(SELECT COUNT(*)::int FROM sessions) sessions',[companyId,JSON.stringify(merge)]):{users:0,items:0,logs:0,sessions:0};
    const promoted={users:0};
    response.json({success:true,data:{merged,promoted}});
  }catch(error){response.status(Number(error.status)||500).json({success:false,message:String(error.message||error)});}
}`;
  const fn = await parse('/classes/Function', {
    method: 'POST',
    body: JSON.stringify({
      name: path, desc: '一次性合并阅读账号与原会员', type: 'standalone', path, code,
      params: [], paramList: [], respType: 'json', respJson: { success: true }, enabled: true,
    }),
  });
  temporaryFunctionId = fn.objectId;
  const response = await fetch(`${FUNCTION_URL}/${path}`, {
    method: 'POST',
    headers: { 'X-Parse-Application-Id': APP_ID, 'Content-Type': 'application/json' },
    body: JSON.stringify({ token, params: {} }),
  });
  const payload = await response.json().catch(() => ({}));
  if (!response.ok || payload.success !== true) throw new Error(payload.message || payload.error || `账号合并失败：${response.status}`);
  return payload.data;
}

async function cleanup() {
  if (temporaryFunctionId) await parse(`/classes/Function/${temporaryFunctionId}`, { method: 'DELETE' }).catch(() => undefined);
  if (temporaryUserId) {
    const where = encodeURIComponent(JSON.stringify({ user: { __type: 'Pointer', className: '_User', objectId: temporaryUserId } }));
    const sessions = await parse(`/classes/_Session?where=${where}&limit=1000&keys=objectId`).catch(() => ({ results: [] }));
    for (const session of sessions.results || []) await parse(`/classes/_Session/${session.objectId}`, { method: 'DELETE' }).catch(() => undefined);
    await parse(`/users/${temporaryUserId}`, { method: 'DELETE' }).catch(() => undefined);
  }
}

const userKeys = 'objectId,username,nickname,nickName,realName,name,mobile,phone,mobilePhoneNumber,level,company,sourceKey,legacyUserId,legacyGroupId,legacyUserData,identityType,isAdmin,isDisabled,isDeleted,createdAt,registeredAt';
const usersBefore = await allRows('/users', { company }, userKeys);
const ownersBefore = await ownerStats();
const plan = buildPlan(usersBefore);
const sourceIds = new Set(plan.sources.map((row) => row.objectId));
const readingSources = new Set();
for (const value of Object.values(ownersBefore)) {
  for (const owner of value.counts.keys()) if (sourceIds.has(owner)) readingSources.add(owner);
}
const summary = {
  mode: commit ? 'commit' : 'dry-run',
  usersBefore: usersBefore.length,
  canonicalUsersBefore: plan.targets.length,
  readingImports: plan.sources.length,
  readingImportsWithRecords: readingSources.size,
  mergeUsers: plan.merge.length,
  promoteUsers: plan.promote.length,
  maxMergeSourceIdLength: Math.max(0, ...plan.merge.map((item) => item.source.length)),
  maxMergeTargetIdLength: Math.max(0, ...plan.merge.map((item) => item.target.length)),
  nameResolved: plan.nameResolved,
  heldForVerification: plan.held,
  ambiguousHeld: plan.ambiguous,
  unmatchedHeld: plan.unmatched,
  surveyItemsToMove: plan.merge.reduce((total, item) => total + (ownersBefore.SurveyItem.counts.get(item.source) || 0), 0),
  surveyLogsToMove: plan.merge.reduce((total, item) => total + (ownersBefore.SurveyLog.counts.get(item.source) || 0), 0),
};

try {
  if (commit && (plan.merge.length || plan.promote.length)) summary.executed = await executePlan(plan);
  if (commit) {
    const usersAfter = await allRows('/users', { company }, userKeys);
    const ownersAfter = await ownerStats();
    const remaining = usersAfter.filter(isReadingImport);
    const activeById = new Map(usersAfter.filter((row) => !row.isDeleted).map((row) => [row.objectId, row]));
    const invalidOwners = new Set();
    for (const value of Object.values(ownersAfter)) {
      for (const owner of value.counts.keys()) {
        const row = activeById.get(owner);
        if (!row || (!isCanonical(row) && !text(row.sourceKey).startsWith('reading-unified:user:'))) invalidOwners.add(owner);
      }
    }
    summary.verification = {
      usersAfter: usersAfter.length,
      remainingUnnormalizedUsers: remaining.length,
      invalidReadingOwners: invalidOwners.size,
      surveyItems: ownersAfter.SurveyItem.records,
      surveyLogs: ownersAfter.SurveyLog.records,
    };
    // Unverified reading owners remain untouched, instead of receiving invented legacy IDs.
    const remainingPlan = buildPlan(usersAfter);
    if (remainingPlan.merge.length) throw new Error('合并后仍有已核实待合并账号');
    if (ownersAfter.SurveyItem.records !== ownersBefore.SurveyItem.records || ownersAfter.SurveyLog.records !== ownersBefore.SurveyLog.records) throw new Error('阅读记录数量变化');
  }
  process.stdout.write(`${JSON.stringify(summary, null, 2)}\n`);
} finally {
  await cleanup();
}