Operating an event-driven workflow under failure.

describe("happy path", () => {
  it("keeps correlationId through submit, queue, and completion", () => {
    const telemetry = new InMemoryTelemetry();
    const engine = new WorkflowEngine(telemetry, { defaultMaxAttempts: 3 });

    const submitResult = engine.submit({
      workflowType: "analytics-export",
      correlationId: "corr-happy",
      payload: { accountId: "acct-1" }
    });

    expect(submitResult.status).toBe("queued");
    expect(submitResult.correlationId).toBe("corr-happy");

    const result = engine.handleNextMessage(() => ({
      kind: "success",
      detail: "completed"
    }));
    const finalRequest = engine.getRequest(submitResult.requestId);

    expect(result?.kind).toBe("success");
    expect(finalRequest?.status).toBe("completed");
    expect(finalRequest?.correlationId).toBe("corr-happy");
    expect(engine.getDlq()).toHaveLength(0);
  });
});
// snippet:test-happy-path-end

// snippet:test-transient-start
describe("transient retry", () => {
  it("retries and succeeds on second attempt", () => {
    const telemetry = new InMemoryTelemetry();
    const engine = new WorkflowEngine(telemetry);

    const submitResult = engine.submit({
      workflowType: "analytics-export",
      correlationId: "corr-transient",
      payload: { accountId: "acct-2" }
    });

    const retry = engine.handleNextMessage(() => ({
      kind: "retryableFailure",
      detail: "dependency timeout"
    }));

    expect(retry?.kind).toBe("retryableFailure");
    expect(engine.getQueue()).toHaveLength(1);
    expect(engine.getRequest(submitResult.requestId)?.status).toBe("queued");
    expect(telemetry.countByName("workflow.worker_retries_total", "corr-transient")).toBe(1);

    const success = engine.handleNextMessage(() => ({
      kind: "success",
      detail: "finalized"
    }));

    expect(success?.kind).toBe("success");
    expect(engine.getRequest(submitResult.requestId)?.status).toBe("completed");
  });
});
// snippet:test-transient-end

// snippet:test-poison-start
describe("poison handling", () => {
  it("moves exhausted failures to dlq and marks failed", () => {
    const telemetry = new InMemoryTelemetry();
    const engine = new WorkflowEngine(telemetry, { defaultMaxAttempts: 1 });

    const submitResult = engine.submit({
      workflowType: "analytics-export",
      correlationId: "corr-poison",
      payload: { accountId: "acct-3" }
    });

    const failure = engine.handleNextMessage(() => ({
      kind: "permanentFailure",
      detail: "schema mismatch"
    }));

    expect(failure?.kind).toBe("permanentFailure");
    expect(engine.getRequest(submitResult.requestId)?.status).toBe("failed");
    expect(engine.getDlq()).toHaveLength(1);
    expect(telemetry.countByName("workflow.failed_total", "corr-poison")).toBe(1);
  });
});
// snippet:test-poison-end

// snippet:test-backlog-start
describe("backlog signal", () => {
  it("emits backlog metric and log once queue is deep and aged", () => {
    const telemetry = new InMemoryTelemetry();
    const engine = new WorkflowEngine(telemetry, { backlogWarningAgeMs: 100, queueBacklogThreshold: 3 });

    for (let i = 0; i < 4; i += 1) {
      engine.submit({
        workflowType: "analytics-export",
        correlationId: `corr-backlog-${i + 1}`,
        idempotencyKey: `idem-${i + 1}`,
        payload: { accountId: `${i + 1}` }
      });
    }

    engine.runBacklogChecks(new Date(Date.now() + 5_000));

    expect(telemetry.countByName("workflow.backlog_age_breach", "corr-backlog-1")).toBe(1);
    expect(telemetry.countByName("workflow.backlog_warning", "corr-backlog-1")).toBe(1);
  });
});
// snippet:test-backlog-end

// snippet:test-idempotency-start
describe("idempotent submit", () => {
  it("reuses the same request for repeated idempotency keys", () => {
    const telemetry = new InMemoryTelemetry();
    const engine = new WorkflowEngine(telemetry);

    const first = engine.submit({
      workflowType: "analytics-export",
      correlationId: "corr-idem-1",
      idempotencyKey: "idem-repeat",
      payload: { accountId: "acct-4" }
    });

    const second = engine.submit({
      workflowType: "analytics-export",
      correlationId: "corr-idem-2",
      idempotencyKey: "idem-repeat",
      payload: { accountId: "acct-4" }
    });

    expect(second.reused).toBe(true);
    expect(second.requestId).toBe(first.requestId);
    expect(engine.getQueue()).toHaveLength(1);
  });
});
// snippet:test-idempotency-end