/**
 * Pipeline flow — turns raw `source.stage` / `sink.stage` telemetry records
 * into the inbound and outbound stage rows for one correlation, so the Trace
 * view can draw the full flow: inbound pipeline → execute/fold (the event
 * tree) → outbound pipeline.
 *
 * Pure + framework-free so it unit-tests without a DOM. The server's inspect
 * endpoint streams every telemetry kind; this just selects the two stage
 * kinds, scopes them to a correlation, and orders them the way they ran
 * (position early → middle → terminal, then by timestamp within a position).
 */

/** A `source.stage` telemetry record (inbound pipeline). */
export interface SourceStageRecord {
  kind: "source.stage";
  stage: string;
  position: StagePosition;
  stageKind?: string;
  message?: { name?: string; kind?: "command" | "event" };
  correlationId?: string;
  tenant?: string;
  shortCircuited: boolean;
  durationMs: number;
  appName: string;
  ts: string;
}

/** A `sink.stage` telemetry record (outbound pipeline). */
export interface SinkStageRecord {
  kind: "sink.stage";
  stage: string;
  position: StagePosition;
  stageKind?: string;
  event: string;
  correlationId?: string;
  tenant?: string;
  shortCircuited: boolean;
  durationMs: number;
  appName: string;
  ts: string;
}

export type StagePosition = "early" | "middle" | "terminal";
export type StageRecord = SourceStageRecord | SinkStageRecord;

/** One row in the rendered pipeline. */
export interface StageRow {
  direction: "inbound" | "outbound";
  stage: string;
  position: StagePosition;
  stageKind?: string;
  /** Inbound: the message name; outbound: the event name. */
  label: string;
  shortCircuited: boolean;
  durationMs: number;
  appName: string;
  ts: string;
}

export interface PipelineFlow {
  inbound: StageRow[];
  outbound: StageRow[];
}

const POSITION_ORDER: Record<StagePosition, number> = { early: 0, middle: 1, terminal: 2 };

/** True for the two stage-telemetry kinds this view cares about. */
export function isStageRecord(rec: { kind?: string }): rec is StageRecord {
  return rec.kind === "source.stage" || rec.kind === "sink.stage";
}

function order(a: StageRow, b: StageRow): number {
  const byPos = POSITION_ORDER[a.position] - POSITION_ORDER[b.position];
  return byPos !== 0 ? byPos : a.ts.localeCompare(b.ts);
}

/**
 * Build the inbound + outbound stage rows for one correlation. Records with
 * no correlationId are excluded (they can't be tied to a trace). When
 * `correlationId` is undefined, returns empty flows.
 */
export function pipelineFlow(
  records: readonly StageRecord[],
  correlationId: string | undefined,
): PipelineFlow {
  if (!correlationId) return { inbound: [], outbound: [] };
  const inbound: StageRow[] = [];
  const outbound: StageRow[] = [];
  for (const rec of records) {
    if (rec.correlationId !== correlationId) continue;
    if (rec.kind === "source.stage") {
      inbound.push({
        direction: "inbound",
        stage: rec.stage,
        position: rec.position,
        stageKind: rec.stageKind,
        label: rec.message?.name ?? "",
        shortCircuited: rec.shortCircuited,
        durationMs: rec.durationMs,
        appName: rec.appName,
        ts: rec.ts,
      });
    } else {
      outbound.push({
        direction: "outbound",
        stage: rec.stage,
        position: rec.position,
        stageKind: rec.stageKind,
        label: rec.event,
        shortCircuited: rec.shortCircuited,
        durationMs: rec.durationMs,
        appName: rec.appName,
        ts: rec.ts,
      });
    }
  }
  inbound.sort(order);
  outbound.sort(order);
  return { inbound, outbound };
}