Initial import of verae-middleware from zapier monorepo

src/nats/connection.js

@@ -0,0 +1,116 @@
+/**
+ * @fileoverview NATS + JetStream connection lifecycle.
+ * @module nats/connection
+ */
+
+import { createDebugger } from '../debug/logger.js';
+import { config } from '../config.js';
+import { SUBJECTS, STREAMS } from './subjects.js';
+
+const log = createDebugger('nats');
+
+/** @type {import('nats').NatsConnection|null} */
+let nc = null;
+/** @type {import('nats').JetStreamClient|null} */
+let js = null;
+/** @type {import('nats').JetStreamManager|null} */
+let jsm = null;
+
+/**
+ * Connect to NATS and return JetStream handles.
+ *
+ * @param {string} [url=config.natsUrl]
+ * @returns {Promise<{ nc: import('nats').NatsConnection, js: import('nats').JetStreamClient, jsm: import('nats').JetStreamManager }>}
+ */
+export async function connectNats(url = config.natsUrl) {
+  if (!config.natsEnabled && process.env.NATS_FORCE_CONNECT !== '1') {
+    log.debug('connect skipped — NATS_ENABLED=false');
+    throw new Error('NATS is disabled (NATS_ENABLED=false)');
+  }
+
+  if (nc && js && jsm) {
+    return { nc, js, jsm };
+  }
+
+  log.info('connecting to NATS', { url });
+
+  const { connect } = await import('nats');
+  nc = await connect({ servers: url, name: 'verae-zapier-middleware' });
+  js = nc.jetstream();
+  jsm = await nc.jetstreamManager();
+
+  log.info('NATS connected', { url });
+  return { nc, js, jsm };
+}
+
+/** @returns {import('nats').NatsConnection|null} */
+export function getNatsConnection() {
+  return nc;
+}
+
+/**
+ * Idempotently create JetStream streams required by this middleware.
+ *
+ * @param {import('nats').JetStreamManager} [manager]
+ * @returns {Promise<void>}
+ */
+export async function ensureStreams(manager) {
+  const m = manager ?? jsm;
+  if (!m) {
+    throw new Error('JetStream manager not available — call connectNats first');
+  }
+
+  /** @type {Array<{ name: string, subjects: string[] }>} */
+  const defs = [
+    { name: STREAMS.ZAPIER_JOBS, subjects: [SUBJECTS.JOBS_WATCH] },
+    { name: STREAMS.ZAPIER_EVENTS, subjects: [SUBJECTS.JOBS_EVENTS] },
+    { name: STREAMS.ZAPIER_WEBHOOKS, subjects: [SUBJECTS.WEBHOOKS_DELIVER] },
+  ];
+
+  for (const def of defs) {
+    try {
+      await m.streams.info(def.name);
+      log.debug('stream exists', { stream: def.name });
+    } catch {
+      await m.streams.add({
+        name: def.name,
+        subjects: def.subjects,
+        retention: 'limits',
+        storage: 'file',
+        max_age: 24 * 60 * 60 * 1e9, // 24h in ns
+        num_replicas: 1,
+      });
+      log.info('stream created', { stream: def.name, subjects: def.subjects });
+    }
+  }
+}
+
+/**
+ * Close the shared NATS connection if open.
+ * @returns {Promise<void>}
+ */
+export async function closeNats() {
+  if (!nc) {
+    log.debug('closeNats: no active connection');
+    return;
+  }
+  log.info('closing NATS connection');
+  await nc.drain();
+  nc = null;
+  js = null;
+  jsm = null;
+}
+
+/**
+ * @returns {import('nats').JetStreamClient|null}
+ */
+export function getJetStream() {
+  return js;
+}
+
+/**
+ * @returns {boolean}
+ */
+export function isNatsConnected() {
+  return Boolean(nc && !nc.isClosed());
+}