fix(relay): make the HTTP fallback work and handle SIGTERM

relay.ts

@@ -1,7 +1,7 @@
 import { createHmac } from 'node:crypto';
 import { WebSocket } from 'ws';
 
-interface RelayConfig {
+export interface RelayConfig {
   wsUrl: string;
   token: string;
   hermesWebhookUrl: string;
@@ -54,21 +54,26 @@ async function forwardToHermes(args: {
   return { ok: res.ok, status: res.status, responseBody };
 }
 
-async function httpPullAndForward(config: RelayConfig): Promise<void> {
+export async function httpPullAndForward(config: RelayConfig): Promise<void> {
   try {
     const pullRes = await fetch(config.httpFallback.pullUrl, {
-      method: 'GET',
+      method: 'POST',
       headers: {
         authorization: `Bearer ${config.token}`,
+        'content-type': 'application/json',
       },
+      body: JSON.stringify({ maxEvents: config.httpFallback.maxEvents }),
     });
 
     if (!pullRes.ok) {
       console.error('[relay] HTTP pull failed:', pullRes.status, await pullRes.text());
       return;
     }
 
-    const events = (await pullRes.json()) as Array<{ id: string; body: string; contentType?: string }>;
+    const payload = (await pullRes.json()) as {
+      events?: Array<{ id: string; body: string; contentType?: string }>;
+    };
+    const events = payload.events;
     if (!Array.isArray(events) || events.length === 0) return;
 
     const ackedIds: string[] = [];
@@ -203,16 +208,24 @@ function startWebSocketRelay(config: RelayConfig): void {
     if (!connected) httpPullAndForward(config);
   }, config.httpFallback.intervalSeconds * 1000);
 
-  // Graceful shutdown
-  process.on('SIGINT', () => {
-    console.log('[relay] Shutting down...');
+  registerShutdown(() => {
     if (reconnectTimer) clearTimeout(reconnectTimer);
     if (httpFallbackTimer) clearInterval(httpFallbackTimer);
     ws?.close();
-    process.exit(0);
   });
 }
 
+/** systemd stops the service with SIGTERM; Ctrl-C sends SIGINT. Handle both. */
+export function registerShutdown(cleanup: () => void): void {
+  for (const signal of ['SIGINT', 'SIGTERM'] as const) {
+    process.on(signal, () => {
+      console.log(`[relay] ${signal} received, shutting down...`);
+      cleanup();
+      process.exit(0);
+    });
+  }
+}
+
 async function main() {
   const config = loadConfig();
   console.log('[relay] Starting clawproxy relay');
@@ -221,7 +234,10 @@ async function main() {
   startWebSocketRelay(config);
 }
 
-main().catch((err) => {
-  console.error('[relay] Fatal error:', err);
-  process.exit(1);
-});
+// Only run when executed directly (`yarn relay`), not when imported by tests.
+if (process.argv[1] && /relay\.ts$/.test(process.argv[1])) {
+  main().catch((err) => {
+    console.error('[relay] Fatal error:', err);
+    process.exit(1);
+  });
+}

tests/relay/relay.test.ts

@@ -0,0 +1,82 @@
+import { createHmac } from 'node:crypto';
+
+import { afterEach, describe, expect, test, vi } from 'vitest';
+
+import { httpPullAndForward, registerShutdown, type RelayConfig } from '@/relay';
+
+const config: RelayConfig = {
+  wsUrl: 'wss://clawproxy.test/api/nodes/ws',
+  token: 'cpn_test',
+  hermesWebhookUrl: 'http://hermes.test/webhook',
+  hermesWebhookSecret: 'secret',
+  httpFallback: {
+    pullUrl: 'https://clawproxy.test/api/nodes/pull',
+    ackUrl: 'https://clawproxy.test/api/nodes/ack',
+    intervalSeconds: 30,
+    maxEvents: 7,
+  },
+};
+
+function jsonResponse(body: unknown, status = 200) {
+  return new Response(JSON.stringify(body), {
+    status,
+    headers: { 'content-type': 'application/json' },
+  });
+}
+
+afterEach(() => {
+  vi.unstubAllGlobals();
+  vi.restoreAllMocks();
+});
+
+describe('httpPullAndForward', () => {
+  test('pulls with POST, the node token, and maxEvents', async () => {
+    const fetchMock = vi.fn().mockResolvedValue(jsonResponse({ ok: true, nodeId: 'n', events: [] }));
+    vi.stubGlobal('fetch', fetchMock);
+
+    await httpPullAndForward(config);
+
+    const [url, init] = fetchMock.mock.calls[0] as [string, RequestInit];
+    expect(url).toBe(config.httpFallback.pullUrl);
+    expect(init.method).toBe('POST');
+    expect((init.headers as Record<string, string>).authorization).toBe('Bearer cpn_test');
+    expect(JSON.parse(init.body as string)).toEqual({ maxEvents: 7 });
+  });
+
+  test('forwards events from the { events } response to Hermes, then acks them', async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce(
+        jsonResponse({ ok: true, nodeId: 'n', events: [{ id: 'evt-1', body: '{"a":1}' }] }),
+      )
+      .mockResolvedValueOnce(new Response('ok', { status: 200 }))
+      .mockResolvedValueOnce(jsonResponse({ ok: true }));
+    vi.stubGlobal('fetch', fetchMock);
+    vi.spyOn(console, 'log').mockImplementation(() => {});
+
+    await httpPullAndForward(config);
+
+    const [hermesUrl, hermesInit] = fetchMock.mock.calls[1] as [string, RequestInit];
+    expect(hermesUrl).toBe(config.hermesWebhookUrl);
+    expect(hermesInit.body).toBe('{"a":1}');
+    const expectedSig = createHmac('sha256', 'secret').update('{"a":1}').digest('hex');
+    expect((hermesInit.headers as Record<string, string>)['x-webhook-signature']).toBe(expectedSig);
+
+    const [ackUrl, ackInit] = fetchMock.mock.calls[2] as [string, RequestInit];
+    expect(ackUrl).toBe(config.httpFallback.ackUrl);
+    expect(ackInit.method).toBe('POST');
+    expect(JSON.parse(ackInit.body as string)).toEqual({ eventIds: ['evt-1'] });
+  });
+});
+
+describe('registerShutdown', () => {
+  test('handles SIGTERM (what systemd sends) as well as SIGINT', () => {
+    const onSpy = vi.spyOn(process, 'on').mockImplementation(() => process);
+
+    registerShutdown(() => {});
+
+    const signals = onSpy.mock.calls.map(([event]) => event);
+    expect(signals).toContain('SIGTERM');
+    expect(signals).toContain('SIGINT');
+  });
+});