memind/scripts/sync-plaza-to-prod.mjs

#!/usr/bin/env node
/**
 * Sync local plaza posts (+ publications/pages/assets) to production.
 *
 * Usage:
 *   node scripts/sync-plaza-to-prod.mjs
 *   node scripts/sync-plaza-to-prod.mjs --dry-run
 *   node scripts/sync-plaza-to-prod.mjs --include-offline   # re-online offline publications
 *
 * Env:
 *   DATABASE_URL              local source (default from .env)
 *   PLAZA_PROD_DATABASE_URL   production MySQL (required)
 *   PLAZA_PROD_SSH            e.g. root@ssh105
 *   MINDSPACE_STORAGE_ROOT    local storage (default data/mindspace)
 *   PLAZA_PROD_STORAGE_ROOT   remote storage (default .../h5/data/mindspace)
 */
import crypto from 'node:crypto';
import fs from 'node:fs';
import path from 'node:path';
import { execSync } from 'node:child_process';
import { fileURLToPath } from 'node:url';
import mysql from 'mysql2/promise';

const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..');
const dryRun = process.argv.includes('--dry-run');
const includeOffline = process.argv.includes('--include-offline');

function loadEnvFile(filePath) {
  if (!fs.existsSync(filePath)) return;
  for (const line of fs.readFileSync(filePath, 'utf8').split('\n')) {
    const trimmed = line.trim();
    if (!trimmed || trimmed.startsWith('#')) continue;
    const eq = trimmed.indexOf('=');
    if (eq < 0) continue;
    const key = trimmed.slice(0, eq).trim();
    const value = trimmed.slice(eq + 1).trim();
    if (!process.env[key]) process.env[key] = value;
  }
}

loadEnvFile(path.join(root, '../../.env.local'));
loadEnvFile(path.join(root, '.env'));

const sourceUrl = process.env.DATABASE_URL;
const targetUrl = process.env.PLAZA_PROD_DATABASE_URL;
const prodSsh = process.env.PLAZA_PROD_SSH ?? 'root@ssh105';
const localStorageRoot = path.resolve(
  process.env.MINDSPACE_STORAGE_ROOT ?? path.join(root, 'data', 'mindspace'),
);
const remoteStorageRoot =
  process.env.PLAZA_PROD_STORAGE_ROOT ?? '/root/tkmind_go/ui/h5/data/mindspace';

if (!sourceUrl) {
  console.error('缺少 DATABASE_URL');
  process.exit(1);
}
if (!targetUrl) {
  console.error('缺少 PLAZA_PROD_DATABASE_URL');
  process.exit(1);
}

function quoteId(value) {
  return mysql.escape(value);
}

async function resolveProdSpaceId(target, userId) {
  const [rows] = await target.query(`SELECT id FROM h5_user_spaces WHERE user_id = ? LIMIT 1`, [userId]);
  if (rows[0]?.id) return rows[0].id;
  throw new Error(`生产环境缺少用户 ${userId} 的 MindSpace，请先在 prod 初始化空间`);
}

async function buildProdCategoryMap(source, target, localCategoryIds) {
  if (localCategoryIds.length === 0) return new Map();
  const [localCats] = await source.query(`SELECT * FROM h5_space_categories WHERE id IN (?)`, [
    localCategoryIds,
  ]);
  const map = new Map();
  for (const cat of localCats) {
    const prodSpaceId = await resolveProdSpaceId(target, cat.user_id);
    const [prodCats] = await target.query(
      `SELECT id FROM h5_space_categories WHERE space_id = ? AND category_code = ? LIMIT 1`,
      [prodSpaceId, cat.category_code],
    );
    let prodCategoryId = prodCats[0]?.id;
    if (!prodCategoryId) {
      prodCategoryId = cat.id;
      await upsertRows(target, 'h5_space_categories', [{ ...cat, space_id: prodSpaceId }]);
    }
    map.set(cat.id, { spaceId: prodSpaceId, categoryId: prodCategoryId });
  }
  return map;
}

function remapSpaceFields(row, categoryMap) {
  const mapped = categoryMap.get(row.category_id);
  if (!mapped) return row;
  return { ...row, space_id: mapped.spaceId, category_id: mapped.categoryId };
}

async function fetchRows(conn, table, ids) {
  if (ids.length === 0) return [];
  const [rows] = await conn.query(`SELECT * FROM ${table} WHERE id IN (?)`, [ids]);
  return rows;
}

async function buildPlazaCategoryMap(source, target, plazaCategoryIds) {
  if (plazaCategoryIds.length === 0) return new Map();
  const [localCats] = await source.query(`SELECT id, slug FROM plaza_categories WHERE id IN (?)`, [
    plazaCategoryIds,
  ]);
  const map = new Map();
  for (const cat of localCats) {
    const [prodCats] = await target.query(`SELECT id FROM plaza_categories WHERE slug = ? LIMIT 1`, [
      cat.slug,
    ]);
    if (!prodCats[0]) throw new Error(`生产环境缺少 plaza 分类：${cat.slug}`);
    map.set(cat.id, prodCats[0].id);
  }
  return map;
}

async function upsertRows(conn, table, rows) {
  if (rows.length === 0) return 0;
  let count = 0;
  for (const row of rows) {
    const cols = Object.keys(row);
    const values = cols.map((col) => {
      const value = row[col];
      if (value instanceof Date) return value;
      if (value !== null && typeof value === 'object') return JSON.stringify(value);
      return value;
    });
    const placeholders = cols.map(() => '?').join(', ');
    const updates = cols.filter((c) => c !== 'id').map((c) => `${c}=VALUES(${c})`).join(', ');
    await conn.query(
      `INSERT INTO ${table} (${cols.join(', ')}) VALUES (${placeholders})
       ON DUPLICATE KEY UPDATE ${updates}`,
      values,
    );
    count += 1;
  }
  return count;
}

function storagePaths(keys) {
  const relPaths = new Set();
  for (const key of keys) {
    const normalized = String(key ?? '').replace(/^\/+/, '');
    if (!normalized.startsWith('users/')) continue;
    relPaths.add(normalized);
    relPaths.add(path.dirname(normalized));
  }
  return [...relPaths];
}

async function main() {
  const source = await mysql.createConnection(sourceUrl);
  const target = await mysql.createConnection(targetUrl);

  const [posts] = await source.query(
    `SELECT pp.*, pr.status AS publication_status
     FROM plaza_posts pp
     JOIN h5_publish_records pr ON pr.id = pp.publication_id
     WHERE pp.status = 'published'
     ORDER BY pp.published_at DESC`,
  );

  if (posts.length === 0) {
    console.log('本地没有已发布的广场帖子。');
    await source.end();
    await target.end();
    return;
  }

  const publicationIds = posts.map((p) => p.publication_id);
  const [publications] = await source.query(`SELECT * FROM h5_publish_records WHERE id IN (?)`, [
    publicationIds,
  ]);
  const pageIds = [...new Set(publications.map((p) => p.page_id))];
  const pageVersionIds = [...new Set(publications.map((p) => p.page_version_id))];
  const scanIds = [...new Set(publications.map((p) => p.security_scan_id))];

  const [pageVersions] = await source.query(`SELECT * FROM h5_page_versions WHERE id IN (?)`, [
    pageVersionIds,
  ]);
  const [pages] = await source.query(`SELECT * FROM h5_page_records WHERE id IN (?)`, [pageIds]);

  const assetIds = new Set();
  for (const version of pageVersions) {
    if (version.content_asset_id) assetIds.add(version.content_asset_id);
    if (version.bundle_asset_id) assetIds.add(version.bundle_asset_id);
  }
  const [assets] = await source.query(`SELECT * FROM h5_assets WHERE id IN (?)`, [[...assetIds]]);
  const [assetVersions] = await source.query(
    `SELECT * FROM h5_asset_versions WHERE asset_id IN (?)`,
    [[...assetIds]],
  );
  const [scans] = await source.query(`SELECT * FROM h5_security_scans WHERE id IN (?)`, [scanIds]);

  const spaceIds = [...new Set(pages.map((p) => p.space_id))];
  const categoryIds = [
    ...new Set([
      ...pages.map((p) => p.category_id),
      ...assets.map((a) => a.category_id),
    ]),
  ];
  const categoryMap = dryRun ? new Map() : await buildProdCategoryMap(source, target, categoryIds);
  const plazaCategoryMap = dryRun
    ? new Map()
    : await buildPlazaCategoryMap(
        source,
        target,
        [...new Set(posts.map((p) => p.category_id))],
      );

  const pagesToSync = pages.map((row) => {
    const next = remapSpaceFields(row, categoryMap);
    const assetIdSet = new Set(assets.map((a) => a.id));
    if (next.source_asset_id && !assetIdSet.has(next.source_asset_id)) {
      next.source_asset_id = null;
    }
    if (next.cover_image_asset_id && !assetIdSet.has(next.cover_image_asset_id)) {
      next.cover_image_asset_id = null;
    }
    return next;
  });
  const assetsToSync = assets.map((row) => remapSpaceFields(row, categoryMap));

  const offlineCount = publications.filter((p) => p.status !== 'online').length;
  const publicationsToSync = publications.map((row) => {
    if (row.status === 'online' || !includeOffline) return row;
    return { ...row, status: 'online', offline_at: null, updated_at: Date.now() };
  });

  const storageKeys = assetVersions.map((row) => row.storage_key).filter(Boolean);
  const filePaths = storagePaths(storageKeys);

  console.log(`准备同步 ${posts.length} 条广场帖（${publications.length} 个发布，${offlineCount} 个当前 offline）`);
  console.log(`  页面 ${pages.length} · 资产 ${assets.length} · 文件路径 ${filePaths.length}`);

  if (dryRun) {
    console.log('\n[dry-run] 将同步的帖子：');
    for (const post of posts) {
      const pub = publications.find((p) => p.id === post.publication_id);
      console.log(`  - ${post.title} (${pub?.status ?? '?'})`);
    }
    await source.end();
    await target.end();
    return;
  }

  await target.beginTransaction();
  try {
    const counts = {};
    counts.scans = await upsertRows(target, 'h5_security_scans', scans);
    counts.assets = await upsertRows(target, 'h5_assets', assetsToSync);
    counts.assetVersions = await upsertRows(target, 'h5_asset_versions', assetVersions);
    counts.pages = await upsertRows(target, 'h5_page_records', pagesToSync);
    counts.pageVersions = await upsertRows(target, 'h5_page_versions', pageVersions);
    counts.publications = await upsertRows(target, 'h5_publish_records', publicationsToSync);
    counts.plazaPosts = await upsertRows(
      target,
      'plaza_posts',
      posts.map(({ publication_status, ...post }) => ({
        ...post,
        category_id: plazaCategoryMap.get(post.category_id) ?? post.category_id,
      })),
    );

    const userIds = [...new Set(posts.map((p) => p.user_id))];
    for (const userId of userIds) {
      const [countRows] = await target.query(
        `SELECT COUNT(*) AS c FROM plaza_posts WHERE user_id = ? AND status = 'published'`,
        [userId],
      );
      await target.query(`UPDATE h5_users SET plaza_post_count = ?, updated_at = ? WHERE id = ?`, [
        countRows[0].c,
        Date.now(),
        userId,
      ]);
    }

    await target.commit();
    console.log('数据库同步完成：', counts);
  } catch (error) {
    await target.rollback();
    throw error;
  }

  await source.end();
  await target.end();

  if (filePaths.length === 0) {
    console.log('无需同步用户文件。');
    return;
  }

  const includeArgs = filePaths.map((rel) => `--include=${rel}`);
  includeArgs.push('--include=*/');
  const rsyncCmd = [
    'rsync -az',
    ...includeArgs,
    '--exclude=*',
    `${localStorageRoot}/`,
    `${prodSsh}:${remoteStorageRoot}/`,
  ].join(' ');

  console.log('==> 同步发布页面文件到 105...');
  execSync(rsyncCmd, { stdio: 'inherit' });

  console.log('\n✅ 广场内容已同步到生产环境');
  console.log('   首页: https://plaza.tkmind.cn/plaza');
  if (offlineCount > 0 && !includeOffline) {
    console.log(`\n⚠  有 ${offlineCount} 条帖子关联的发布为 offline，未纳入同步。`);
    console.log('   若也要迁移，请重跑：node scripts/sync-plaza-to-prod.mjs --include-offline');
  }
}

main().catch((error) => {
  console.error(error instanceof Error ? error.message : error);
  process.exit(1);
});