From ae76a8d7e6d865607b1d813aaf9267e21141d5d2 Mon Sep 17 00:00:00 2001 From: Eric Dahl Date: Tue, 22 Sep 2026 11:51:15 -0400 Subject: [PATCH] fix(relay): make the HTTP fallback work and handle SIGTERM The relay's HTTP fallback has never delivered anything: - it called the pull endpoint with GET, but the route only accepts POST - it expected a bare array, but the route returns { ok, nodeId, events } It now POSTs { maxEvents } and reads `events` from the response. systemd stops the service with SIGTERM, but only SIGINT was handled, so the graceful shutdown never ran. Both signals now go through registerShutdown(). main() now runs only when relay.ts is executed directly, so the module can be imported by tests. --- relay.ts | 40 +++++++++++++------ tests/relay/relay.test.ts | 82 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 12 deletions(-) create mode 100644 tests/relay/relay.test.ts diff --git a/relay.ts b/relay.ts index 0e1bfab..1ec05c1 100644 --- a/relay.ts +++ b/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,13 +54,15 @@ async function forwardToHermes(args: { return { ok: res.ok, status: res.status, responseBody }; } -async function httpPullAndForward(config: RelayConfig): Promise { +export async function httpPullAndForward(config: RelayConfig): Promise { 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) { @@ -68,7 +70,10 @@ async function httpPullAndForward(config: RelayConfig): Promise { 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); + }); +} diff --git a/tests/relay/relay.test.ts b/tests/relay/relay.test.ts new file mode 100644 index 0000000..ea812b9 --- /dev/null +++ b/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).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)['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'); + }); +});