processor/test/live-server.test.ts

/**
 * Unit tests for src/live/server.ts — WebSocket server scaffold + heartbeat.
 *
 * Uses a real `ws` client against an in-process server bound on a random port.
 * No Redis or Postgres required. The message handler is stubbed.
 *
 * Covers:
 *  - Server starts on a random port and accepts a WS connection.
 *  - Server sends a ping within PING_INTERVAL_MS + 100ms; pong updates lastSeenAt.
 *  - Inbound message that fails zod validation receives protocol-violation error
 *    and the connection stays open.
 *  - Valid subscribe message reaches the onMessage handler.
 *  - stop() sends a close frame to existing connections and resolves within
 *    the drain timeout.
 *  - processor_live_connections gauge increments on connect and decrements on close.
 */

import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import WebSocket from 'ws';
import type { Logger } from 'pino';
import type { Config } from '../src/config/load.js';
import type { Metrics } from '../src/core/types.js';
import { createLiveServer } from '../src/live/server.js';
import type { LiveConnection, MessageHandler } from '../src/live/server.js';
import type { InboundMessage } from '../src/live/protocol.js';

// ---------------------------------------------------------------------------
// Test helpers
// ---------------------------------------------------------------------------

function makeSilentLogger(): Logger {
  return {
    debug: vi.fn(),
    info: vi.fn(),
    warn: vi.fn(),
    error: vi.fn(),
    fatal: vi.fn(),
    trace: vi.fn(),
    child: vi.fn().mockReturnThis(),
    level: 'silent',
    silent: vi.fn(),
  } as unknown as Logger;
}

type TestMetrics = Metrics & {
  readonly incCalls: Array<{ name: string; labels?: Record<string, string>; value?: number }>;
  readonly observeCalls: Array<{ name: string; value: number }>;
};

function makeMetrics(): TestMetrics {
  const incCalls: Array<{ name: string; labels?: Record<string, string>; value?: number }> = [];
  const observeCalls: Array<{ name: string; value: number }> = [];
  return {
    incCalls,
    observeCalls,
    inc(name, labels, value) {
      incCalls.push({ name, labels, value });
    },
    observe(name, value) {
      observeCalls.push({ name, value });
    },
  };
}

function makeConfig(overrides: Partial<Config> = {}): Config {
  return {
    NODE_ENV: 'test',
    INSTANCE_ID: 'test-1',
    LOG_LEVEL: 'silent',
    REDIS_URL: 'redis://localhost:6379',
    POSTGRES_URL: 'postgres://localhost:5432/test',
    REDIS_TELEMETRY_STREAM: 'telemetry:t',
    REDIS_CONSUMER_GROUP: 'processor',
    REDIS_CONSUMER_NAME: 'test-consumer',
    METRICS_PORT: 0,
    BATCH_SIZE: 100,
    BATCH_BLOCK_MS: 500,
    WRITE_BATCH_SIZE: 50,
    DEVICE_STATE_LRU_CAP: 10_000,
    LIVE_WS_PORT: 0, // OS-assigned port
    LIVE_WS_HOST: '127.0.0.1',
    LIVE_WS_PING_INTERVAL_MS: 200, // short for tests
    LIVE_WS_DRAIN_TIMEOUT_MS: 500,
    LIVE_WS_BACKPRESSURE_THRESHOLD_BYTES: 1_048_576,
    DIRECTUS_BASE_URL: 'http://localhost:8055',
    DIRECTUS_AUTH_TIMEOUT_MS: 5_000,
    DIRECTUS_AUTHZ_TIMEOUT_MS: 5_000,
    LIVE_BROADCAST_GROUP_PREFIX: 'live-broadcast',
    LIVE_BROADCAST_BATCH_SIZE: 100,
    LIVE_BROADCAST_BATCH_BLOCK_MS: 1_000,
    LIVE_DEVICE_EVENT_REFRESH_MS: 30_000,
    ...overrides,
  };
}

/**
 * Connects a ws client to the given URL and resolves when the connection is open.
 */
function connectClient(url: string): Promise<WebSocket> {
  return new Promise((resolve, reject) => {
    const ws = new WebSocket(url);
    ws.once('open', () => resolve(ws));
    ws.once('error', reject);
  });
}

/**
 * Waits for the next message on the given WebSocket and returns its parsed JSON.
 */
function waitForMessage(ws: WebSocket, timeoutMs = 3_000): Promise<unknown> {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error('timeout waiting for message')), timeoutMs);
    ws.once('message', (data) => {
      clearTimeout(timer);
      resolve(JSON.parse(data.toString()));
    });
  });
}

/**
 * Waits for the next pong frame on the ws client socket.
 */
function waitForPing(ws: WebSocket, timeoutMs = 3_000): Promise<void> {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error('timeout waiting for ping')), timeoutMs);
    // ws client emits 'ping' when a server ping frame arrives.
    ws.once('ping', () => {
      clearTimeout(timer);
      resolve();
    });
  });
}

/**
 * Waits for the ws client to receive a close frame.
 */
function waitForClose(ws: WebSocket, timeoutMs = 3_000): Promise<number> {
  return new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error('timeout waiting for close')), timeoutMs);
    ws.once('close', (code) => {
      clearTimeout(timer);
      resolve(code);
    });
  });
}

// ---------------------------------------------------------------------------
// Tests — server with discoverable port
// ---------------------------------------------------------------------------

import * as httpModule from 'node:http';
import type { AddressInfo } from 'node:net';

/**
 * Finds a free TCP port by letting the OS assign one, then closing the listener.
 */
function getFreePort(): Promise<number> {
  return new Promise((resolve, reject) => {
    const srv = new httpModule.Server();
    srv.listen(0, '127.0.0.1', () => {
      const addr = srv.address() as AddressInfo;
      const p = addr.port;
      srv.close((err) => (err ? reject(err) : resolve(p)));
    });
  });
}

describe('live server — lifecycle and message routing', () => {
  let server: ReturnType<typeof createLiveServer>;
  let wsUrl: string;
  let clients: WebSocket[] = [];
  let metrics: TestMetrics;
  let capturedConnections: LiveConnection[] = [];
  let capturedMessages: Array<{ conn: LiveConnection; msg: InboundMessage }> = [];

  beforeEach(async () => {
    clients = [];
    capturedConnections = [];
    capturedMessages = [];

    const logger = makeSilentLogger();
    metrics = makeMetrics();
    const port = await getFreePort();

    const handler: MessageHandler = async (conn, msg) => {
      capturedConnections.push(conn);
      capturedMessages.push({ conn, msg });
      // Echo not-implemented for subscribe
      conn.ws.send(JSON.stringify({ type: 'error', code: 'not-implemented' }));
    };

    const config = makeConfig({
      LIVE_WS_PORT: port,
      LIVE_WS_HOST: '127.0.0.1',
      LIVE_WS_PING_INTERVAL_MS: 200,
      LIVE_WS_DRAIN_TIMEOUT_MS: 500,
    });

    server = createLiveServer(config, logger, metrics, handler);
    await server.start();
    wsUrl = `ws://127.0.0.1:${port}`;
  });

  afterEach(async () => {
    for (const client of clients) {
      if (client.readyState !== WebSocket.CLOSED) {
        client.terminate();
      }
    }
    await server.stop(200).catch(() => {});
  });

  it('accepts a WS connection', async () => {
    const client = await connectClient(wsUrl);
    clients.push(client);
    expect(client.readyState).toBe(WebSocket.OPEN);
  });

  it('sends a ping within PING_INTERVAL_MS + buffer', async () => {
    const client = await connectClient(wsUrl);
    clients.push(client);
    // PING_INTERVAL_MS = 200ms; allow 300ms total.
    await waitForPing(client, 300);
    // If we get here without timeout, the ping arrived.
  });

  it('inbound message failing zod validation receives protocol-violation error; connection stays open', async () => {
    const client = await connectClient(wsUrl);
    clients.push(client);

    // Send a JSON object with an unknown type — zod will reject.
    client.send(JSON.stringify({ type: 'unknown-action', data: 42 }));
    const msg = await waitForMessage(client, 2_000) as Record<string, unknown>;

    expect(msg['type']).toBe('error');
    expect(msg['code']).toBe('protocol-violation');
    expect(client.readyState).toBe(WebSocket.OPEN);
  });

  it('inbound malformed JSON receives protocol-violation error', async () => {
    const client = await connectClient(wsUrl);
    clients.push(client);

    client.send('not valid json {{{');
    const msg = await waitForMessage(client, 2_000) as Record<string, unknown>;

    expect(msg['type']).toBe('error');
    expect(msg['code']).toBe('protocol-violation');
    expect(client.readyState).toBe(WebSocket.OPEN);
  });

  it('valid subscribe message reaches the onMessage handler', async () => {
    const client = await connectClient(wsUrl);
    clients.push(client);

    const subMsg = {
      type: 'subscribe',
      topic: 'event:ada60b3d-b29f-4017-b702-cd6b700f9f6c',
      id: 'corr-1',
    };
    client.send(JSON.stringify(subMsg));
    await waitForMessage(client, 2_000); // Wait for not-implemented reply

    expect(capturedMessages.length).toBe(1);
    const captured = capturedMessages[0];
    expect(captured).toBeDefined();
    expect(captured!.msg.type).toBe('subscribe');
    expect(captured!.msg.topic).toBe('event:ada60b3d-b29f-4017-b702-cd6b700f9f6c');
  });

  it('processor_live_connections observe is called on connect and disconnect', async () => {
    const before = metrics.observeCalls.filter(
      (c) => c.name === 'processor_live_connections',
    ).length;

    const client = await connectClient(wsUrl);
    clients.push(client);

    // Wait briefly for the connection event.
    await new Promise<void>((resolve) => setTimeout(resolve, 50));

    const afterConnect = metrics.observeCalls.filter(
      (c) => c.name === 'processor_live_connections',
    );
    expect(afterConnect.length).toBeGreaterThan(before);
    // After connect, value should be >= 1.
    const lastAfterConnect = afterConnect[afterConnect.length - 1];
    expect(lastAfterConnect).toBeDefined();
    expect(lastAfterConnect!.value).toBeGreaterThanOrEqual(1);

    // Disconnect the client.
    const closePromise = waitForClose(client, 1_000);
    client.close();
    await closePromise;

    // Wait briefly for the close event to propagate.
    await new Promise<void>((resolve) => setTimeout(resolve, 50));

    const afterDisconnect = metrics.observeCalls.filter(
      (c) => c.name === 'processor_live_connections',
    );
    const lastAfterDisconnect = afterDisconnect[afterDisconnect.length - 1];
    expect(lastAfterDisconnect).toBeDefined();
    expect(lastAfterDisconnect!.value).toBe(0);
  });

  it('stop() sends close frame to existing connections and resolves within drain timeout', async () => {
    const client = await connectClient(wsUrl);
    clients.push(client);

    const closePromise = waitForClose(client, 2_000);
    const startMs = Date.now();
    await server.stop(500);
    const elapsedMs = Date.now() - startMs;

    // Should resolve within the drain timeout + a small buffer.
    expect(elapsedMs).toBeLessThan(1_000);

    // The client should have received a close frame.
    const code = await closePromise;
    // 1001 = GOING_AWAY (server shutting down).
    expect(code).toBe(1001);
  });
});