superlog/apps/proxy/src/ingest-queue.test.ts

import { strict as assert } from "node:assert";
import { test } from "node:test";
import {
  PoisonMessageError,
  describeCollectorFailure,
  encodeIngestMessage,
  getIngestQueueConfig,
  isPermanentCollectorFailure,
  isPoisonMessageError,
  parseIngestMessage,
  queueDeliveryMetricFromParsedMessage,
} from "./ingest-queue.js";

const baseInput = {
  path: "/v1/logs",
  projectId: "project-1",
  contentType: "application/x-protobuf",
  body: Buffer.from("hello"),
};

test("getIngestQueueConfig leaves direct forwarding enabled by default", () => {
  assert.equal(getIngestQueueConfig({}), null);
});

test("getIngestQueueConfig reads bounded consumer concurrency", () => {
  const config = getIngestQueueConfig({
    INGEST_QUEUE_URL: "https://sqs.us-west-2.amazonaws.com/123/superlog-test-ingest",
    INGEST_QUEUE_CONSUMER_CONCURRENCY: "99",
  });

  assert.equal(config?.consumerConcurrency, 32);
});

test("encodeIngestMessage stores small payloads inline", () => {
  const encoded = encodeIngestMessage(
    baseInput,
    {
      maxMessageBytes: 240_000,
      oversizePrefix: "otlp-oversize",
    },
    new Date("2026-05-24T00:00:00.000Z"),
    "payload-1",
  );

  assert.equal(encoded.storage, "inline");
  assert.equal(encoded.s3Object, undefined);

  const message = JSON.parse(encoded.messageBody);
  assert.equal(message.body.storage, "inline");
  assert.equal(Buffer.from(message.body.base64, "base64").toString("utf8"), "hello");
});

test("encodeIngestMessage offloads payloads that would exceed the SQS envelope budget", () => {
  const encoded = encodeIngestMessage(
    { ...baseInput, body: Buffer.alloc(32, 7), contentEncoding: "gzip" },
    {
      maxMessageBytes: 10,
      oversizeBucket: "superlog-ingest-oversize",
      oversizePrefix: "/otlp-oversize/",
    },
    new Date("2026-05-24T12:34:56.000Z"),
    "payload-2",
  );

  assert.equal(encoded.storage, "s3");
  assert.equal(encoded.s3Object?.bucket, "superlog-ingest-oversize");
  assert.equal(encoded.s3Object?.key, "otlp-oversize/2026/05/24/payload-2.otlp");
  assert.equal(encoded.s3Object?.body.byteLength, 32);

  const message = JSON.parse(encoded.messageBody);
  assert.equal(message.body.storage, "s3");
  assert.equal(message.body.bucket, "superlog-ingest-oversize");
  assert.equal(message.body.key, "otlp-oversize/2026/05/24/payload-2.otlp");
  assert.equal(message.contentEncoding, "gzip");
});

test("describeCollectorFailure includes bounded collector response body", () => {
  const failure = describeCollectorFailure(400, "bad request: invalid OTLP payload");

  assert.equal(failure.message, "collector returned 400: bad request: invalid OTLP payload");
  assert.equal(failure.status, 400);
  assert.equal(failure.body, "bad request: invalid OTLP payload");
});

test("isPermanentCollectorFailure drops 4xx the collector can never accept on retry", () => {
  // A real prod incident: a client sent http.response_content_length above uint64 max,
  // the collector 400s the whole batch, and the consumer cycled the same message every
  // 900s (visibility timeout) for ~12.5h before the DLQ — pinning oldest-message-age into
  // a sawtooth. Re-sending identical bytes can never succeed, so 4xx must be dropped.
  assert.equal(isPermanentCollectorFailure(400), true);
  assert.equal(isPermanentCollectorFailure(401), true);
  assert.equal(isPermanentCollectorFailure(403), true);
  assert.equal(isPermanentCollectorFailure(404), true);
  assert.equal(isPermanentCollectorFailure(413), true);
  assert.equal(isPermanentCollectorFailure(422), true);
});

test("isPermanentCollectorFailure keeps retryable statuses on the queue", () => {
  // 408/429 are backpressure/timeout signals and 5xx are transient — the same payload can
  // succeed once the collector recovers, so these stay in the queue for redelivery.
  assert.equal(isPermanentCollectorFailure(408), false);
  assert.equal(isPermanentCollectorFailure(429), false);
  assert.equal(isPermanentCollectorFailure(500), false);
  assert.equal(isPermanentCollectorFailure(502), false);
  assert.equal(isPermanentCollectorFailure(503), false);
  assert.equal(isPermanentCollectorFailure(504), false);
  // Not a failure status at all — defensively never treated as a permanent drop.
  assert.equal(isPermanentCollectorFailure(200), false);
});

test("parseIngestMessage round-trips a well-formed inline message", () => {
  const encoded = encodeIngestMessage(
    baseInput,
    { maxMessageBytes: 240_000, oversizePrefix: "otlp-oversize" },
    new Date("2026-05-24T00:00:00.000Z"),
    "payload-ok",
  );

  const parsed = parseIngestMessage(encoded.messageBody);
  assert.equal(parsed.path, "/v1/logs");
  assert.equal(parsed.body.storage, "inline");
});

test("parseIngestMessage flags an empty inline body as a poison message", () => {
  // An empty OTLP HTTP body base64-encodes to "" — there is nothing to forward and
  // no retry can ever succeed, so it must be classified as poison (delete, not cycle).
  const emptyBody = JSON.stringify({
    version: 1,
    kind: "otlp",
    path: "/v1/logs",
    projectId: "project-1",
    contentType: "application/json",
    receivedAt: "2026-05-29T00:00:00.000Z",
    body: { storage: "inline", base64: "" },
  });

  assert.throws(() => parseIngestMessage(emptyBody), PoisonMessageError);
  try {
    parseIngestMessage(emptyBody);
  } catch (err) {
    assert.equal(isPoisonMessageError(err), true);
  }
});

test("parseIngestMessage flags non-JSON message bodies as poison", () => {
  assert.throws(() => parseIngestMessage("not json at all"), PoisonMessageError);
});

test("parseIngestMessage flags valid JSON with malformed envelope fields as poison", () => {
  // Valid JSON, but `path` is a number — assertIngestMessage's `.startsWith` would throw a
  // TypeError. A structurally-broken envelope can never succeed on retry, so it must be poison.
  const malformed = JSON.stringify({
    version: 1,
    kind: "otlp",
    path: 123,
    projectId: "project-1",
    contentType: "application/json",
    body: { storage: "inline", base64: "aGVsbG8=" },
  });

  assert.throws(() => parseIngestMessage(malformed), PoisonMessageError);
});

test("isPoisonMessageError discriminates transient errors", () => {
  assert.equal(isPoisonMessageError(new Error("collector returned 503")), false);
  assert.equal(isPoisonMessageError(new PoisonMessageError("bad")), true);
});

test("queueDeliveryMetricFromParsedMessage ignores malformed parsed payloads", () => {
  assert.equal(
    queueDeliveryMetricFromParsedMessage(
      { path: "/v1/logs", projectId: "project-1", body: {} },
      "delivery_error",
      undefined,
      12,
      34,
    ),
    null,
  );
});