export type DatasetColumnExecutionBucket =
  | 'queued'
  | 'running'
  | 'completed:executed'
  | 'completed:reused'
  | 'skipped:condition'
  | 'skipped:missed'
  | 'failed';

export type DatasetColumnExecutionStats = Record<
  DatasetColumnExecutionBucket,
  string
>;

export type DatasetColumnSummary = {
  non_empty?: string;
  unique?: number;
  execution?: DatasetColumnExecutionStats;
  sample_value?: unknown;
  sample_type?: string;
  top_values?: Record<string, string>;
};

/**
 * Persisted-row accounting for a dataset, sourced from the sheet summary's
 * row-level stats (the single source of truth). `persisted` is every durable
 * row (the count `runs export` returns); `succeeded`/`failed` partition it.
 * `runs get --full` renders these so its row count matches what export emits.
 */
export type DatasetRowCounts = {
  persisted: number;
  succeeded: number;
  failed: number;
};

export type DatasetSummary = {
  total_rows: number;
  /** Persisted/succeeded/failed breakdown when the sheet summary is available. */
  rowCounts?: DatasetRowCounts;
  columnStats: Record<string, DatasetColumnSummary>;
};

/**
 * `rows: 12 persisted (9 succeeded, 3 failed)` — the canonical row-count line.
 * Falls back to `rows: N persisted` when nothing failed/the breakdown is the
 * full total, so the parenthetical only appears when it adds information.
 */
export function formatDatasetRowCountsLine(counts: DatasetRowCounts): string {
  const { persisted, succeeded, failed } = counts;
  if (succeeded === persisted && failed === 0) {
    return `${persisted} persisted`;
  }
  return `${persisted} persisted (${succeeded} succeeded, ${failed} failed)`;
}

export type DatasetExecutionStatsInput = {
  tableNamespace?: string;
  columnStats: Record<
    string,
    {
      queued?: number;
      running?: number;
      completed?: number;
      cached?: number;
      skipped?: number;
      missed?: number;
      failed?: number;
    }
  >;
};

export function datasetSummaryPercentText(
  numerator: number,
  denominator: number,
): string {
  return denominator > 0
    ? `${numerator}/${denominator} (${Math.round((100 * numerator) / denominator)}%)`
    : '0/0 (0%)';
}

function readCount(value: unknown): number {
  return typeof value === 'number' && Number.isFinite(value) && value >= 0
    ? Math.trunc(value)
    : 0;
}

/**
 * Sum of every per-cell execution bucket for a column. Each bucket is a
 * terminal (or in-flight) state count for one cell of one row, so the sum is
 * the number of cell attempts the column actually has — never less than any
 * single bucket. We use this as the percentage denominator so no displayed
 * execution stat can exceed 100%.
 *
 * Why not the persisted-row total: the persisted total counts distinct rows,
 * but the column buckets can be sourced from a maintained transition-delta
 * table that double-counts retried cells (e.g. a cell that fails then succeeds
 * lands in both `failed` and `completed`). Dividing those numerators by the
 * persisted-row total produced impossible figures like `executed=12/9 (133%)`.
 * Summing the buckets is the semantically honest denominator: each percent is
 * the share of this column's attempts that ended in that bucket.
 */
function executionAttemptTotal(
  raw: DatasetExecutionStatsInput['columnStats'][string],
): number {
  return (
    readCount(raw.queued) +
    readCount(raw.running) +
    readCount(raw.completed) +
    readCount(raw.cached) +
    readCount(raw.skipped) +
    readCount(raw.missed) +
    readCount(raw.failed)
  );
}

export function formatDatasetExecutionStats(
  raw: DatasetExecutionStatsInput['columnStats'][string],
  // `persistedRowTotal` is kept for callers that pass the run's persisted row
  // count, but it is intentionally NOT the denominator (see
  // `executionAttemptTotal`). The denominator is the per-column attempt total,
  // which guarantees every bucket is <=100%.
  _persistedRowTotal: number,
): DatasetColumnExecutionStats {
  const denominator = executionAttemptTotal(raw);
  const stats: DatasetColumnExecutionStats = {
    queued: datasetSummaryPercentText(readCount(raw.queued), denominator),
    running: datasetSummaryPercentText(readCount(raw.running), denominator),
    'completed:executed': datasetSummaryPercentText(
      readCount(raw.completed),
      denominator,
    ),
    'completed:reused': datasetSummaryPercentText(
      readCount(raw.cached),
      denominator,
    ),
    'skipped:condition': datasetSummaryPercentText(
      readCount(raw.skipped),
      denominator,
    ),
    'skipped:missed': datasetSummaryPercentText(
      readCount(raw.missed),
      denominator,
    ),
    failed: datasetSummaryPercentText(readCount(raw.failed), denominator),
  };
  // Invariant guard: with an attempt-sum denominator no single bucket can
  // exceed the denominator. If it ever does, the upstream summary is corrupt
  // (a bucket count larger than the sum of all buckets is impossible) — flag it
  // loudly as a data bug rather than silently capping.
  if (
    Object.values(stats).some((text) => {
      const match = /\((\d+)%\)/.exec(text);
      return match ? Number(match[1]) > 100 : false;
    })
  ) {
    throw new Error(
      `formatDatasetExecutionStats produced a >100% execution stat; column counts are corrupt: ${JSON.stringify(
        raw,
      )}`,
    );
  }
  return stats;
}