Initial import of verae-middleware from zapier monorepo

src/workers/inProcessJobPoller.js

@@ -0,0 +1,108 @@
+/**
+ * @fileoverview In-process job poller when NATS_ENABLED=false.
+ * @module workers/inProcessJobPoller
+ */
+
+import { config } from '../config.js';
+import { veraeClient } from '../clients/veraeClient.js';
+import {
+  listPendingJobs,
+  updateJobWatcher,
+  removeJobWatcher,
+} from '../store/jobWatchers.js';
+import { getActiveWebhooks } from '../store/webhooks.js';
+import { deliverWebhook } from '../services/webhookService.js';
+import { createDebugger } from '../debug/logger.js';
+
+const log = createDebugger('jobs');
+
+let timer = null;
+let running = false;
+
+/**
+ * @param {import('../store/jobWatchers.js').JobWatcher} job
+ */
+async function processJob(job) {
+  const attempts = job.attempts + 1;
+  updateJobWatcher(job.id, { attempts });
+
+  if (attempts > config.jobPollMaxAttempts) {
+    updateJobWatcher(job.id, { status: 'timeout' });
+    removeJobWatcher(job.id);
+    log.warn('job timeout', { jobId: job.jobId });
+    return;
+  }
+
+  let status;
+  try {
+    status = await veraeClient.getStatus(job.veraeToken, job.jobId);
+  } catch (err) {
+    log.debug('poll error', { jobId: job.jobId, error: err.message });
+    return;
+  }
+
+  if (status.status === 'pending') {
+    return;
+  }
+
+  const event = status.status === 'completed' ? 'timestamp.completed' : 'timestamp.failed';
+  const hooks = getActiveWebhooks(job.tenantId, event);
+
+  for (const hook of hooks) {
+    try {
+      await deliverWebhook(hook.targetUrl, {
+        event,
+        jobId: job.jobId,
+        tenantId: job.tenantId,
+        status,
+      });
+    } catch (err) {
+      log.error('webhook deliver failed', { hookId: hook.id, error: err.message });
+    }
+  }
+
+  updateJobWatcher(job.id, { status: status.status });
+  removeJobWatcher(job.id);
+  log.debug('job terminal', { jobId: job.jobId, status: status.status, hooks: hooks.length });
+}
+
+async function tick() {
+  if (running) return;
+  running = true;
+  try {
+    const jobs = listPendingJobs();
+    await Promise.all(jobs.map((job) => processJob(job)));
+  } finally {
+    running = false;
+  }
+}
+
+/**
+ * Start interval poller (no-op if already started or NATS enabled).
+ * @returns {void}
+ */
+export function startInProcessJobPoller() {
+  if (config.natsEnabled) {
+    log.info('in-process poller skipped (NATS_ENABLED=true)');
+    return;
+  }
+  if (timer) return;
+
+  const interval = config.jobPollIntervalMs;
+  timer = setInterval(() => {
+    tick().catch((err) => log.error('poller tick failed', { error: err.message }));
+  }, interval);
+
+  log.info('in-process job poller started', { intervalMs: interval });
+}
+
+/**
+ * Stop interval poller.
+ * @returns {void}
+ */
+export function stopInProcessJobPoller() {
+  if (!timer) return;
+  clearInterval(timer);
+  timer = null;
+  log.info('in-process job poller stopped');
+}