cangku/backend/src/jobs/goodsFirstSeenJob.js

const cron = require('node-cron');
const jianSunDb = require('../config/jianSunDatabase');
const planningDb = require('../services/planningDatabase');
const { pickSkuCaptureTime } = require('../services/opsBIncentiveBoardService');

const JOB_CRON = String(process.env.GOODS_FIRST_SEEN_CRON || '20 5 * * *').trim();
const TIMEZONE = 'Asia/Shanghai';
const SYNC_CHUNK_SIZE = 500;
const BACKFILL_CHUNK_SIZE = 800;
const PRODUCT_SNAPSHOT_TABLE = 'op_pdd_sku_dimension_goods_daily';

let schemaEnsured = false;

async function ensureTable() {
  if (schemaEnsured) return;
  await planningDb.query(`
    CREATE TABLE IF NOT EXISTS pdd_goods_first_seen (
      goods_id VARCHAR(50) NOT NULL PRIMARY KEY,
      first_created_at DATETIME NOT NULL COMMENT '首次采集到的SKU创建时间（约等于上架时间）',
      last_capture_date DATE NOT NULL COMMENT '最近一次出现在采集批次的日期',
      is_onsale TINYINT NOT NULL DEFAULT 0,
      updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4`);
  schemaEnsured = true;
}

function epochToDatetime(value) {
  const num = Number(value);
  if (!Number.isFinite(num) || num <= 0) return null;
  const ms = num >= 1e12 ? num : num * 1000;
  const date = new Date(ms);
  return Number.isNaN(date.getTime()) ? null : date;
}

// 同步单个采集批次：新商品写入首见时间，已存在商品只刷新最近采集日/在售状态（保留最初 first_created_at）
async function syncCaptureDate(captureDate) {
  await ensureTable();
  const startedAt = Date.now();
  const rows = await jianSunDb.query(
    `SELECT LTRIM(RTRIM(TRY_CONVERT(NVARCHAR(50), s.goodsId))) AS goodsId,
            MIN(s.createdAt) AS minCreatedAt,
            MAX(TRY_CONVERT(INT, s.isOnsale)) AS onsale
     FROM dbo.pdd_Skus s WITH (NOLOCK, INDEX(pdd_Skus_createTime_index))
     WHERE s.createTime = @captureDate
       AND s.goodsId IS NOT NULL AND LTRIM(RTRIM(TRY_CONVERT(NVARCHAR(50), s.goodsId))) <> ''
     GROUP BY LTRIM(RTRIM(TRY_CONVERT(NVARCHAR(50), s.goodsId)))`,
    { captureDate },
    { requestTimeout: 300000 }
  );

  const records = [];
  rows.forEach((row) => {
    const createdAt = epochToDatetime(row.minCreatedAt);
    if (!createdAt) return;
    records.push([String(row.goodsId), createdAt, captureDate, Number(row.onsale) === 1 ? 1 : 0]);
  });

  const countBefore = await planningDb.query('SELECT COUNT(*) AS n FROM pdd_goods_first_seen');
  for (let offset = 0; offset < records.length; offset += SYNC_CHUNK_SIZE) {
    const chunk = records.slice(offset, offset + SYNC_CHUNK_SIZE);
    await planningDb.query(
      `INSERT INTO pdd_goods_first_seen (goods_id, first_created_at, last_capture_date, is_onsale)
       VALUES ?
       ON DUPLICATE KEY UPDATE
         last_capture_date = VALUES(last_capture_date),
         is_onsale = VALUES(is_onsale)`,
      [chunk]
    );
  }
  const countAfter = await planningDb.query('SELECT COUNT(*) AS n FROM pdd_goods_first_seen');

  return {
    captureDate,
    goodsInBatch: records.length,
    upserted: records.length,
    newGoods: Math.max(0, Number(countAfter[0]?.n || 0) - Number(countBefore[0]?.n || 0)),
    total: Number(countAfter[0]?.n || 0),
    elapsedMs: Date.now() - startedAt,
  };
}

// 每日增量：取最近一个完整采集批次（复用激励看板的 7 天完整性校验）
async function refreshLatestCapture() {
  const captureDate = await pickSkuCaptureTime(new Date().toISOString().slice(0, 10));
  if (!captureDate) {
    console.warn('[GoodsFirstSeen] 近7天无可用完整采集批次，跳过本次同步');
    return { skipped: true };
  }
  return syncCaptureDate(captureDate);
}

// 历史回填：对当前快照表里的全部商品，走 goodsId 索引精确点查首次创建时间（不受采集历史跨度限制）
async function backfillAllHistory() {
  await ensureTable();
  const startedAt = Date.now();
  const goodsRows = await planningDb.query(
    `SELECT DISTINCT goodsId FROM ${PRODUCT_SNAPSHOT_TABLE} WHERE statDate = (SELECT MAX(statDate) FROM ${PRODUCT_SNAPSHOT_TABLE})`
  );
  const goodsIds = [...new Set(goodsRows.map((row) => String(row.goodsId || '').trim()).filter(Boolean))];
  console.log(`[GoodsFirstSeen] 历史回填开始，快照商品 ${goodsIds.length} 个`);

  const countBefore = Number((await planningDb.query('SELECT COUNT(*) AS n FROM pdd_goods_first_seen'))[0]?.n || 0);
  let synced = 0;
  for (let offset = 0; offset < goodsIds.length; offset += BACKFILL_CHUNK_SIZE) {
    const chunk = goodsIds.slice(offset, offset + BACKFILL_CHUNK_SIZE);
    const params = {};
    const placeholders = chunk.map((goodsId, index) => {
      params[`g${index}`] = Number(goodsId);
      return `@g${index}`;
    }).join(', ');
    const rows = await jianSunDb.query(
      `SELECT s.goodsId AS goodsId,
              MIN(TRY_CONVERT(BIGINT, s.createdAt)) AS minCreatedAt,
              MAX(s.createTime) AS lastCaptureRaw,
              MAX(TRY_CONVERT(INT, s.isOnsale)) AS onsale
       FROM dbo.pdd_Skus s WITH (NOLOCK, INDEX(INDEXOF_goodsId_pdd_Skus))
       WHERE s.goodsId IN (${placeholders})
       GROUP BY s.goodsId`,
      params,
      { requestTimeout: 300000 }
    );

    const records = [];
    rows.forEach((row) => {
      const createdAt = epochToDatetime(row.minCreatedAt);
      const lastCapture = epochToDatetime(row.lastCaptureRaw);
      if (!createdAt) return;
      records.push([String(row.goodsId), createdAt, lastCapture || createdAt, Number(row.onsale) === 1 ? 1 : 0]);
    });
    for (let i = 0; i < records.length; i += SYNC_CHUNK_SIZE) {
      const chunk = records.slice(i, i + SYNC_CHUNK_SIZE);
      await planningDb.query(
        `INSERT INTO pdd_goods_first_seen (goods_id, first_created_at, last_capture_date, is_onsale)
         VALUES ?
         ON DUPLICATE KEY UPDATE
           last_capture_date = VALUES(last_capture_date),
           is_onsale = VALUES(is_onsale)`,
        [chunk]
      );
    }
    synced += records.length;
    console.log(`[GoodsFirstSeen] 回填进度 ${Math.min(offset + BACKFILL_CHUNK_SIZE, goodsIds.length)}/${goodsIds.length}，命中 ${records.length}`);
  }
  const countAfter = Number((await planningDb.query('SELECT COUNT(*) AS n FROM pdd_goods_first_seen'))[0]?.n || 0);
  const summary = { goodsInSnapshot: goodsIds.length, synced, newGoods: Math.max(0, countAfter - countBefore), total: countAfter, elapsedMs: Date.now() - startedAt };
  console.log(`[GoodsFirstSeen] 历史回填结束: ${JSON.stringify(summary)}`);
  return summary;
}

function startGoodsFirstSeenScheduler() {
  const disabled = String(process.env.DISABLE_GOODS_FIRST_SEEN_JOB || '').trim().toLowerCase() === 'true';
  if (disabled) {
    console.log('[GoodsFirstSeen] 首见时间同步任务已被禁用');
    return;
  }
  cron.schedule(JOB_CRON, () => {
    refreshLatestCapture().catch((error) => {
      console.error('[GoodsFirstSeen] 定时同步失败:', error.message);
    });
  }, { timezone: TIMEZONE });
  console.log(`[Scheduler] 商品首见时间同步任务已注册: ${JOB_CRON} (${TIMEZONE})`);
}

module.exports = { refreshLatestCapture, backfillAllHistory, syncCaptureDate, startGoodsFirstSeenScheduler };