jianhen-voc/deploy/process-sp-api-reports.mjs

const parseUrl = process.env.PARSE_URL || 'http://127.0.0.1:3000/parse';
const apiUrl = process.env.API_URL || 'http://127.0.0.1:3000/api/amazon';
const appId = process.env.PARSE_APP_ID;
const masterKey = process.env.PARSE_MASTER_KEY;
const rounds = Number(process.env.REPORT_POLL_ROUNDS || 12);
const roundDelayMs = Number(process.env.REPORT_POLL_DELAY_MS || 30000);

if (!appId || !masterKey) throw new Error('Missing PARSE_APP_ID or PARSE_MASTER_KEY');

const headers = {
  'Content-Type': 'application/json',
  'X-Parse-Application-Id': appId,
  'X-Parse-Master-Key': masterKey,
};
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));

function log(event, data = {}) {
  console.log(JSON.stringify({ at: new Date().toISOString(), event, ...data }));
}

async function parseRequest(path, method = 'GET', body) {
  const response = await fetch(`${parseUrl}${path}`, {
    method,
    headers,
    body: body === undefined ? undefined : JSON.stringify(body),
  });
  const text = await response.text();
  let result = {};
  try { result = JSON.parse(text); } catch {}
  if (!response.ok || result.error) {
    throw new Error(`${method} ${path}: ${result.error?.message || result.error || text.slice(0, 300) || response.status}`);
  }
  return result;
}

async function amazonRequest(shopId, path, body) {
  for (let attempt = 1; attempt <= 5; attempt++) {
    try {
      const response = await fetch(`${apiUrl}${path}`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json', 'shop-objectid': shopId },
        body: JSON.stringify(body),
        signal: AbortSignal.timeout(180000),
      });
      const text = await response.text();
      let result = {};
      try { result = JSON.parse(text); } catch {}
      if (!response.ok || result.success === false) {
        throw new Error(result.message || result.errors?.[0]?.message || text.slice(0, 300) || `HTTP ${response.status}`);
      }
      return result.data || result;
    } catch (error) {
      if (attempt === 5) throw error;
      const delay = attempt * 15000;
      log('amazon_retry', { shopId, path, attempt, delay, error: error.message });
      await sleep(delay);
    }
  }
}

async function getReports() {
  const reports = [];
  let skip = 0;
  while (true) {
    const page = await parseRequest(`/classes/Reports?limit=1000&skip=${skip}&order=createdAt`);
    reports.push(...(page.results || []));
    const length = page.results?.length || 0;
    if (length < 1000) return reports;
    skip += length;
  }
}

async function updateReport(reportId, body) {
  return parseRequest(`/classes/Reports/${reportId}`, 'PUT', body);
}

const finalStates = new Set(['DONE', 'CANCELLED', 'FATAL']);
const processedReports = new Set();
let totalParsedRecords = 0;

for (let round = 1; round <= rounds; round++) {
  const reports = (await getReports()).filter(report => report.isParsed !== true);
  let pending = 0;
  let processedThisRound = 0;
  const statusCounts = {};

  log('report_round_started', { round, reports: reports.length });

  for (const report of reports) {
    const shopId = report.shop?.objectId;
    const amazonReportId = report.reportId;
    if (!shopId || !amazonReportId) {
      log('report_skipped', { objectId: report.objectId, reason: 'missing shop pointer or reportId' });
      continue;
    }

    let status = report.processingStatus || 'IN_QUEUE';
    let documentId = report.reportDocumentId || '';

    if (!finalStates.has(status)) {
      try {
        const result = await amazonRequest(shopId, '/forward', {
          path: `/reports/2021-06-30/reports/${amazonReportId}`,
          method: 'GET',
        });
        status = result.processingStatus || status;
        documentId = result.reportDocumentId || documentId;
        await updateReport(report.objectId, {
          processingStatus: status,
          ...(documentId ? { reportDocumentId: documentId } : {}),
          ...(result.processingStartTime ? { processingStartTime: { __type: 'Date', iso: result.processingStartTime } } : {}),
          ...(result.processingEndTime ? { processingEndTime: { __type: 'Date', iso: result.processingEndTime } } : {}),
        });
        log('report_status', { objectId: report.objectId, shopId, reportId: amazonReportId, status, hasDocument: Boolean(documentId) });
      } catch (error) {
        pending++;
        log('report_status_failed', { objectId: report.objectId, shopId, reportId: amazonReportId, error: error.message });
        continue;
      }
    }

    statusCounts[status] = (statusCounts[status] || 0) + 1;

    if (status !== 'DONE') {
      if (!finalStates.has(status)) pending++;
      continue;
    }
    if (!documentId) {
      pending++;
      log('report_waiting_document', { objectId: report.objectId, shopId, reportId: amazonReportId });
      continue;
    }
    if (processedReports.has(report.objectId)) continue;

    try {
      const result = await amazonRequest(shopId, '/processReport', {
        reportDocumentId: documentId,
        reportType: report.reportType,
      });
      const processResult = result.processResult || result.data?.processResult || {};
      if (processResult.success === false) throw new Error(processResult.message || 'report parser returned failure');
      const parsed = Number(processResult.processed || 0);
      await updateReport(report.objectId, {
        isParsed: true,
        parsedAt: { __type: 'Date', iso: new Date().toISOString() },
        parsedRecords: parsed,
      });
      processedReports.add(report.objectId);
      totalParsedRecords += parsed;
      processedThisRound++;
      log('report_parsed', { objectId: report.objectId, shopId, reportId: amazonReportId, reportType: report.reportType, parsed });
    } catch (error) {
      pending++;
      log('report_parse_failed', { objectId: report.objectId, shopId, reportId: amazonReportId, error: error.message });
    }

    await sleep(1000);
  }

  log('report_round_finished', { round, pending, processedThisRound, statusCounts, totalParsedRecords });
  if (pending === 0) break;
  if (round < rounds) await sleep(roundDelayMs);
}

const remaining = (await getReports()).filter(report => report.isParsed !== true && !['CANCELLED', 'FATAL'].includes(report.processingStatus));
log('report_processing_finished', {
  processedReports: processedReports.size,
  totalParsedRecords,
  remaining: remaining.length,
  remainingStatuses: remaining.reduce((counts, report) => {
    const status = report.processingStatus || 'UNKNOWN';
    counts[status] = (counts[status] || 0) + 1;
    return counts;
  }, {}),
});