export type PlayLatencyPhaseDefinition = {
  header: string;
  phase: string;
  source?: string;
  description: string;
  includeInDominantPhase?: boolean;
};

export type PlayLatencyObservation = {
  ts?: number | null;
  source: string;
  runId?: string | null;
  playLabel?: string | null;
  phase: string;
  ms?: number | null;
  ok?: boolean | null;
  status?: string | null;
  graphHash?: string | null;
  extra?: Record<string, unknown>;
};

export type PlayLatencyPhase = PlayLatencyPhaseDefinition & {
  ms: number | null;
};

export type PlayLatencyTrace = {
  runId: string | null;
  playLabel: string | null;
  status: string | null;
  ok: boolean | null;
  phaseMs: Record<string, number>;
  phases: PlayLatencyPhase[];
  dominantPhase: PlayLatencyPhase | null;
  observations: PlayLatencyObservation[];
};

export const PLAY_LATENCY_SUMMARY_COLUMNS: PlayLatencyPhaseDefinition[] = [
  {
    header: 'wall',
    phase: 'runner.wall',
    source: 'runner',
    description:
      'Measured play execution window only: CLI invoke, stream/tail, and assertions. Pre-run setup is excluded.',
    includeInDominantPhase: false,
  },
  {
    header: 'setup_reset',
    phase: 'setup.reset',
    source: 'setup',
    description:
      'Pre-run history/table cleanup. This is setup time, not play execution time.',
    includeInDominantPhase: false,
  },
  {
    header: 'command',
    phase: 'runner.command',
    source: 'runner',
    description: 'Runner-side CLI command wall time.',
  },
  {
    header: 'assert_status',
    phase: 'runner.fetchAssertionStatus',
    source: 'runner',
    description:
      'Runner-side post-run status fetch used by completed-run assertions.',
  },
  {
    header: 'assertions',
    phase: 'runner.completedAssertions',
    source: 'runner',
    description: 'Runner-side completed-run assertion checks.',
  },
  {
    header: 'submit',
    phase: 'server.total',
    source: 'server',
    description: 'POST /api/v2/plays/run end-to-end on the app server.',
  },
  {
    header: 'start_stream',
    phase: 'server.start_stream_start_run',
    source: 'server',
    description:
      'Streamed start route from request handling through startPlayRunForAuth.',
  },
  {
    header: 'fast_terminal',
    phase: 'server.start_stream_fast_terminal',
    source: 'server',
    description:
      'Streamed start route emitting terminal status without scheduler attach.',
  },
  {
    header: 'route_inputs',
    phase: 'server.start_route_prepare_inputs',
    source: 'server',
    description:
      'Start route artifact preparation, preflight, and input file staging.',
  },
  {
    header: 'route_resolve',
    phase: 'server.start_route_resolve_named_play',
    source: 'server',
    description: 'Start route named play revision resolution.',
  },
  {
    header: 'route_child',
    phase: 'server.start_route_resolve_child_manifests',
    source: 'server',
    description: 'Start route child play manifest resolution.',
  },
  {
    header: 'route_active',
    phase: 'server.start_route_active_run_check',
    source: 'server',
    description: 'Start route active-run checks for guarded child play calls.',
  },
  {
    header: 'route_exec',
    phase: 'server.start_route_start_execution',
    source: 'server',
    description:
      'Start route call into startPlayRunExecution, including validation, persistence, and scheduler submit.',
  },
  {
    header: 'route_ready',
    phase: 'server.start_route_runtime_data_plane_ready',
    source: 'server',
    description:
      'Start route read-only Runtime Sheet Data Plane readiness check.',
  },
  {
    header: 'ready_plane',
    phase: 'server.runtime_data_plane_read_plane',
    source: 'server',
    description: 'Runtime Sheet readiness control-plane record lookup.',
  },
  {
    header: 'ready_markers',
    phase: 'server.runtime_data_plane_read_markers',
    source: 'server',
    description:
      'Single pooled tenant-Postgres read of the storage-contract fingerprint and runtime migration version.',
  },
  {
    header: 'route_wait',
    phase: 'server.start_route_wait_completion',
    source: 'server',
    description: 'Start route bounded wait for a terminal result.',
  },
  {
    header: 'route_total',
    phase: 'server.start_route_total',
    source: 'server',
    description:
      'Start route total time from auth/body normalization to outcome.',
  },
  {
    header: 'validate',
    phase: 'server.validate_before_run_total',
    source: 'server',
    description: 'Server-side play validation and artifact preflight.',
  },
  {
    header: 'exec_validate',
    phase: 'server.start_execution_validate',
    source: 'server',
    description: 'startPlayRunExecution validation phase.',
  },
  {
    header: 'callback',
    phase: 'server.start_execution_callback_health',
    source: 'server',
    description: 'Callback URL health check before launching remote runtime.',
  },
  {
    header: 'authority',
    phase: 'server.start_execution_mint_authority',
    source: 'server',
    description: 'Executor token and runtime authority minting.',
  },
  {
    header: 'session_preload',
    phase: 'server.preload_runtime_sheet_sessions',
    source: 'server',
    description: 'Scoped runtime DB session mint/encrypt preload.',
  },
  {
    header: 'preload_wait',
    phase: 'server.start_execution_preload_wait',
    source: 'server',
    description: 'Wait for runtime DB session preload before run persistence.',
  },
  {
    header: 'persist',
    phase: 'server.persist',
    source: 'server',
    description: 'Initial durable run persistence.',
  },
  {
    header: 'sched_submit',
    phase: 'server.scheduler_submit',
    source: 'server',
    description: 'Server to coordinator submit round trip.',
  },
  {
    header: 'stream_attach',
    phase: 'server.stream_scheduler_attach',
    source: 'server',
    description: 'Server attach to scheduler observer after start.',
  },
  {
    header: 'stream_term',
    phase: 'server.stream_scheduler_terminal_event',
    source: 'server',
    description: 'Scheduler observer wait until terminal coordinator event.',
  },
  {
    header: 'term_snapshot',
    phase: 'server.stream_terminal_snapshot_query',
    source: 'server',
    description: 'Canonical terminal snapshot lookup after scheduler terminal.',
  },
  {
    header: 'term_append',
    phase: 'server.stream_terminal_append_transport',
    source: 'server',
    description:
      'Transport terminal event append when Convex has not converged.',
  },
  {
    header: 'coord_submit',
    phase: 'coordinator.submit_total',
    source: 'coordinator',
    description: 'Coordinator submit route total.',
  },
  {
    header: 'dispatch_entry',
    phase: 'coordinator.dispatch_get_entrypoint',
    source: 'coordinator',
    description:
      'Coordinator Durable Object entrypoint lookup before dynamic start.',
  },
  {
    header: 'dynamic_start',
    phase: 'coordinator.dynamic_start_fetch',
    source: 'coordinator',
    description: 'Coordinator fetch to dynamic worker /start.',
  },
  {
    header: 'workflow_create',
    phase: 'coordinator.workflow_create',
    source: 'coordinator',
    description: 'Coordinator call that creates the Cloudflare Workflow run.',
  },
  {
    header: 'start_gap',
    phase: 'coordinator.workflow_start_gap',
    source: 'coordinator',
    description:
      'Delay after coordinator submit begins before Cloudflare enters the Workflow run.',
  },
  {
    header: 'post_create_gap',
    phase: 'coordinator.workflow_post_create_start_gap',
    source: 'coordinator',
    description:
      'Delay after Cloudflare Workflow create returns before Workflow user code starts.',
  },
  {
    header: 'start_drain',
    phase: 'coordinator.dynamic_start_response_drain',
    source: 'coordinator',
    description:
      'Coordinator time spent reading the dynamic worker /start response body.',
  },
  {
    header: 'coord_dispatch',
    phase: 'coordinator.dispatch_workflow',
    source: 'coordinator',
    description: 'Coordinator dispatch to workflow start.',
  },
  {
    header: 'status_read',
    phase: 'coordinator.submit_status_read',
    source: 'coordinator',
    description: 'Coordinator Workflow status read after submit.',
  },
  {
    header: 'loader_compile',
    phase: 'coordinator.loader_compile',
    source: 'coordinator',
    description: 'Worker Loader bundle compile or cache-hit span.',
  },
  {
    header: 'runner_run',
    phase: 'coordinator.runner_run',
    source: 'coordinator',
    description: 'Inner play runner execution on the dynamic Worker.',
  },
  {
    header: 'wall_minus_runner',
    phase: 'derived.wall_minus_coordinator_runner_run',
    source: 'derived',
    description:
      'Runner wall time not explained by inner coordinator.runner_run.',
  },
];

const PHASE_DEFINITION_BY_KEY = new Map(
  PLAY_LATENCY_SUMMARY_COLUMNS.map((definition) => [
    definition.phase,
    definition,
  ]),
);

function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value && typeof value === 'object' && !Array.isArray(value));
}

function finiteNumber(value: unknown): number | null {
  return typeof value === 'number' && Number.isFinite(value) ? value : null;
}

function nullableString(value: unknown): string | null {
  return typeof value === 'string' && value.trim() ? value.trim() : null;
}

function sourceForPhase(phase: string): string {
  const definition = PHASE_DEFINITION_BY_KEY.get(phase);
  if (definition?.source) {
    return definition.source;
  }
  const prefix = phase.split('.')[0]?.trim();
  return prefix || 'unknown';
}

export function formatPlayLatencyMs(value: number | null | undefined): string {
  if (value === null || value === undefined || !Number.isFinite(value)) {
    return 'n/a';
  }
  return `${Math.round(value)}ms`;
}

export function normalizePlayLatencyObservation(
  value: unknown,
): PlayLatencyObservation | null {
  if (!isRecord(value)) {
    return null;
  }
  const phase = nullableString(value.phase);
  if (!phase) {
    return null;
  }
  const source = nullableString(value.source) ?? sourceForPhase(phase);
  return {
    ts: finiteNumber(value.ts),
    source,
    runId: nullableString(value.runId),
    playLabel: nullableString(value.playLabel),
    phase,
    ms: finiteNumber(value.ms),
    ok: typeof value.ok === 'boolean' ? value.ok : null,
    status: nullableString(value.status),
    graphHash: nullableString(value.graphHash),
    extra: Object.fromEntries(
      Object.entries(value).filter(
        ([key]) =>
          ![
            'ts',
            'source',
            'runId',
            'playLabel',
            'phase',
            'ms',
            'ok',
            'status',
            'graphHash',
          ].includes(key),
      ),
    ),
  };
}

export function parseCliTraceTimings(output: string): Record<string, number> {
  const timings: Record<string, number> = {};
  for (const line of output.split(/\r?\n/)) {
    const observation = parseMarkerJsonObservation(line, '[cli-trace] ');
    if (
      observation?.phase.startsWith('cli.') &&
      observation.ms !== null &&
      observation.ms !== undefined
    ) {
      timings[observation.phase] = observation.ms;
    }
  }
  return timings;
}

export function parsePlayPerfTraceLine(
  line: string,
): PlayLatencyObservation | null {
  return parseMarkerJsonObservation(line, '[perf-trace] ');
}

export function parsePlayPerfTraceJsonLine(
  line: string,
): PlayLatencyObservation | null {
  const trimmed = line.trim();
  if (!trimmed) {
    return null;
  }
  try {
    return normalizePlayLatencyObservation(JSON.parse(trimmed));
  } catch {
    return null;
  }
}

export function parsePlayPerfTraceLines(
  lines: Iterable<string>,
): PlayLatencyObservation[] {
  const observations: PlayLatencyObservation[] = [];
  for (const line of lines) {
    const observation = parsePlayPerfTraceLine(line);
    if (observation) {
      observations.push(observation);
    }
  }
  return observations.sort((left, right) => (left.ts ?? 0) - (right.ts ?? 0));
}

export function parsePlayPerfTraceJsonLines(
  lines: Iterable<string>,
): PlayLatencyObservation[] {
  const observations: PlayLatencyObservation[] = [];
  for (const line of lines) {
    const observation = parsePlayPerfTraceJsonLine(line);
    if (observation) {
      observations.push(observation);
    }
  }
  return observations.sort((left, right) => (left.ts ?? 0) - (right.ts ?? 0));
}

export function buildPlayLatencyTrace(input: {
  runId?: string | null;
  playLabel?: string | null;
  status?: string | null;
  ok?: boolean | null;
  wallMs?: number | null;
  observations: readonly PlayLatencyObservation[];
}): PlayLatencyTrace {
  const normalizedRunId = input.runId ?? null;
  const normalizedPlayLabel = input.playLabel ?? null;
  const relevant = input.observations.filter((observation) => {
    const matchesRun =
      normalizedRunId !== null &&
      observation.runId !== null &&
      observation.runId !== undefined &&
      observation.runId === normalizedRunId;
    const matchesLabel =
      normalizedPlayLabel !== null &&
      observation.playLabel !== null &&
      observation.playLabel !== undefined &&
      observation.playLabel === normalizedPlayLabel;
    return (
      matchesRun ||
      matchesLabel ||
      (normalizedRunId === null && normalizedPlayLabel === null)
    );
  });

  const phaseMs = new Map<string, number>();
  const setMax = (phase: string, ms: number | null | undefined) => {
    if (ms === null || ms === undefined || !Number.isFinite(ms)) {
      return;
    }
    phaseMs.set(phase, Math.max(phaseMs.get(phase) ?? 0, ms));
  };

  for (const observation of relevant) {
    setMax(observation.phase, observation.ms);
  }
  setMax('runner.wall', input.wallMs);

  const wallMs = phaseMs.get('runner.wall') ?? null;
  const runnerRunMs = phaseMs.get('coordinator.runner_run') ?? null;
  if (wallMs !== null && runnerRunMs !== null) {
    phaseMs.set(
      'derived.wall_minus_coordinator_runner_run',
      Math.max(0, wallMs - runnerRunMs),
    );
  }
  const workflowStartGapMs =
    phaseMs.get('coordinator.workflow_start_gap') ?? null;
  const workflowCreateMs = phaseMs.get('coordinator.workflow_create') ?? null;
  if (
    !phaseMs.has('coordinator.workflow_post_create_start_gap') &&
    workflowStartGapMs !== null &&
    workflowCreateMs !== null
  ) {
    phaseMs.set(
      'coordinator.workflow_post_create_start_gap',
      Math.max(0, workflowStartGapMs - workflowCreateMs),
    );
  }

  const definedPhaseKeys = new Set(
    PLAY_LATENCY_SUMMARY_COLUMNS.map((definition) => definition.phase),
  );
  const unknownDefinitions: PlayLatencyPhaseDefinition[] = [...phaseMs.keys()]
    .filter((phase) => !definedPhaseKeys.has(phase))
    .sort()
    .map((phase) => ({
      header: phase,
      phase,
      source: sourceForPhase(phase),
      description: `Uncatalogued latency phase ${phase}.`,
    }));

  const phases = [...PLAY_LATENCY_SUMMARY_COLUMNS, ...unknownDefinitions].map(
    (definition): PlayLatencyPhase => ({
      ...definition,
      ms: phaseMs.get(definition.phase) ?? null,
    }),
  );
  const dominantPhase =
    phases
      .filter(
        (phase) =>
          phase.ms !== null &&
          phase.ms !== undefined &&
          (phase.includeInDominantPhase ?? true),
      )
      .sort((left, right) => (right.ms ?? 0) - (left.ms ?? 0))[0] ?? null;

  return {
    runId: normalizedRunId,
    playLabel: normalizedPlayLabel,
    status: input.status ?? null,
    ok: input.ok ?? null,
    phaseMs: Object.fromEntries(phaseMs.entries()),
    phases,
    dominantPhase,
    observations: relevant,
  };
}

export function formatPlayLatencyBreakdownLines(
  trace: PlayLatencyTrace,
  options: { prefix?: string; includeHeader?: boolean } = {},
): string[] {
  const prefix = options.prefix ?? '';
  const lines: string[] = [];
  if (options.includeHeader ?? true) {
    lines.push(
      `${prefix}latency breakdown ${trace.playLabel ?? trace.runId ?? 'unknown'}`,
    );
    lines.push(`${prefix}  run: ${trace.runId ?? 'n/a'}`);
  }
  for (const phase of trace.phases) {
    if (phase.ms === null || phase.ms === undefined) {
      continue;
    }
    lines.push(`${prefix}  ${phase.phase}: ${formatPlayLatencyMs(phase.ms)}`);
  }
  if (
    trace.dominantPhase?.ms !== null &&
    trace.dominantPhase?.ms !== undefined
  ) {
    lines.push(
      `${prefix}  dominant: ${trace.dominantPhase.phase} ${formatPlayLatencyMs(
        trace.dominantPhase.ms,
      )}`,
    );
  }
  return lines;
}

export function phaseLegendLines(): string[] {
  return PLAY_LATENCY_SUMMARY_COLUMNS.map(
    (phase) => `- **${phase.header}** - ${phase.description}`,
  );
}

function parseMarkerJsonObservation(
  line: string,
  marker: string,
): PlayLatencyObservation | null {
  const markerIndex = line.indexOf(marker);
  if (markerIndex < 0) {
    return null;
  }
  const jsonText = extractFirstJsonObject(line, markerIndex + marker.length);
  if (!jsonText) {
    return null;
  }
  try {
    return normalizePlayLatencyObservation(JSON.parse(jsonText));
  } catch {
    return null;
  }
}

function extractFirstJsonObject(
  line: string,
  startIndex: number,
): string | null {
  const start = line.indexOf('{', startIndex);
  if (start < 0) {
    return null;
  }
  let depth = 0;
  let inString = false;
  let escaped = false;
  for (let index = start; index < line.length; index += 1) {
    const char = line[index]!;
    if (escaped) {
      escaped = false;
      continue;
    }
    if (char === '\\') {
      escaped = true;
      continue;
    }
    if (char === '"') {
      inString = !inString;
      continue;
    }
    if (inString) {
      continue;
    }
    if (char === '{') {
      depth += 1;
      continue;
    }
    if (char === '}') {
      depth -= 1;
      if (depth === 0) {
        return line.slice(start, index + 1);
      }
    }
  }
  return null;
}