export const NODE_RUNTIME_MAP_MEMORY_BUDGET_BYTES = 512 * 1024 * 1024;
export const NODE_RUNTIME_MAP_ACTIVE_ROWS_MEMORY_BUDGET_BYTES =
  384 * 1024 * 1024;
export const NODE_RUNTIME_MAP_MATERIALIZED_MEMORY_MULTIPLIER = 4;
export const NODE_RUNTIME_MAP_ACTIVE_ROW_MEMORY_MULTIPLIER = 3;
export const NODE_RUNTIME_MAP_ROW_OVERHEAD_BYTES = 512;

export type RuntimeMapMemoryLimits = {
  materializedBudgetBytes: number;
  activeRowsBudgetBytes: number;
};

export type RuntimeMapMemoryEstimate = {
  rowCount: number;
  serializedBytes: number;
  estimatedResidentBytes: number;
  largestRowBytes: number;
};

export type RuntimeMapRowAdmission = {
  concurrency: number;
  estimatedBytesPerActiveRow: number;
  estimatedResidentBytes: number;
};

export class RuntimeMapMemoryLimitError extends Error {
  readonly code = 'RUNTIME_MAP_MEMORY_LIMIT_EXCEEDED';
  constructor(input: {
    mapName: string;
    phase: 'materialize' | 'active_rows' | 'retained_rows';
    estimate: RuntimeMapMemoryEstimate;
    budgetBytes: number;
    activeRows?: number;
    rowConcurrency?: number;
  }) {
    const active =
      input.activeRows === undefined
        ? ''
        : ` activeRows=${input.activeRows}` +
          (input.rowConcurrency === undefined
            ? ''
            : ` rowConcurrency=${input.rowConcurrency}`);
    const stopPoint =
      input.phase === 'materialize'
        ? ' before provider calls'
        : input.phase === 'active_rows'
          ? ' before row resolver bodies start'
          : '';
    super(
      `Runtime map memory limit exceeded for ctx.dataset("${input.mapName}") during ${input.phase}: ` +
        `estimatedResidentBytes=${input.estimate.estimatedResidentBytes} ` +
        `budgetBytes=${input.budgetBytes} rowCount=${input.estimate.rowCount} ` +
        `serializedBytes=${input.estimate.serializedBytes} largestRowBytes=${input.estimate.largestRowBytes}${active}. ` +
        `This run would risk sandbox OOM. Deepline stops here${stopPoint} instead of silently lowering concurrency. ` +
        'Reduce row payload size, stage bulky blobs, or split the input into smaller runs.',
    );
    this.name = 'RuntimeMapMemoryLimitError';
  }
}

const textEncoder =
  typeof TextEncoder === 'undefined' ? null : new TextEncoder();

export function runtimeMapJsonByteLength(value: unknown): number {
  const serialized = JSON.stringify(value);
  if (serialized === undefined) return 0;
  return textEncoder?.encode(serialized).byteLength ?? serialized.length;
}

export function defaultRuntimeMapMemoryLimits(): RuntimeMapMemoryLimits {
  return {
    materializedBudgetBytes: NODE_RUNTIME_MAP_MEMORY_BUDGET_BYTES,
    activeRowsBudgetBytes: NODE_RUNTIME_MAP_ACTIVE_ROWS_MEMORY_BUDGET_BYTES,
  };
}

export function resolveRuntimeMapMemoryLimits(
  override?: Partial<RuntimeMapMemoryLimits> | null,
): RuntimeMapMemoryLimits {
  const defaults = defaultRuntimeMapMemoryLimits();
  return {
    materializedBudgetBytes: positiveBudget(
      override?.materializedBudgetBytes,
      defaults.materializedBudgetBytes,
    ),
    activeRowsBudgetBytes: positiveBudget(
      override?.activeRowsBudgetBytes,
      defaults.activeRowsBudgetBytes,
    ),
  };
}

function positiveBudget(value: number | undefined, fallback: number): number {
  if (!Number.isFinite(value) || value === undefined) return fallback;
  return Math.max(1, Math.floor(value));
}

export function estimateRuntimeMapMaterializedMemory(input: {
  rowCount: number;
  serializedBytes: number;
  largestRowBytes: number;
}): RuntimeMapMemoryEstimate {
  return {
    rowCount: input.rowCount,
    serializedBytes: input.serializedBytes,
    largestRowBytes: input.largestRowBytes,
    estimatedResidentBytes:
      input.serializedBytes * NODE_RUNTIME_MAP_MATERIALIZED_MEMORY_MULTIPLIER +
      input.rowCount * NODE_RUNTIME_MAP_ROW_OVERHEAD_BYTES,
  };
}

export function estimateRuntimeMapRowsMemory(
  rows: readonly unknown[],
): RuntimeMapMemoryEstimate {
  let serializedBytes = 0;
  let largestRowBytes = 0;
  for (const row of rows) {
    const rowBytes = runtimeMapJsonByteLength(row);
    serializedBytes += rowBytes;
    largestRowBytes = Math.max(largestRowBytes, rowBytes);
  }
  return estimateRuntimeMapMaterializedMemory({
    rowCount: rows.length,
    serializedBytes,
    largestRowBytes,
  });
}

type RuntimeMapRowsTrackerInput = {
  mapName: string;
  budgetBytes?: number;
};

type RuntimeMapRowsTracker = {
  track(row: unknown): void;
  estimate(): RuntimeMapMemoryEstimate;
};

function createRuntimeMapRowsTracker(
  input: RuntimeMapRowsTrackerInput,
  phase: 'materialize' | 'retained_rows',
): RuntimeMapRowsTracker {
  const budgetBytes = positiveBudget(
    input.budgetBytes,
    NODE_RUNTIME_MAP_MEMORY_BUDGET_BYTES,
  );
  let rowCount = 0;
  let serializedBytes = 0;
  let largestRowBytes = 0;
  const estimate = () =>
    estimateRuntimeMapMaterializedMemory({
      rowCount,
      serializedBytes,
      largestRowBytes,
    });
  return {
    track(row) {
      const rowBytes = runtimeMapJsonByteLength(row);
      rowCount += 1;
      serializedBytes += rowBytes;
      largestRowBytes = Math.max(largestRowBytes, rowBytes);
      const current = estimate();
      if (current.estimatedResidentBytes > budgetBytes) {
        throw new RuntimeMapMemoryLimitError({
          mapName: input.mapName,
          phase,
          estimate: current,
          budgetBytes,
        });
      }
    },
    estimate,
  };
}

export function createRuntimeMapMaterializationTracker(
  input: RuntimeMapRowsTrackerInput,
): RuntimeMapRowsTracker {
  return createRuntimeMapRowsTracker(input, 'materialize');
}

export function createRuntimeMapRetainedRowsTracker(
  input: RuntimeMapRowsTrackerInput,
): RuntimeMapRowsTracker {
  return createRuntimeMapRowsTracker(input, 'retained_rows');
}

export function assertRuntimeMapActiveRowsWithinMemoryLimit(input: {
  mapName: string;
  rowCount: number;
  rowConcurrency: number;
  largestRowBytes: number;
  activeRowsBudgetBytes?: number;
}): void {
  const activeRows = Math.min(
    Math.max(0, input.rowCount),
    Math.max(1, Math.floor(input.rowConcurrency)),
  );
  const estimate: RuntimeMapMemoryEstimate = {
    rowCount: input.rowCount,
    serializedBytes: input.largestRowBytes * activeRows,
    largestRowBytes: input.largestRowBytes,
    estimatedResidentBytes:
      input.largestRowBytes *
        activeRows *
        NODE_RUNTIME_MAP_ACTIVE_ROW_MEMORY_MULTIPLIER +
      activeRows * NODE_RUNTIME_MAP_ROW_OVERHEAD_BYTES,
  };
  const budgetBytes = positiveBudget(
    input.activeRowsBudgetBytes,
    NODE_RUNTIME_MAP_ACTIVE_ROWS_MEMORY_BUDGET_BYTES,
  );
  if (estimate.estimatedResidentBytes > budgetBytes) {
    throw new RuntimeMapMemoryLimitError({
      mapName: input.mapName,
      phase: 'active_rows',
      estimate,
      budgetBytes,
      activeRows,
      rowConcurrency: input.rowConcurrency,
    });
  }
}

/**
 * Resolve a row-worker window from the shared concurrency target and a concrete
 * resident-byte budget. Pending rows remain indexes, so only this admitted
 * window contributes active row copies/promises. If one row cannot fit, fail
 * before any resolver body starts; otherwise memory pressure is normal
 * backpressure and reduces the window explicitly.
 */
export function resolveRuntimeMapRowAdmission(input: {
  rowCount: number;
  requestedConcurrency: number;
  largestRowBytes: number;
  activeRowsBudgetBytes?: number;
}): RuntimeMapRowAdmission {
  const rowCount = Math.max(0, Math.floor(input.rowCount));
  const requestedConcurrency = Math.max(
    1,
    Math.floor(input.requestedConcurrency),
  );
  const largestRowBytes = Math.max(0, Math.floor(input.largestRowBytes));
  const budgetBytes = positiveBudget(
    input.activeRowsBudgetBytes,
    NODE_RUNTIME_MAP_ACTIVE_ROWS_MEMORY_BUDGET_BYTES,
  );
  const estimatedBytesPerActiveRow =
    largestRowBytes * NODE_RUNTIME_MAP_ACTIVE_ROW_MEMORY_MULTIPLIER +
    NODE_RUNTIME_MAP_ROW_OVERHEAD_BYTES;

  const maxConcurrencyByMemory = Math.floor(
    budgetBytes / Math.max(1, estimatedBytesPerActiveRow),
  );
  const concurrency =
    rowCount === 0
      ? 0
      : Math.min(rowCount, requestedConcurrency, maxConcurrencyByMemory);
  return {
    concurrency,
    estimatedBytesPerActiveRow,
    estimatedResidentBytes: estimatedBytesPerActiveRow * concurrency,
  };
}