tcp-ingestion/test/publish.integration.test.ts

/**
 * Integration test: Redis Streams publisher round-trip via testcontainers.
 *
 * Spins up a real Redis 7 container, publishes a Position containing bigint
 * and Buffer attributes, XREADs it back, and verifies byte-perfect round-trip
 * after sentinel decoding.
 *
 * If Docker is unavailable (CI without Docker, local dev without Docker Desktop),
 * the test suite logs a clear message and skips — it does not fail the build.
 * Docker availability is established by a container start attempt, with the
 * skip condition set before any test runs.
 */

import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { GenericContainer, type StartedTestContainer } from 'testcontainers';
import type Redis from 'ioredis';
import type { Position } from '../src/core/types.js';
import { createPublisher, serializePosition } from '../src/core/publish.js';
import type { Config } from '../src/config/load.js';
import type { Logger } from 'pino';
import { vi } from 'vitest';

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

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

function makeConfig(overrides: Partial<Config> = {}): Config {
  return {
    NODE_ENV: 'test',
    INSTANCE_ID: 'test-integration',
    LOG_LEVEL: 'silent',
    TELTONIKA_PORT: 5027,
    REDIS_URL: 'redis://localhost:6379',
    REDIS_TELEMETRY_STREAM: 'telemetry:test',
    REDIS_STREAM_MAXLEN: 10_000,
    METRICS_PORT: 9090,
    PUBLISH_QUEUE_CAPACITY: 100,
    STRICT_DEVICE_AUTH: false,
    ...overrides,
  };
}

// ---------------------------------------------------------------------------
// Container lifecycle
// ---------------------------------------------------------------------------

let container: StartedTestContainer | null = null;
let redisClient: Redis | null = null;
let dockerAvailable = true;

beforeAll(async () => {
  try {
    container = await new GenericContainer('redis:7-alpine')
      .withExposedPorts(6379)
      .start();
  } catch {
    console.warn(
      '[publish.integration.test] Docker not available — skipping Redis integration tests',
    );
    dockerAvailable = false;
    return;
  }

  const mappedPort = container.getMappedPort(6379);
  const host = container.getHost();

  const { default: Redis } = await import('ioredis');
  redisClient = new Redis(`redis://${host}:${mappedPort}`, {
    enableOfflineQueue: false,
    lazyConnect: true,
    maxRetriesPerRequest: 0,
  });
  await redisClient.connect();
}, 60_000);

afterAll(async () => {
  if (redisClient) {
    await redisClient.quit().catch(() => {});
  }
  if (container) {
    await container.stop().catch(() => {});
  }
}, 30_000);

// ---------------------------------------------------------------------------
// Integration tests
// ---------------------------------------------------------------------------

describe('Redis Streams publisher — integration', () => {
  it('round-trips a Position with bigint and Buffer attributes via XADD/XREAD', async () => {
    if (!dockerAvailable || !redisClient) {
      console.warn('[publish.integration.test] skipping: Docker not available');
      return;
    }

    // Arrange: position with all attribute types
    const original: Position = {
      device_id: '356307042441013',
      timestamp: new Date('2024-06-15T12:00:00.000Z'),
      latitude: 54.687157,
      longitude: 25.279652,
      altitude: 130,
      angle: 90,
      speed: 45,
      satellites: 12,
      priority: 0,
      attributes: {
        num_attr: 255,
        big_attr: BigInt('18446744073709551615'), // u64 max
        buf_attr: Buffer.from([0xde, 0xad, 0xbe, 0xef]),
        event: 0,
      },
    };

    const stream = 'telemetry:test';
    const config = makeConfig({ REDIS_TELEMETRY_STREAM: stream });

    const publisher = createPublisher(
      redisClient,
      config,
      makeSilentLogger(),
      { inc: vi.fn(), observe: vi.fn() },
    );

    // Act: publish and wait for worker to drain
    await publisher.publish(original, '8E');
    await publisher.drain(5_000);

    // Assert: XREAD the record back from Redis
    const results = await redisClient.xread(
      'COUNT',
      '1',
      'STREAMS',
      stream,
      '0',
    );

    expect(results).not.toBeNull();
    expect(results).toHaveLength(1);

    const [_streamName, messages] = results![0]!;
    expect(messages).toHaveLength(1);

    const [_id, fieldValues] = messages[0]!;

    // fieldValues is a flat [k1, v1, k2, v2, ...] array from ioredis
    const record: Record<string, string> = {};
    for (let i = 0; i < fieldValues.length; i += 2) {
      record[fieldValues[i]!] = fieldValues[i + 1]!;
    }

    // Top-level fields
    expect(record['ts']).toBe('2024-06-15T12:00:00.000Z');
    expect(record['device_id']).toBe('356307042441013');
    expect(record['codec']).toBe('8E');

    // Payload round-trip
    const payload = JSON.parse(record['payload']!) as {
      device_id: string;
      timestamp: string;
      latitude: number;
      longitude: number;
      attributes: Record<string, unknown>;
    };

    expect(payload.device_id).toBe(original.device_id);
    expect(payload.latitude).toBe(original.latitude);
    expect(payload.longitude).toBe(original.longitude);

    // Sentinel decoding
    const bigSentinel = payload.attributes['big_attr'] as { __bigint: string };
    const bufSentinel = payload.attributes['buf_attr'] as { __buffer_b64: string };
    const numAttr = payload.attributes['num_attr'] as number;

    expect(BigInt(bigSentinel.__bigint)).toBe(BigInt('18446744073709551615'));
    expect(Buffer.from(bufSentinel.__buffer_b64, 'base64')).toEqual(
      Buffer.from([0xde, 0xad, 0xbe, 0xef]),
    );
    expect(numAttr).toBe(255);
  }, 30_000);

  it('serializePosition produces fields consumed correctly by XREAD', async () => {
    if (!dockerAvailable || !redisClient) {
      console.warn('[publish.integration.test] skipping: Docker not available');
      return;
    }

    const stream = 'telemetry:serialize-test';
    const pos: Position = {
      device_id: 'DIRECT123',
      timestamp: new Date('2024-01-01T00:00:00.000Z'),
      latitude: 0,
      longitude: 0,
      altitude: 0,
      angle: 0,
      speed: 0,
      satellites: 4,
      priority: 0,
      attributes: {},
    };

    const fields = serializePosition(pos, '16');
    const args: string[] = [];
    for (const [k, v] of Object.entries(fields)) {
      args.push(k, v);
    }

    // Push directly to verify field layout is correct
    await redisClient.xadd(stream, '*', ...args);

    const results = await redisClient.xread('COUNT', '1', 'STREAMS', stream, '0');
    expect(results).not.toBeNull();
    const [_sName, msgs] = results![0]!;
    const [_id, fv] = msgs[0]!;

    const record: Record<string, string> = {};
    for (let i = 0; i < fv.length; i += 2) {
      record[fv[i]!] = fv[i + 1]!;
    }

    expect(record['codec']).toBe('16');
    expect(record['device_id']).toBe('DIRECT123');
  }, 30_000);
});