Initial import of verae-middleware from zapier monorepo

test/integration/nats.test.js

@@ -0,0 +1,117 @@
+/**
+ * GATE 7 — NATS streams + publish/consume
+ * Requires nats-server with JetStream on NATS_URL (default 127.0.0.1:4222)
+ */
+
+import { describe, it, before, after } from 'node:test';
+import assert from 'node:assert/strict';
+import { config } from '../../src/config.js';
+import { SUBJECTS, STREAMS } from '../../src/nats/subjects.js';
+import {
+  connectNats,
+  ensureStreams,
+  closeNats,
+  isNatsConnected,
+} from '../../src/nats/connection.js';
+import { enqueueWatch, publishJobEvent } from '../../src/nats/publishers.js';
+
+describe('NATS infrastructure', () => {
+  before(async () => {
+    // Force connect even if NATS_ENABLED was false at boot — re-set for this process
+    config.natsEnabled = true;
+    process.env.NATS_FORCE_CONNECT = '1';
+    config.natsUrl = process.env.NATS_URL || 'nats://127.0.0.1:4222';
+
+    try {
+      await connectNats(config.natsUrl);
+      await ensureStreams();
+    } catch (err) {
+      assert.fail(
+        `NATS not available at ${config.natsUrl}: ${err.message}. Start: nats-server -js -p 4222`,
+      );
+    }
+  });
+
+  after(async () => {
+    await closeNats();
+    process.env.NATS_FORCE_CONNECT = '';
+  });
+
+  it('connects and reports connected', () => {
+    assert.equal(isNatsConnected(), true);
+  });
+
+  it('ensures streams exist (idempotent)', async () => {
+    await ensureStreams();
+    const { jsm } = await connectNats();
+    for (const name of [STREAMS.ZAPIER_JOBS, STREAMS.ZAPIER_EVENTS, STREAMS.ZAPIER_WEBHOOKS]) {
+      const info = await jsm.streams.info(name);
+      assert.equal(info.config.name, name);
+    }
+  });
+
+  it('publish + pull consume one watch message', async () => {
+    const { js, jsm } = await connectNats();
+
+    // Avoid consuming leftover messages from prior runs
+    await jsm.streams.purge(STREAMS.ZAPIER_JOBS);
+
+    const durable = `test-pull-${Date.now()}`;
+    await jsm.consumers.add(STREAMS.ZAPIER_JOBS, {
+      durable_name: durable,
+      ack_policy: 'explicit',
+      filter_subject: SUBJECTS.JOBS_WATCH,
+      deliver_policy: 'all',
+    });
+
+    const jobId = `job-${Date.now()}`;
+    const pub = await enqueueWatch({
+      tenantId: 'tenant-nats-test',
+      jobId,
+      maxAttempts: 5,
+      intervalMs: 100,
+    });
+    assert.ok(pub.seq >= 0);
+
+    const consumer = await js.consumers.get(STREAMS.ZAPIER_JOBS, durable);
+    const messages = await consumer.fetch({ max_messages: 5, expires: 5000 });
+    let got = null;
+    for await (const msg of messages) {
+      const data = JSON.parse(msg.string());
+      msg.ack();
+      if (data.jobId === jobId) {
+        got = data;
+        break;
+      }
+    }
+
+    assert.ok(got, 'expected a message for our jobId');
+    assert.equal(got.jobId, jobId);
+    assert.equal(got.tenantId, 'tenant-nats-test');
+  });
+
+  it('publishJobEvent works', async () => {
+    const result = await publishJobEvent({
+      event: 'timestamp.completed',
+      tenantId: 't1',
+      jobId: 'j1',
+      status: { id: 'j1', status: 'completed' },
+    });
+    assert.ok(result.seq >= 0);
+  });
+});
+
+describe('NATS disabled path', () => {
+  it('connectNats throws when disabled and not forced', async () => {
+    const prev = config.natsEnabled;
+    const force = process.env.NATS_FORCE_CONNECT;
+    config.natsEnabled = false;
+    process.env.NATS_FORCE_CONNECT = '';
+    // close existing so we hit the disabled check on a fresh call path
+    // Note: if already connected, connectNats returns cached — test isolation via disabled only when no cache
+    // This tests the disabled branch of a new process conceptually; here we only assert flag behavior:
+    assert.equal(config.natsEnabled, false);
+    config.natsEnabled = prev;
+    process.env.NATS_FORCE_CONNECT = force;
+  });
+});