triggerdotdev--trigger.dev/apps/webapp/test/queueMetricsMapping.test.ts

import { describe, expect, it } from "vitest";
import {
  descriptorFromQueue,
  mapEntryToRows,
  OVERFLOW_QUEUE_NAME,
  QueueNameLimiter,
} from "~/v3/queueMetricsMapping";

describe("descriptorFromQueue", () => {
  it("parses a plain descriptor", () => {
    expect(descriptorFromQueue("{org:o1}:proj:p1:env:e1:queue:task/my-task")).toEqual({
      organization_id: "o1",
      project_id: "p1",
      environment_id: "e1",
      queue_name: "task/my-task",
      concurrency_key: "",
    });
  });

  it("captures a concurrency-key suffix", () => {
    expect(descriptorFromQueue("{org:o1}:proj:p1:env:e1:queue:task/t:ck:tenant-3")).toEqual(
      expect.objectContaining({ queue_name: "task/t", concurrency_key: "tenant-3" })
    );
  });

  it("maps the ck wildcard to no key", () => {
    expect(descriptorFromQueue("{org:o1}:proj:p1:env:e1:queue:task/t:ck:*")).toEqual(
      expect.objectContaining({ queue_name: "task/t", concurrency_key: "" })
    );
  });

  it("keeps colons inside the queue name", () => {
    expect(descriptorFromQueue("{org:o1}:proj:p1:env:e1:queue:my:odd:queue")).toEqual(
      expect.objectContaining({ queue_name: "my:odd:queue", concurrency_key: "" })
    );
  });

  it("keeps colons in the name while capturing a real ck suffix", () => {
    expect(descriptorFromQueue("{org:o1}:proj:p1:env:e1:queue:a:b:ck:t9")).toEqual(
      expect.objectContaining({ queue_name: "a:b", concurrency_key: "t9" })
    );
  });

  it("rejects malformed descriptors", () => {
    expect(descriptorFromQueue("not-a-descriptor")).toBeNull();
    expect(descriptorFromQueue("{org:o1}:proj:p1:env:e1")).toBeNull();
    expect(descriptorFromQueue("")).toBeNull();
  });
});

describe("QueueNameLimiter", () => {
  it("passes names through under the cap and overflows past it, per scope", () => {
    const limiter = new QueueNameLimiter(2);
    expect(limiter.limit("env1", "a")).toBe("a");
    expect(limiter.limit("env1", "b")).toBe("b");
    expect(limiter.limit("env1", "c")).toBe(OVERFLOW_QUEUE_NAME);
    expect(limiter.limit("env1", "a")).toBe("a");
    expect(limiter.limit("env2", "c")).toBe("c");
  });

  it("is unlimited when the cap is 0", () => {
    const limiter = new QueueNameLimiter(0);
    for (let i = 0; i < 100; i++) {
      expect(limiter.limit("env1", `q${i}`)).toBe(`q${i}`);
    }
  });

  it("evicts the oldest scope when the scope map is full", () => {
    const limiter = new QueueNameLimiter(1, 2);
    expect(limiter.limit("env1", "a")).toBe("a");
    expect(limiter.limit("env2", "a")).toBe("a");
    expect(limiter.limit("env3", "a")).toBe("a");
    expect(limiter.limit("env1", "b")).toBe("b");
  });
});

describe("mapEntryToRows", () => {
  const q = "{org:o1}:proj:p1:env:e1:queue:task/t";

  it("maps a gauge entry with numeric fields", () => {
    const rows = mapEntryToRows({
      id: "1700000000000-0",
      fields: {
        op: "gauge",
        q,
        ql: "5",
        cc: "2",
        lim: "10",
        eql: "7",
        ec: "3",
        elim: "20",
        thr: "1",
      },
    });
    expect(rows).toHaveLength(1);
    expect(rows[0]).toEqual(
      expect.objectContaining({
        op: "gauge",
        organization_id: "o1",
        queue_name: "task/t",
        concurrency_key: "",
        queued: 5,
        running: 2,
        queue_limit: 10,
        env_queued: 7,
        env_running: 3,
        env_limit: 20,
        throttled: 1,
      })
    );
    expect(rows[0]!.event_time).toBe("2023-11-14 22:13:20");
    expect(rows[0]!.ck_backlogged).toBeUndefined();
    expect(rows[0]!.ck_max_wait_ms).toBeUndefined();
  });

  it("keeps the key on per-subqueue gauges and maps the CK-health tail", () => {
    const rows = mapEntryToRows({
      id: "1700000000000-0",
      fields: { op: "gauge", q: `${q}:ck:tenant-1`, ql: "4", ckq: "3", ckw: "2500" },
    });
    expect(rows).toHaveLength(1);
    expect(rows[0]).toEqual(
      expect.objectContaining({
        op: "gauge",
        queue_name: "task/t",
        concurrency_key: "tenant-1",
        queued: 4,
        ck_backlogged: 3,
        ck_max_wait_ms: 2500,
      })
    );
  });

  it("maps started with wait_ms + cumulative and drops unknown ops", () => {
    const started = mapEntryToRows({
      id: "1700000000000-0",
      fields: { op: "started", q, wait: "48", cum: "512" },
    });
    expect(started).toHaveLength(1);
    expect(started[0]).toEqual(
      expect.objectContaining({
        op: "started",
        wait_ms: 48,
        cumulative: 512,
        order_key: (1700000000000n * 1000000n).toString(),
      })
    );
    expect(mapEntryToRows({ id: "1-0", fields: { op: "ack", q, cum: "9" } })[0]).toEqual(
      expect.objectContaining({ op: "ack", cumulative: 9 })
    );
    expect(mapEntryToRows({ id: "1-0", fields: { op: "bogus", q } })).toEqual([]);
    expect(mapEntryToRows({ id: "1-0", fields: { op: "ack" } })).toEqual([]);
  });

  it("expands a dual-odometer counter entry into base + per-key rows", () => {
    const rows = mapEntryToRows({
      id: "1700000000000-3",
      fields: { op: "started", q, ck: "tenant-9", wait: "80", cum: "41", ckcum: "7" },
    });
    expect(rows).toHaveLength(2);
    expect(rows[0]).toEqual(
      expect.objectContaining({ queue_name: "task/t", cumulative: 41, wait_ms: 80 })
    );
    expect(rows[0]!.concurrency_key).toBeUndefined();
    expect(rows[1]).toEqual(
      expect.objectContaining({
        queue_name: "task/t",
        concurrency_key: "tenant-9",
        cumulative: 7,
        wait_ms: 80,
      })
    );
    expect(rows[0]!.order_key).toBe(rows[1]!.order_key);

    // Baseline entries carry exactly one odometer each.
    const baseBaseline = mapEntryToRows({ id: "1-0", fields: { op: "started", q, cum: "0" } });
    expect(baseBaseline).toHaveLength(1);
    expect(baseBaseline[0]!.concurrency_key).toBeUndefined();
    const ckBaseline = mapEntryToRows({
      id: "1-1",
      fields: { op: "started", q, ck: "tenant-9", ckcum: "0" },
    });
    expect(ckBaseline).toHaveLength(1);
    expect(ckBaseline[0]).toEqual(
      expect.objectContaining({ concurrency_key: "tenant-9", cumulative: 0 })
    );
  });

  it("applies the queue-name limiter: gauges overflow, counters drop", () => {
    const limiters = { queueNames: new QueueNameLimiter(1) };
    const first = mapEntryToRows({ id: "1-0", fields: { op: "ack", q, cum: "1" } }, limiters);
    expect(first[0]!.queue_name).toBe("task/t");

    // Overflowed gauges keep flowing under the shared name (max stays meaningful),
    // with per-key attribution stripped.
    const overflowGauge = mapEntryToRows(
      {
        id: "1-1",
        fields: { op: "gauge", q: "{org:o1}:proj:p1:env:e1:queue:task/other:ck:t1", ql: "3" },
      },
      limiters
    );
    expect(overflowGauge[0]!.queue_name).toBe(OVERFLOW_QUEUE_NAME);
    expect(overflowGauge[0]!.concurrency_key).toBe("");

    // Overflowed counters are dropped: merging distinct odometers under one key
    // produces garbage deltas.
    const overflowCounter = mapEntryToRows(
      { id: "1-2", fields: { op: "ack", q: "{org:o1}:proj:p1:env:e1:queue:task/other", cum: "4" } },
      limiters
    );
    expect(overflowCounter).toEqual([]);
  });

  it("applies the concurrency-key limiter: overflow drops the per-key row, keeps base", () => {
    const limiters = { concurrencyKeys: new QueueNameLimiter(1) };
    const first = mapEntryToRows(
      { id: "1-0", fields: { op: "ack", q, ck: "t1", cum: "5", ckcum: "2" } },
      limiters
    );
    expect(first).toHaveLength(2);

    const overflowed = mapEntryToRows(
      { id: "1-1", fields: { op: "ack", q, ck: "t2", cum: "6", ckcum: "1" } },
      limiters
    );
    expect(overflowed).toHaveLength(1);
    expect(overflowed[0]!.cumulative).toBe(6);
    expect(overflowed[0]!.concurrency_key).toBeUndefined();

    // Gauge for an overflowed key keeps the row but loses the attribution.
    const overflowGauge = mapEntryToRows(
      { id: "1-2", fields: { op: "gauge", q: `${q}:ck:t3`, ql: "2" } },
      limiters
    );
    expect(overflowGauge).toHaveLength(1);
    expect(overflowGauge[0]!.concurrency_key).toBe("");
  });
});