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; }, {}), });