import type { PlayExecutionFileRef } from './file-refs';

const PLAY_DATASET_BRAND = Symbol.for('deepline.play.dataset');
const NODE_INSPECT_CUSTOM = Symbol.for('nodejs.util.inspect.custom');
const residentRowsByDataset = new WeakMap<object, readonly unknown[]>();
const DEFAULT_MATERIALIZE_LIMIT = 10_000;
export const PLAY_DATASET_EXECUTION_PAGE_ROWS = 1_000;
export const PLAY_DATASET_EXECUTION_PAGE_BYTES = 64 * 1024 * 1024;

export type PlayDatasetKind = 'csv' | 'map';

export type PlayDatasetBacking =
  | {
      storage: 'neon_sheet';
      sheet: {
        playName: string;
        tableNamespace: string;
      };
    }
  | {
      storage: 'r2_file';
      file: PlayExecutionFileRef;
    };

export type PlayDatasetWorkProgressSummary = {
  total: number;
  executed: number;
  reused: number;
  skipped: number;
  pending: number;
  failed: number;
  degraded?: boolean;
  duplicates?: {
    exact?: number;
    semantic?: number;
    rejected?: number;
  };
};

export interface SerializedPlayDataset<T> {
  kind: 'dataset';
  datasetKind: PlayDatasetKind;
  datasetId: string;
  count: number;
  backing?: PlayDatasetBacking;
  sourceLabel?: string | null;
  tableNamespace?: string | null;
  columns?: string[];
  _metadata?: {
    workProgress?: PlayDatasetWorkProgressSummary;
  };
  preview: T[];
}

/**
 * Durable cell representation for a PlayDataset returned from a map column.
 *
 * `PlayDataset.toJSON()` intentionally exposes only a bounded preview for
 * normal output rendering. A runtime-sheet cell needs the complete rows so it
 * can revive a live handle after a persistence or resume boundary.
 */
export interface SerializedPlayDatasetCell<T> {
  __kind: 'deepline.play_dataset_cell.v1';
  dataset: SerializedPlayDataset<T>;
  rows: T[];
}

const PLAY_DATASET_CELL_KIND = 'deepline.play_dataset_cell.v1';
const PLAY_DATASET_CELL_MAX_BYTES = 5 * 1024 * 1024;
const DATASET_CELL_REHYDRATION_INCOMPLETE =
  'DATASET_CELL_REHYDRATION_INCOMPLETE';
const DATASET_CELL_CORRUPT = 'DATASET_CELL_CORRUPT';

export type PlayDatasetInput<T> =
  | ReadonlyArray<T>
  | Iterable<T>
  | AsyncIterable<T>
  | PlayDataset<T>;

export type PlayDatasetRow<TInput> =
  TInput extends PlayDataset<infer Row>
    ? Row
    : TInput extends ReadonlyArray<infer Row>
      ? Row
      : TInput extends Iterable<infer Row>
        ? Row
        : TInput extends AsyncIterable<infer Row>
          ? Row
          : never;

export type PlayDatasetTransformOptions = {
  key?: string;
  sourceLabel?: string | null;
};

/**
 * Durable handle for rows produced by `ctx.csv(...)` or `ctx.dataset(...).run()`.
 *
 * A `PlayDataset` is not a normal in-memory array. It points at runtime-managed
 * rows, usually backed by persisted sheet storage, and carries metadata such as
 * dataset kind, dataset id, table namespace, count, and preview rows.
 *
 * Pass dataset handles directly into later `ctx.dataset(...)` stages by default so
 * Deepline keeps row progress, retries, memory use, and table output under
 * runtime control. Use `count()` and `peek()` for bounded inspection. Use
 * `materialize(limit)` or async iteration only when the dataset is intentionally
 * small and bounded. `PlayDataset` intentionally does not expose `.rows`,
 * `.toArray()`, `.length`, numeric indexing, spread, or synchronous iteration;
 * those hide the runtime cost of loading persisted rows into memory or make
 * behavior depend on whether rows happen to be resident.
 *
 * @sdkReference runtime 190
 */
export interface PlayDataset<T> extends AsyncIterable<T> {
  readonly [PLAY_DATASET_BRAND]: true;
  /** Dataset kind. */
  readonly datasetKind: PlayDatasetKind;
  /** Dataset id. */
  readonly datasetId: string;
  /** Backing store info. */
  readonly backing?: PlayDatasetBacking;
  /** Display label. */
  readonly sourceLabel?: string | null;
  /** Runtime table name. */
  readonly tableNamespace?: string | null;
  /** Row count. */
  count(): Promise<number>;
  /** Preview rows. */
  peek(limit?: number): Promise<T[]>;
  /** First row, loading it asynchronously when necessary. */
  first(): Promise<T | undefined>;
  /** Row at an array-style index, loading it asynchronously when necessary. */
  at(index: number): Promise<T | undefined>;
  map<U>(
    mapper: (row: T, index: number) => U | Promise<U>,
    options?: PlayDatasetTransformOptions,
  ): PlayDataset<U>;
  filter(
    predicate: (row: T, index: number) => boolean | Promise<boolean>,
    options?: PlayDatasetTransformOptions,
  ): PlayDataset<T>;
  slice(
    start?: number,
    end?: number,
    options?: PlayDatasetTransformOptions,
  ): PlayDataset<T>;
  take(limit: number, options?: PlayDatasetTransformOptions): PlayDataset<T>;
  /**
   * Explicit escape hatch for bounded result sets.
   * Large datasets should flow by handle through Neon-backed storage, not
   * through worker memory as giant arrays.
   */
  materialize(options?: number | PlayDatasetMaterializeOptions): Promise<T[]>;
  toJSON(): {
    kind: 'dataset';
    datasetKind: PlayDatasetKind;
    datasetId: string;
    count: number;
    backing?: PlayDatasetBacking;
    sourceLabel?: string | null;
    tableNamespace?: string | null;
    columns?: string[];
    _metadata?: {
      workProgress?: PlayDatasetWorkProgressSummary;
    };
    preview: T[];
  };
}

type PlayDatasetResolvers<T> = {
  count: () => Promise<number>;
  peek: (limit: number) => Promise<T[]>;
  at?: (index: number) => Promise<T | undefined>;
  materialize: (limit?: number) => Promise<T[]>;
  materializeFullPersistedDataset?: (limit?: number) => Promise<T[]>;
  iterate: () => AsyncIterable<T>;
};

export type PlayDatasetMaterializeScope = 'result' | 'full_persisted_dataset';

export type PlayDatasetMaterializeOptions = {
  /** Rows returned by this operation, or every current row in its persisted dataset. */
  scope?: PlayDatasetMaterializeScope;
  /** Maximum number of rows to load into memory. */
  limit?: number;
};

type PlayDatasetTransform<T, U = T> =
  | {
      kind: 'map';
      mapper: (row: T, index: number) => U | Promise<U>;
    }
  | {
      kind: 'filter';
      predicate: (row: T, index: number) => boolean | Promise<boolean>;
    }
  | {
      kind: 'slice';
      start?: number;
      end?: number;
    };

export function resolveMaterializeLimitCap(): number {
  const raw = process.env.DEEPLINE_PLAY_DATASET_MATERIALIZE_LIMIT;
  const parsed = raw ? Number(raw) : NaN;
  if (Number.isFinite(parsed) && parsed > 0) {
    return Math.floor(parsed);
  }
  return DEFAULT_MATERIALIZE_LIMIT;
}

function inferPreviewColumns<T>(rows: readonly T[]): string[] | undefined {
  const columns = new Set<string>();
  for (const row of rows) {
    if (!row || typeof row !== 'object' || Array.isArray(row)) {
      continue;
    }
    for (const key of Object.keys(row as Record<string, unknown>)) {
      columns.add(key);
    }
  }
  return columns.size > 0 ? [...columns] : undefined;
}

export function isPlayDataset<T>(value: unknown): value is PlayDataset<T> {
  return Boolean(
    value &&
    typeof value === 'object' &&
    (value as Record<PropertyKey, unknown>)[PLAY_DATASET_BRAND] === true,
  );
}

/** Internal compatibility view for preserving proven-complete rows on replay. */
export function residentPlayDatasetRows<T>(
  dataset: PlayDataset<T>,
): readonly T[] | null {
  return (residentRowsByDataset.get(dataset as object) as readonly T[]) ?? null;
}

export function isSerializedPlayDataset<T>(
  value: unknown,
): value is SerializedPlayDataset<T> {
  return Boolean(
    value &&
    typeof value === 'object' &&
    !Array.isArray(value) &&
    (value as Record<string, unknown>).kind === 'dataset' &&
    typeof (value as Record<string, unknown>).datasetKind === 'string' &&
    typeof (value as Record<string, unknown>).datasetId === 'string' &&
    typeof (value as Record<string, unknown>).count === 'number' &&
    Array.isArray((value as Record<string, unknown>).preview),
  );
}

export function isSerializedPlayDatasetCell<T>(
  value: unknown,
): value is SerializedPlayDatasetCell<T> {
  if (!value || typeof value !== 'object' || Array.isArray(value)) {
    return false;
  }
  const record = value as Record<string, unknown>;
  return (
    record.__kind === PLAY_DATASET_CELL_KIND &&
    isSerializedPlayDataset<T>(record.dataset) &&
    Array.isArray(record.rows)
  );
}

/** Serialize every row because a Dataset cell must survive a new worker. */
export async function serializePlayDatasetCell<T>(
  dataset: PlayDataset<T>,
  serializeRow: (row: T) => Promise<T> = async (row) => row,
): Promise<SerializedPlayDatasetCell<T>> {
  const source = dataset.toJSON();
  const datasetMetadata = {
    ...source,
    count: 0,
    preview: [] as T[],
  };
  // The persisted envelope does not repeat the preview rows. That keeps the
  // byte budget honest; deserialize derives its preview from `rows`.
  let encodedBytes = new TextEncoder().encode(
    JSON.stringify({
      __kind: PLAY_DATASET_CELL_KIND,
      dataset: datasetMetadata,
      rows: [],
    }),
  ).length;
  const rows: T[] = [];
  for await (const row of dataset) {
    const serializedRow = await serializeRow(row);
    // Array serialization gives the exact JSON representation even for an
    // undefined row, which JSON encodes as null inside `rows`.
    const rowJson = JSON.stringify([serializedRow]).slice(1, -1);
    encodedBytes +=
      new TextEncoder().encode(rowJson).length + (rows.length > 0 ? 1 : 0);
    if (encodedBytes > PLAY_DATASET_CELL_MAX_BYTES) {
      throw new Error(
        `OUTPUT_TOO_LARGE: Dataset cell ${dataset.datasetId} exceeds the 5 MiB customer-output limit. ` +
          'Keep large datasets as a top-level pipeline or return a bounded dataset from the column.',
      );
    }
    rows.push(serializedRow);
  }
  return {
    __kind: PLAY_DATASET_CELL_KIND,
    dataset: {
      ...source,
      count: rows.length,
      preview: [],
    },
    rows,
  };
}

/** Restore the public lazy-handle contract from a persisted Dataset cell. */
export function deserializePlayDatasetCell<T>(
  value: SerializedPlayDatasetCell<T>,
  deserializeRow: (row: T) => T = (row) => row,
): PlayDataset<T> {
  const { dataset } = value;
  const rows = value.rows.map(deserializeRow);
  if (dataset.count !== rows.length) {
    throw new Error(
      `${DATASET_CELL_CORRUPT}: Dataset cell ${dataset.datasetId} declares ${dataset.count} rows but stores ${rows.length}.`,
    );
  }
  return createDeferredPlayDataset({
    datasetKind: dataset.datasetKind,
    datasetId: dataset.datasetId,
    count: rows.length,
    backing: dataset.backing,
    previewRows: rows.slice(0, 10),
    residentRows: rows,
    sourceLabel: dataset.sourceLabel ?? null,
    tableNamespace: dataset.tableNamespace ?? null,
    workProgress: dataset._metadata?.workProgress,
    resolvers: {
      count: async () => rows.length,
      peek: async (limit) => rows.slice(0, Math.max(0, limit)),
      materialize: async (limit) =>
        limit === undefined ? [...rows] : rows.slice(0, Math.max(0, limit)),
      iterate: () =>
        ({
          async *[Symbol.asyncIterator]() {
            yield* rows;
          },
        }) as AsyncIterable<T>,
    },
  });
}

/**
 * Compatibility reader for cells persisted before Dataset cells carried their
 * full rows. Small legacy lists remain usable; incomplete previews fail with a
 * clear migration error instead of masquerading as a live handle.
 */
export function deserializeLegacyPlayDataset<T>(
  dataset: SerializedPlayDataset<T>,
): PlayDataset<T> {
  const complete = dataset.count === dataset.preview.length;
  const incomplete = (): Error =>
    new Error(
      `${DATASET_CELL_REHYDRATION_INCOMPLETE}: Dataset cell ${dataset.datasetId} only stored ${dataset.preview.length} preview row(s) for ${dataset.count} total row(s). Re-run the producing column to persist the complete Dataset Handle.`,
    );
  const rows = dataset.preview;
  return createDeferredPlayDataset({
    datasetKind: dataset.datasetKind,
    datasetId: dataset.datasetId,
    count: dataset.count,
    backing: dataset.backing,
    previewRows: rows,
    residentRows: complete ? rows : null,
    sourceLabel: dataset.sourceLabel ?? null,
    tableNamespace: dataset.tableNamespace ?? null,
    workProgress: dataset._metadata?.workProgress,
    resolvers: {
      count: async () => dataset.count,
      peek: async (limit) => {
        if (limit > rows.length && !complete) throw incomplete();
        return rows.slice(0, Math.max(0, limit));
      },
      materialize: async (limit) => {
        if (!complete && (limit === undefined || limit > rows.length)) {
          throw incomplete();
        }
        return limit === undefined
          ? [...rows]
          : rows.slice(0, Math.max(0, limit));
      },
      iterate: () =>
        ({
          async *[Symbol.asyncIterator]() {
            if (!complete) throw incomplete();
            yield* rows;
          },
        }) as AsyncIterable<T>,
    },
  });
}

export function trimSerializedPlayDatasetPreview<T>(
  dataset: SerializedPlayDataset<T>,
  limit: number,
): SerializedPlayDataset<T> {
  return {
    ...dataset,
    preview: dataset.preview.slice(0, Math.max(0, limit)),
  };
}

class DeferredPlayDataset<T> implements PlayDataset<T> {
  readonly [PLAY_DATASET_BRAND] = true as const;
  readonly datasetKind: PlayDatasetKind;
  readonly datasetId: string;
  readonly backing?: PlayDatasetBacking;
  readonly sourceLabel?: string | null;
  readonly tableNamespace?: string | null;
  private readonly previewRows: readonly T[];
  private readonly previewColumns?: string[];
  private readonly workProgress?: PlayDatasetWorkProgressSummary;
  private cachedCount: number;
  private knownCount: number | null;
  private readonly residentRows: readonly T[] | null;
  private readonly resolvers: PlayDatasetResolvers<T>;

  constructor(input: {
    datasetKind: PlayDatasetKind;
    datasetId: string;
    count: number;
    knownCount?: number | null;
    backing?: PlayDatasetBacking;
    previewRows: readonly T[];
    residentRows?: readonly T[] | null;
    sourceLabel?: string | null;
    tableNamespace?: string | null;
    workProgress?: PlayDatasetWorkProgressSummary;
    resolvers: PlayDatasetResolvers<T>;
  }) {
    this.datasetKind = input.datasetKind;
    this.datasetId = input.datasetId;
    this.cachedCount = input.count;
    this.knownCount =
      input.knownCount === undefined ? input.count : input.knownCount;
    this.backing = input.backing;
    this.previewRows = input.previewRows;
    this.residentRows = input.residentRows ?? null;
    this.previewColumns = inferPreviewColumns(this.previewRows);
    this.sourceLabel = input.sourceLabel ?? null;
    this.tableNamespace = input.tableNamespace ?? null;
    this.workProgress = input.workProgress;
    this.resolvers = input.resolvers;
  }

  async count(): Promise<number> {
    this.cachedCount = await this.resolvers.count();
    this.knownCount = this.cachedCount;
    return this.cachedCount;
  }

  async peek(limit = 10): Promise<T[]> {
    if (limit <= this.previewRows.length) {
      return this.previewRows.slice(0, Math.max(0, limit));
    }
    return await this.resolvers.peek(limit);
  }

  async first(): Promise<T | undefined> {
    return await this.at(0);
  }

  async at(index: number): Promise<T | undefined> {
    if (!Number.isFinite(index)) return undefined;
    const integer = Math.trunc(index);
    const count = integer < 0 ? await this.count() : this.knownCount;
    const normalized =
      integer < 0 && count !== null ? count + integer : integer;
    if (normalized < 0 || (count !== null && normalized >= count)) {
      return undefined;
    }
    if (this.residentRows) return this.residentRows[normalized];
    if (normalized < this.previewRows.length) {
      return this.previewRows[normalized];
    }
    if (this.resolvers.at) return await this.resolvers.at(normalized);
    let current = 0;
    for await (const row of this.resolvers.iterate()) {
      if (current === normalized) return row;
      current += 1;
    }
    return undefined;
  }

  map<U>(
    mapper: (row: T, index: number) => U | Promise<U>,
    options?: PlayDatasetTransformOptions,
  ): PlayDataset<U> {
    return createTransformedPlayDataset(this, { kind: 'map', mapper }, options);
  }

  filter(
    predicate: (row: T, index: number) => boolean | Promise<boolean>,
    options?: PlayDatasetTransformOptions,
  ): PlayDataset<T> {
    return createTransformedPlayDataset(
      this,
      { kind: 'filter', predicate },
      options,
    );
  }

  slice(
    start?: number,
    end?: number,
    options?: PlayDatasetTransformOptions,
  ): PlayDataset<T> {
    return createTransformedPlayDataset(
      this,
      { kind: 'slice', start, end },
      options,
    );
  }

  take(limit: number, options?: PlayDatasetTransformOptions): PlayDataset<T> {
    return this.slice(0, limit, options);
  }

  async materialize(
    options?: number | PlayDatasetMaterializeOptions,
  ): Promise<T[]> {
    const scope =
      typeof options === 'object' ? (options.scope ?? 'result') : 'result';
    const limit = typeof options === 'number' ? options : options?.limit;
    const requestedLimit =
      limit !== undefined ? Math.max(0, Math.floor(limit)) : undefined;
    const cap = resolveMaterializeLimitCap();
    const materialize =
      scope === 'full_persisted_dataset'
        ? this.resolvers.materializeFullPersistedDataset
        : this.resolvers.materialize;
    if (!materialize) {
      throw new Error(
        'PlayDataset.materialize({ scope: "full_persisted_dataset" }) is only available ' +
          'for a persisted dataset returned by ctx.dataset(...).run().',
      );
    }
    if (requestedLimit !== undefined) {
      if (requestedLimit > cap) {
        throw new Error(
          `PlayDataset.materialize(${requestedLimit}) exceeds the hard limit of ${cap} rows. ` +
            'Return the dataset handle instead, or request a smaller bounded slice.',
        );
      }
      return await materialize(requestedLimit);
    }

    if (scope === 'full_persisted_dataset') {
      const rows = await materialize(cap + 1);
      if (rows.length > cap) {
        throw new Error(
          'PlayDataset.materialize({ scope: "full_persisted_dataset" }) refuses to load ' +
            `more than ${cap} rows into memory. Pass an explicit bounded limit.`,
        );
      }
      return rows;
    }

    const count = await this.count();
    if (count > cap) {
      throw new Error(
        `PlayDataset.materialize() refuses to load ${count} rows into memory. ` +
          `The hard limit is ${cap}. Return the dataset handle instead or call materialize(limit).`,
      );
    }
    return await materialize();
  }

  async *[Symbol.asyncIterator](): AsyncIterator<T> {
    for await (const row of this.resolvers.iterate()) {
      yield row;
    }
  }

  toJSON() {
    return {
      kind: 'dataset' as const,
      datasetKind: this.datasetKind,
      datasetId: this.datasetId,
      count: this.cachedCount,
      ...(this.backing ? { backing: this.backing } : {}),
      ...(this.sourceLabel ? { sourceLabel: this.sourceLabel } : {}),
      ...(this.tableNamespace ? { tableNamespace: this.tableNamespace } : {}),
      ...(this.previewColumns ? { columns: this.previewColumns } : {}),
      ...(this.workProgress
        ? { _metadata: { workProgress: this.workProgress } }
        : {}),
      preview: [...this.previewRows],
    };
  }

  [NODE_INSPECT_CUSTOM]() {
    return this.toJSON();
  }
}

function normalizeSliceBounds(input: {
  start?: number;
  end?: number;
  count: number;
}): { start: number; end: number } {
  const count = Math.max(0, Math.floor(input.count));
  const rawStart = input.start ?? 0;
  const rawEnd = input.end ?? count;
  const startInteger = Number.isFinite(rawStart) ? Math.trunc(rawStart) : 0;
  const endInteger = Number.isFinite(rawEnd) ? Math.trunc(rawEnd) : count;
  const start =
    startInteger < 0
      ? Math.max(count + startInteger, 0)
      : Math.min(startInteger, count);
  const end =
    endInteger < 0
      ? Math.max(count + endInteger, 0)
      : Math.min(endInteger, count);
  return { start, end: Math.max(start, end) };
}

function transformDatasetId(input: {
  source: PlayDataset<unknown>;
  kind: string;
  key?: string;
}): string {
  const key = input.key?.trim();
  return key
    ? `${input.source.datasetId}:${input.kind}:${key}`
    : `${input.source.datasetId}:${input.kind}`;
}

function createTransformedPlayDataset<T, U>(
  source: PlayDataset<T>,
  transform: PlayDatasetTransform<T, U>,
  options?: PlayDatasetTransformOptions,
): PlayDataset<U> {
  const sourceLabel =
    options?.sourceLabel ??
    `${source.sourceLabel ?? source.tableNamespace ?? source.datasetId}.${transform.kind}`;

  const iterate = async function* (): AsyncIterable<U> {
    if (transform.kind === 'slice') {
      const bounds = normalizeSliceBounds({
        start: transform.start,
        end: transform.end,
        count: await source.count(),
      });
      let index = 0;
      for await (const row of source) {
        if (index >= bounds.end) break;
        if (index >= bounds.start) {
          yield row as unknown as U;
        }
        index += 1;
      }
      return;
    }

    let inputIndex = 0;
    let outputIndex = 0;
    for await (const row of source) {
      if (transform.kind === 'filter') {
        if (await transform.predicate(row, inputIndex)) {
          yield row as unknown as U;
          outputIndex += 1;
        }
      } else {
        yield await transform.mapper(row, outputIndex);
        outputIndex += 1;
      }
      inputIndex += 1;
    }
  };

  const collect = async (limit?: number): Promise<U[]> => {
    const rows: U[] = [];
    const boundedLimit =
      limit === undefined ? undefined : Math.max(0, Math.floor(limit));
    if (boundedLimit === 0) return rows;
    for await (const row of iterate()) {
      rows.push(row);
      if (boundedLimit !== undefined && rows.length >= boundedLimit) break;
    }
    return rows;
  };

  const count = async (): Promise<number> => {
    if (transform.kind === 'map') return await source.count();
    if (transform.kind === 'slice') {
      const bounds = normalizeSliceBounds({
        start: transform.start,
        end: transform.end,
        count: await source.count(),
      });
      return Math.max(0, bounds.end - bounds.start);
    }
    let total = 0;
    for await (const _row of iterate()) {
      void _row;
      total += 1;
    }
    return total;
  };

  return createDeferredPlayDataset({
    datasetKind: source.datasetKind,
    datasetId: transformDatasetId({
      source: source as PlayDataset<unknown>,
      kind: transform.kind,
      key: options?.key,
    }),
    count: 0,
    knownCount: null,
    backing: source.backing,
    sourceLabel,
    tableNamespace: options?.key ?? null,
    resolvers: {
      count,
      peek: async (limit) => collect(limit),
      materialize: async (limit) => collect(limit),
      iterate: () =>
        ({
          async *[Symbol.asyncIterator]() {
            yield* iterate();
          },
        }) as AsyncIterable<U>,
    },
  });
}

export function createDeferredPlayDataset<T>(input: {
  datasetKind: PlayDatasetKind;
  datasetId: string;
  count: number;
  /** Authoritative synchronous count, or null when execution must determine it. */
  knownCount?: number | null;
  backing?: PlayDatasetBacking;
  previewRows?: readonly T[];
  /** Complete in-memory row set. Never pass a bounded preview here. */
  residentRows?: readonly T[] | null;
  sourceLabel?: string | null;
  tableNamespace?: string | null;
  workProgress?: PlayDatasetWorkProgressSummary;
  resolvers: PlayDatasetResolvers<T>;
}): PlayDataset<T> {
  if (
    input.residentRows &&
    input.knownCount !== null &&
    input.residentRows.length !== (input.knownCount ?? input.count)
  ) {
    throw new Error(
      `Resident Dataset Handle ${input.datasetId} has ${input.residentRows.length} rows but declares ${input.knownCount ?? input.count}.`,
    );
  }
  const target = new DeferredPlayDataset({
    ...input,
    previewRows: input.previewRows ?? [],
  });
  const boundMethods = new Map<PropertyKey, unknown>();
  const dataset = new Proxy(target, {
    get(dataset, property) {
      if (property === 'length') {
        throw datasetAsyncOnlyError(
          dataset,
          'Dataset Handles do not expose synchronous .length.',
          'Use await dataset.count().',
        );
      }
      if (typeof property === 'string' && /^(0|[1-9]\d*)$/.test(property)) {
        throw datasetAsyncOnlyError(
          dataset,
          `Dataset Handles do not expose synchronous row indexing (${property}).`,
          `Use await dataset.at(${property}) or await dataset.first().`,
        );
      }
      if (property === Symbol.iterator) {
        return () => {
          throw datasetAsyncOnlyError(
            dataset,
            'Dataset Handles do not support synchronous iteration.',
            'Use for await...of or await dataset.materialize(limit).',
          );
        };
      }
      const value = Reflect.get(dataset, property, dataset);
      if (typeof value !== 'function') return value;
      if (!boundMethods.has(property)) {
        boundMethods.set(property, value.bind(dataset));
      }
      return boundMethods.get(property);
    },
    set(_dataset, property) {
      if (
        property === 'length' ||
        (typeof property === 'string' && /^(0|[1-9]\d*)$/.test(property))
      ) {
        throw datasetAsyncOnlyError(
          target,
          'Dataset Handles do not support synchronous array assignment.',
          'Transform rows with dataset.map(...) or explicitly materialize a bounded array.',
        );
      }
      return false;
    },
  });
  if (input.residentRows) {
    residentRowsByDataset.set(target, input.residentRows);
    residentRowsByDataset.set(dataset, input.residentRows);
  }
  return dataset;
}

function datasetAsyncOnlyError(
  dataset: Pick<PlayDataset<unknown>, 'datasetId' | 'sourceLabel'>,
  detail: string,
  guidance: string,
): Error {
  return new Error(
    `PLAY_DATASET_ASYNC_ONLY: ${detail} ` +
      `Dataset Handle ${dataset.sourceLabel ?? dataset.datasetId}. ${guidance}`,
  );
}

export function createPlayDataset<T>(
  rows: readonly T[],
  metadata?: {
    kind?: PlayDatasetKind;
    sourceLabel?: string | null;
    tableNamespace?: string | null;
    datasetId?: string;
  },
): PlayDataset<T> {
  // Test/dev helper only. Production play execution should construct
  // Neon-backed handles via createDeferredPlayDataset + runtime-host helpers.
  const materializedRows = [...rows];
  return createDeferredPlayDataset({
    datasetKind: metadata?.kind ?? 'map',
    datasetId:
      metadata?.datasetId ??
      `${metadata?.kind ?? 'map'}:${metadata?.tableNamespace ?? metadata?.sourceLabel ?? 'inline'}`,
    count: materializedRows.length,
    previewRows: materializedRows.slice(0, 5),
    residentRows: materializedRows,
    sourceLabel: metadata?.sourceLabel ?? null,
    tableNamespace: metadata?.tableNamespace ?? null,
    resolvers: {
      count: async () => materializedRows.length,
      peek: async (limit) => materializedRows.slice(0, Math.max(0, limit)),
      materialize: async (limit) =>
        limit === undefined
          ? [...materializedRows]
          : materializedRows.slice(0, Math.max(0, limit)),
      iterate: () =>
        ({
          async *[Symbol.asyncIterator]() {
            for (const row of materializedRows) {
              yield row;
            }
          },
        }) as AsyncIterable<T>,
    },
  });
}

export type MaterializePlayDatasetInputOptions<T> = {
  onRow?: (row: T, index: number) => void;
};

export type IteratePlayDatasetInputPagesOptions<T> = {
  pageSize?: number;
  maxPageBytes?: number;
  estimateRowBytes?: (row: T) => number;
  onRow?: (row: T, index: number) => void;
};

function normalizeDatasetExecutionPageSize(value: number | undefined): number {
  if (typeof value !== 'number' || !Number.isFinite(value)) {
    return PLAY_DATASET_EXECUTION_PAGE_ROWS;
  }
  return Math.max(1, Math.floor(value));
}

export async function* iteratePlayDatasetInputPages<T>(
  input: PlayDatasetInput<T>,
  options?: IteratePlayDatasetInputPagesOptions<T>,
): AsyncIterable<{ rows: T[]; offset: number }> {
  const pageSize = normalizeDatasetExecutionPageSize(options?.pageSize);
  const maxPageBytes = Math.max(
    1,
    Math.floor(options?.maxPageBytes ?? PLAY_DATASET_EXECUTION_PAGE_BYTES),
  );
  const estimateRowBytes =
    options?.estimateRowBytes ??
    ((row: T) => {
      const serialized = JSON.stringify(row);
      return serialized === undefined ? 0 : serialized.length;
    });
  let page: T[] = [];
  let pageBytes = 0;
  let offset = 0;
  let index = 0;
  const pushRow = (row: T): Array<{ rows: T[]; offset: number }> => {
    options?.onRow?.(row, index);
    index += 1;
    const rowBytes = Math.max(0, Math.ceil(estimateRowBytes(row)));
    const ready: Array<{ rows: T[]; offset: number }> = [];
    if (page.length > 0 && pageBytes + rowBytes > maxPageBytes) {
      ready.push({ rows: page, offset });
      offset += page.length;
      page = [];
      pageBytes = 0;
    }
    page.push(row);
    pageBytes += rowBytes;
    if (page.length >= pageSize || pageBytes >= maxPageBytes) {
      ready.push({ rows: page, offset });
      offset += page.length;
      page = [];
      pageBytes = 0;
    }
    return ready;
  };

  const flush = (): { rows: T[]; offset: number } | null => {
    if (page.length === 0) return null;
    const finalPage = page;
    const finalOffset = offset;
    page = [];
    pageBytes = 0;
    offset += finalPage.length;
    return { rows: finalPage, offset: finalOffset };
  };

  const emitRow = async function* (row: T) {
    for (const ready of pushRow(row)) yield ready;
  };

  if (isPlayDataset(input)) {
    for await (const row of input) {
      yield* emitRow(row);
    }
    const finalPage = flush();
    if (finalPage) yield finalPage;
    return;
  }

  if (Array.isArray(input)) {
    for (const row of input) {
      yield* emitRow(row);
    }
    const finalPage = flush();
    if (finalPage) yield finalPage;
    return;
  }

  if (
    input != null &&
    typeof input === 'object' &&
    Symbol.asyncIterator in input
  ) {
    for await (const row of input as AsyncIterable<T>) {
      yield* emitRow(row);
    }
    const finalPage = flush();
    if (finalPage) yield finalPage;
    return;
  }

  for (const row of input as Iterable<T>) {
    yield* emitRow(row);
  }
  const finalPage = flush();
  if (finalPage) yield finalPage;
}

export async function materializePlayDatasetInput<T>(
  input: PlayDatasetInput<T>,
  options?: MaterializePlayDatasetInputOptions<T>,
): Promise<T[]> {
  const pushRow = (rows: T[], row: T): void => {
    options?.onRow?.(row, rows.length);
    rows.push(row);
  };

  if (isPlayDataset(input)) {
    const rows: T[] = [];
    for await (const row of input) {
      pushRow(rows, row);
    }
    return rows;
  }

  if (Array.isArray(input)) {
    const rows: T[] = [];
    for (const row of input) {
      pushRow(rows, row);
    }
    return rows;
  }

  const rows: T[] = [];
  if (
    input != null &&
    typeof input === 'object' &&
    Symbol.asyncIterator in input
  ) {
    for await (const row of input as AsyncIterable<T>) {
      pushRow(rows, row);
    }
    return rows;
  }

  for (const row of input as Iterable<T>) {
    pushRow(rows, row);
  }
  return rows;
}