From 6933de074f2c246ad3c428c4e457b09ad7fb570d Mon Sep 17 00:00:00 2001 From: George Lambert Date: Fri, 11 Sep 2026 14:07:19 -0400 Subject: [PATCH] Initial import of verae-nats-process from zapier monorepo --- NATS.md | 13 +++++++++ README.md | 20 +++++++++++++ ROUTING.md | 9 ++++++ SUMMARY.md | 7 +++++ fleet.service.json | 14 +++++++++ package.json | 11 +++++++ src/handle.js | 20 +++++++++++++ src/subjects.js | 13 +++++++++ src/worker.js | 70 +++++++++++++++++++++++++++++++++++++++++++++ test/handle.test.js | 21 ++++++++++++++ 10 files changed, 198 insertions(+) create mode 100644 NATS.md create mode 100644 README.md create mode 100644 ROUTING.md create mode 100644 SUMMARY.md create mode 100644 fleet.service.json create mode 100644 package.json create mode 100644 src/handle.js create mode 100644 src/subjects.js create mode 100644 src/worker.js create mode 100644 test/handle.test.js diff --git a/NATS.md b/NATS.md new file mode 100644 index 0000000..ce966ae --- /dev/null +++ b/NATS.md @@ -0,0 +1,13 @@ +# NATS — verae-nats-process (template) + +Connect to the **Verae 3-node JetStream cluster** (private URLs). Zapier never connects. + +| Direction | Address | From / to | +|-----------|---------|-----------| +| IN | `verae.example.process.in` | middleware or another worker | +| OUT | `verae.example.process.out` | peers | +| OUT | `verae.example.process.reply.` | requester | + +Queue group `example-process` shares work. For fan-out (like archive query) **omit** the queue group. + +Fleet: HTTP `/health` on loopback. `keepFloor` in `verae-fleet` if this process must stay up. diff --git a/README.md b/README.md new file mode 100644 index 0000000..b9bac35 --- /dev/null +++ b/README.md @@ -0,0 +1,20 @@ +# verae-nats-process + +Template for a new **addressed messaging process** on the central Verae NATS.IO cluster. + +**Repo:** https://git.georgelambert.org/marchon/verae-nats-process +**Clone:** `ssh://git@git.georgelambert.org:2223/marchon/verae-nats-process.git` +**Overview:** https://git.georgelambert.org/marchon/overview + +```bash +npm test +npm start # /health on 127.0.0.1:13900 +``` + +1. Duplicate this repository (new Forgejo name). +2. Edit `src/subjects.js` — `verae...*`. +3. Replace `handle()` in `src/handle.js`. +4. Update `ROUTING.md`; add a fleet service with `min`/`max`. +5. If Zapier needs the result, add HTTPS on **verae-middleware** only. + +Zapier cloud must not subscribe. See [overview 06-address-routing](https://git.georgelambert.org/marchon/overview/src/branch/main/06-address-routing.md). diff --git a/ROUTING.md b/ROUTING.md new file mode 100644 index 0000000..76b85ce --- /dev/null +++ b/ROUTING.md @@ -0,0 +1,9 @@ +# Address row (copy into overview INDEX and docs-master) + +| Address | Kind | Queue group | Publisher | Subscriber | Body | +|---------|------|-------------|-----------|------------|------| +| `verae.example.process.in` | JetStream work | `example-process` | middleware or peer | this worker | `{ correlationId, payload, traceId }` | +| `verae.example.process.out` | events | — | this worker | interested peers | handle() result | +| `verae.example.process.reply.` | reply | — | this worker | original requester | handle() result | + +Rename `example` / `process` before production. Do not subscribe from Zapier. diff --git a/SUMMARY.md b/SUMMARY.md new file mode 100644 index 0000000..2c9e58d --- /dev/null +++ b/SUMMARY.md @@ -0,0 +1,7 @@ +# verae-nats-process + +**Job:** Reference worker for a **new NATS address**. Copy this repo to add search, storage, or any unplanned function. + +**Expects:** `verae.example.process.in` +**Sends:** `verae.example.process.out` and `reply.` +**Test:** `npm test` diff --git a/fleet.service.json b/fleet.service.json new file mode 100644 index 0000000..21dc446 --- /dev/null +++ b/fleet.service.json @@ -0,0 +1,14 @@ +{ + "id": "verae-nats-process", + "title": "Template addressed process", + "kind": "nats-worker", + "package": "verae-nats-process", + "role": "verae-nats-process", + "managed": true, + "health": { "type": "http", "path": "/health", "timeoutMs": 800 }, + "ports": { "healthBase": 13900 }, + "nats": { + "in": ["verae.example.process.in"], + "out": ["verae.example.process.out", "verae.example.process.reply."] + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..739a547 --- /dev/null +++ b/package.json @@ -0,0 +1,11 @@ +{ + "name": "verae-nats-process", + "version": "0.1.0", + "type": "module", + "description": "Template JetStream worker for a new verae.* address (copy this repo to expand)", + "scripts": { + "start": "node src/worker.js", + "test": "node --test test/*.test.js" + }, + "engines": { "node": ">=20" } +} diff --git a/src/handle.js b/src/handle.js new file mode 100644 index 0000000..c62d537 --- /dev/null +++ b/src/handle.js @@ -0,0 +1,20 @@ +/** + * Replace this with search, store, or any new function. + * Must be idempotent: the same correlationId may be redelivered. + * + * @param {object} msg + * @returns {Promise|object} + */ +export function handle(msg = {}) { + const correlationId = msg.correlationId || msg.traceId; + if (!correlationId) { + throw new Error('correlationId or traceId is required'); + } + return { + ok: true, + correlationId, + echo: msg.payload ?? msg, + processedAt: new Date().toISOString(), + note: 'Template handler — replace handle() in src/handle.js', + }; +} diff --git a/src/subjects.js b/src/subjects.js new file mode 100644 index 0000000..9df5b52 --- /dev/null +++ b/src/subjects.js @@ -0,0 +1,13 @@ +/** + * Rename `example.process` before production. + * Pattern: verae... + */ +export const AREA = 'example'; +export const RESOURCE = 'process'; + +export const SUBJECTS = Object.freeze({ + IN: `verae.${AREA}.${RESOURCE}.in`, + OUT: `verae.${AREA}.${RESOURCE}.out`, + QUEUE: `${AREA}-${RESOURCE}`, + reply: (correlationId) => `verae.${AREA}.${RESOURCE}.reply.${correlationId}`, +}); diff --git a/src/worker.js b/src/worker.js new file mode 100644 index 0000000..cea6c86 --- /dev/null +++ b/src/worker.js @@ -0,0 +1,70 @@ +#!/usr/bin/env node +/** + * Health HTTP + optional NATS subscribe. + * Fleet probes /health. Live NATS is off unless NATS_URL is set. + */ +import http from 'node:http'; +import { SUBJECTS } from './subjects.js'; +import { handle } from './handle.js'; + +const port = Number(process.env.FLEET_HEALTH_PORT || process.env.PORT || 13900); +const bind = process.env.FLEET_HEALTH_BIND || '127.0.0.1'; +const instance = process.env.FLEET_INSTANCE || 'verae-nats-process-0'; +let paused = false; +let processed = 0; + +function snapshot() { + return { + ok: !paused, + paused, + instance, + role: 'verae-nats-process', + subjects: { in: SUBJECTS.IN, out: SUBJECTS.OUT, queue: SUBJECTS.QUEUE }, + processed, + }; +} + +const server = http.createServer(async (req, res) => { + const url = new URL(req.url, 'http://127.0.0.1'); + const json = (code, obj) => { + res.writeHead(code, { 'content-type': 'application/json' }); + res.end(JSON.stringify(obj)); + }; + if (url.pathname === '/health' && req.method === 'GET') { + const s = snapshot(); + return json(s.ok ? 200 : 503, s); + } + if (url.pathname === '/pause' && req.method === 'POST') { + paused = true; + return json(200, snapshot()); + } + if (url.pathname === '/resume' && req.method === 'POST') { + paused = false; + return json(200, snapshot()); + } + if (url.pathname === '/message' && req.method === 'POST') { + const chunks = []; + for await (const c of req) chunks.push(c); + let body = {}; + try { + body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}'); + } catch { + body = {}; + } + if (paused) return json(503, snapshot()); + try { + const out = await handle(body); + processed += 1; + return json(200, { rttMs: 0, result: out, replyTo: SUBJECTS.reply(out.correlationId) }); + } catch (err) { + return json(400, { error: err.message }); + } + } + json(404, { error: 'not found' }); +}); + +server.listen(port, bind, () => { + process.stdout.write(`${instance} health http://${bind}:${port}/health in=${SUBJECTS.IN}\n`); +}); + +process.on('SIGTERM', () => server.close(() => process.exit(0))); diff --git a/test/handle.test.js b/test/handle.test.js new file mode 100644 index 0000000..cfecbf6 --- /dev/null +++ b/test/handle.test.js @@ -0,0 +1,21 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { handle } from '../src/handle.js'; +import { SUBJECTS } from '../src/subjects.js'; + +describe('verae-nats-process template', () => { + it('uses the verae.area.resource.action address pattern', () => { + assert.equal(SUBJECTS.IN, 'verae.example.process.in'); + assert.equal(SUBJECTS.OUT, 'verae.example.process.out'); + assert.match(SUBJECTS.reply('abc'), /^verae\.example\.process\.reply\.abc$/); + }); + + it('handle requires correlation and is idempotent-shaped', () => { + assert.throws(() => handle({}), /correlationId/); + const a = handle({ correlationId: 'c1', payload: { q: 1 } }); + const b = handle({ correlationId: 'c1', payload: { q: 1 } }); + assert.equal(a.ok, true); + assert.equal(a.correlationId, b.correlationId); + assert.equal(a.echo.q, 1); + }); +});