From b7b89128a29b1b6495e48bf990c579edf8d63d4c Mon Sep 17 00:00:00 2001
From: George Lambert <marchon@gmail.com>
Date: Fri, 11 Sep 2026 18:38:54 -0400
Subject: [PATCH] Initial import of verae-jobs-events from zapier monorepo

---
 NATS.md             |  5 +++
 README.md           |  7 +++++
 package.json        | 14 +++++++++
 src/server.js       | 75 +++++++++++++++++++++++++++++++++++++++++++++
 src/subjects.js     |  6 ++++
 test/health.test.js | 23 ++++++++++++++
 6 files changed, 130 insertions(+)
 create mode 100644 NATS.md
 create mode 100644 README.md
 create mode 100644 package.json
 create mode 100644 src/server.js
 create mode 100644 src/subjects.js
 create mode 100644 test/health.test.js

diff --git a/NATS.md b/NATS.md
new file mode 100644
index 0000000..cd40e3f
--- /dev/null
+++ b/NATS.md
@@ -0,0 +1,5 @@
+# NATS — verae-jobs-events
+
+IN `verae.zapier.jobs.events` queue `jobs-events`.
+
+When `JOBS_EVENTS_EXCLUSIVE=1`, this process is the durable JetStream consumer `jobs-events-exclusive` on `ZAPIER_EVENTS` and republishes to `verae.internal.jobs.events`. Middleware must set the same env so it does not also consume `verae.zapier.jobs.events`.
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..936ee4d
--- /dev/null
+++ b/README.md
@@ -0,0 +1,7 @@
+# verae-jobs-events
+
+Mailbox process for `verae.zapier.jobs.events`. Today it counts events (passthrough). Webhook routing still lives in middleware until this process takes the exclusive consumer.
+
+**Forgejo:** https://git.georgelambert.org/marchon/verae-jobs-events
+
+Port `:3030`.
diff --git a/package.json b/package.json
new file mode 100644
index 0000000..89faf9a
--- /dev/null
+++ b/package.json
@@ -0,0 +1,14 @@
+{
+  "name": "verae-jobs-events",
+  "version": "0.1.0",
+  "private": true,
+  "type": "module",
+  "description": "Mailbox for verae.zapier.jobs.events (passthrough)",
+  "scripts": {
+    "start": "node src/server.js",
+    "test": "node --test test/*.test.js"
+  },
+  "dependencies": {
+    "nats": "^2.28.2"
+  }
+}
diff --git a/src/server.js b/src/server.js
new file mode 100644
index 0000000..e079086
--- /dev/null
+++ b/src/server.js
@@ -0,0 +1,75 @@
+#!/usr/bin/env node
+/** Mailbox for verae.zapier.jobs.events. Passthrough: count + optional fan-out later. */
+import http from 'node:http';
+import { SUBJECTS } from './subjects.js';
+
+const PORT = Number(process.env.PORT || process.env.FLEET_HEALTH_PORT || 3030);
+const BIND = process.env.FLEET_HEALTH_BIND || '0.0.0.0';
+let processed = 0;
+let last = null;
+
+async function startNats() {
+  const url = process.env.NATS_URL;
+  if (!url) return;
+  const { connect, StringCodec } = await import('nats');
+  const nc = await connect({ servers: url.split(','), name: 'verae-jobs-events' });
+  const sc = StringCodec();
+  const exclusive = process.env.JOBS_EVENTS_EXCLUSIVE === '1';
+  const deliver = async (m) => {
+    processed += 1;
+    try {
+      last = JSON.parse(sc.decode(m.data) || '{}');
+    } catch {
+      last = {};
+    }
+    if (exclusive) {
+      nc.publish(SUBJECTS.INTERNAL, m.data);
+    }
+    if (m.reply) m.respond(sc.encode(JSON.stringify({ ok: true, processed, exclusive })));
+    if (typeof m.ack === 'function') await m.ack();
+  };
+  if (exclusive) {
+    try {
+      const js = nc.jetstream();
+      const jsm = await nc.jetstreamManager();
+      await jsm.consumers.add('ZAPIER_EVENTS', {
+        durable_name: SUBJECTS.DURABLE,
+        ack_policy: 'explicit',
+        filter_subject: SUBJECTS.IN,
+        max_deliver: 10,
+      }).catch(() => {});
+      const consumer = await js.consumers.get('ZAPIER_EVENTS', SUBJECTS.DURABLE);
+      const messages = await consumer.consume({ max_messages: 10 });
+      for await (const m of messages) await deliver(m);
+      return;
+    } catch (err) {
+      process.stderr.write(`js exclusive fallback core sub: ${err.message}\n`);
+    }
+  }
+  for await (const m of nc.subscribe(SUBJECTS.IN, { queue: SUBJECTS.QUEUE })) {
+    await deliver(m);
+  }
+}
+
+const server = http.createServer((req, res) => {
+  const url = new URL(req.url || '/', `http://127.0.0.1:${PORT}`);
+  if (url.pathname === '/health') {
+    res.writeHead(200, { 'content-type': 'application/json' });
+    res.end(JSON.stringify({
+      ok: true,
+      role: 'verae-jobs-events',
+      processed,
+      subject: SUBJECTS.IN,
+      exclusive: process.env.JOBS_EVENTS_EXCLUSIVE === '1',
+      lastEvent: last?.event || null,
+    }));
+    return;
+  }
+  res.writeHead(404, { 'content-type': 'application/json' });
+  res.end(JSON.stringify({ error: 'not found' }));
+});
+
+server.listen(PORT, BIND, () => {
+  process.stdout.write(`verae-jobs-events http://${BIND}:${PORT}/\n`);
+});
+startNats().catch((err) => process.stderr.write(`nats optional: ${err.message}\n`));
diff --git a/src/subjects.js b/src/subjects.js
new file mode 100644
index 0000000..b320bf3
--- /dev/null
+++ b/src/subjects.js
@@ -0,0 +1,6 @@
+export const SUBJECTS = {
+  IN: 'verae.zapier.jobs.events',
+  INTERNAL: 'verae.internal.jobs.events',
+  QUEUE: 'jobs-events',
+  DURABLE: 'jobs-events-exclusive',
+};
diff --git a/test/health.test.js b/test/health.test.js
new file mode 100644
index 0000000..71dbb2f
--- /dev/null
+++ b/test/health.test.js
@@ -0,0 +1,23 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { spawn } from 'node:child_process';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const root = path.join(path.dirname(fileURLToPath(import.meta.url)), '..');
+
+test('jobs-events health', async () => {
+  const port = 18030;
+  const child = spawn(process.execPath, ['src/server.js'], {
+    cwd: root,
+    env: { ...process.env, PORT: String(port) },
+    stdio: ['ignore', 'pipe', 'pipe'],
+  });
+  await new Promise((r) => setTimeout(r, 400));
+  try {
+    const h = await (await fetch(`http://127.0.0.1:${port}/health`)).json();
+    assert.equal(h.role, 'verae-jobs-events');
+  } finally {
+    child.kill('SIGTERM');
+  }
+});