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'); + }); +});