gongxue-base/apps/worker/src/index.ts

import { closePool } from './db.js';
import { config } from './config.js';
import { processCrmBatch } from './jobs/crm.js';
import { processCommerceBatch } from './jobs/commerce.js';
import { processAssetBatch } from './jobs/assets.js';

const extraClosers = new Set<() => Promise<void>>();

function hasArg(name: string) {
  return process.argv.includes(name);
}

function argValue(name: string, fallback = '') {
  const index = process.argv.indexOf(name);
  return index >= 0 ? process.argv[index + 1] || fallback : fallback;
}

async function runOnce() {
  const job = argValue('--job', 'crm');
  if (job === 'crm') {
    const result = await processCrmBatch();
    console.log(`[worker] crm batch processed=${result.processed} sent=${result.sent} failed=${result.failed} retrying=${result.retrying} discarded=${result.discarded}`);
    return;
  }
  if (job === 'commerce') {
    const result = await processCommerceBatch();
    console.log(
      `[worker] commerce batch processed=${result.processed}`
        + ` payments=${result.payments.processed} paid=${result.payments.paid} pending=${result.payments.pending} closed=${result.payments.closed} failed=${result.payments.failed} paymentErrors=${result.payments.errors}`
        + ` refunds=${result.refunds.processed} succeeded=${result.refunds.succeeded} processing=${result.refunds.processing} refundFailed=${result.refunds.failed} refundErrors=${result.refunds.errors}`,
    );
    return;
  }
  if (job === 'provider-bills') {
    const { closePool: closeApiPool } = await import('../../api/src/core/db.js');
    const { processProviderBillBatch } = await import('./jobs/provider-bills.js');
    extraClosers.add(closeApiPool);
    const result = await processProviderBillBatch();
    console.log(
      `[worker] provider-bills batch processed=${result.processed}`
        + ` completed=${result.completed} failed=${result.failed} skipped=${result.skipped}`,
    );
    return;
  }
  if (job === 'platform-billing') {
    const { processPlatformBillingBatch } = await import('./jobs/platform-billing.js');
    const result = await processPlatformBillingBatch();
    console.log(
      `[worker] platform-billing batch processed=${result.processed}`
        + ` created=${result.created} skipped=${result.skipped} failed=${result.failed}`,
    );
    return;
  }
  if (job === 'platform-dunning') {
    const { processPlatformDunningBatch } = await import('./jobs/platform-dunning.js');
    const result = await processPlatformDunningBatch();
    console.log(
      `[worker] platform-dunning batch processed=${result.processed}`
        + ` markedOverdue=${result.markedOverdue} reminderCreated=${result.reminderCreated}`
        + ` skippedReminder=${result.skippedReminder}`,
    );
    return;
  }
  if (job === 'assets') {
    const result = await processAssetBatch();
    console.log(
      `[worker] assets batch processed=${result.processed}`
        + ` verified=${result.verified} failed=${result.failed} skipped=${result.skipped} errors=${result.errors}`,
    );
    return;
  }
  if (job === 'imports') {
    const { closeImportExecutorPool, processImportBatch } = await import('./jobs/imports.js');
    extraClosers.add(closeImportExecutorPool);
    const result = await processImportBatch();
    console.log(
      `[worker] imports batch processed=${result.processed}`
        + ` completed=${result.completed} completedWithErrors=${result.completedWithErrors}`
        + ` failed=${result.failed} retrying=${result.retrying} skipped=${result.skipped}`,
    );
    return;
  }
  if (job === 'public-banks') {
    const { closePublicBankSyncExecutorPool, processPublicBankSyncBatch } = await import('./jobs/public-banks.js');
    extraClosers.add(closePublicBankSyncExecutorPool);
    const result = await processPublicBankSyncBatch();
    console.log(
      `[worker] public-banks batch processed=${result.processed}`
        + ` synced=${result.synced} conflicts=${result.conflicts}`
        + ` failed=${result.failed} skipped=${result.skipped}`,
    );
    return;
  }
  if (job === 'exports') {
    const { closeExportExecutorPool, processExportBatch } = await import('./jobs/exports.js');
    extraClosers.add(closeExportExecutorPool);
    const result = await processExportBatch();
    console.log(
      `[worker] exports batch processed=${result.processed}`
        + ` completed=${result.completed} failed=${result.failed}`
        + ` retrying=${result.retrying} skipped=${result.skipped}`,
    );
    return;
  }
  throw new Error(`Unsupported worker job: ${job}`);
}

async function runLoop() {
  console.log('[worker] started');
  let stopped = false;
  const stop = () => {
    stopped = true;
  };
  process.once('SIGINT', stop);
  process.once('SIGTERM', stop);

  while (!stopped) {
    try {
      await runOnce();
    } catch (error) {
      console.error('[worker] job failed', error);
    }
    await new Promise(resolve => setTimeout(resolve, config.crmPollIntervalMs));
  }
}

try {
  if (hasArg('--loop')) {
    await runLoop();
  } else {
    await runOnce();
  }
} finally {
  for (const closeExtra of extraClosers) {
    await closeExtra();
  }
  await closePool();
}