import {
  canonicalJson, fnv1a64, snapshotDiagnosticJson, CUSTOMIZATION_CAPABILITIES,
  createBoundedGroupedDevelopmentAccumulator,
  BOUNDED_GROUPED_DEVELOPMENT_CONTRACT_VERSION,
  isDiagnosticToken,
  revisionInstantKey,
  type BoundedGroupedDevelopmentRequest,
  type BoundedGroupedDevelopmentResult,
  type CustomizationResourceLimits,
  type HistoricalDatasetInput, type JsonValue,
} from "@actuarial-ts/core";
import { z } from "zod";
import { parseReviewBoundary } from "./reviewBoundary.js";
import {
  analysisDefinitionSchema,
  analysisRecipeSchema,
  parseCustomizationResourceLimits,
  parseHistoricalDatasetInput,
  parseHistoricalObservationRecord,
} from "./customizationContracts.js";
import { CustomizationExecutionError, type CustomizationExternalRecord, type CustomizationExternalStateSession, type CustomizationExternalStateAdapter } from "./customizationExternalState.js";
import { streamPreparedHistoricalDataset, type PreparedHistoryStreamEvent } from "./customizationHistoryStream.js";

const token = z.string().refine(isDiagnosticToken, "Expected a valid SDK token");
const date = z.string().regex(/^\d{4}-\d{2}-\d{2}$/);
export const boundedGroupedDevelopmentRequestSchema: z.ZodType<BoundedGroupedDevelopmentRequest> = z.object({
  definition: analysisDefinitionSchema,
  recipe: analysisRecipeSchema,
  origins: z.array(z.object({ id: token, startDate: date, endDate: date }).strict()),
  ages: z.array(z.number().int().positive()),
  ageConvention: z.enum(["calendar-period-end", "elapsed-months"]),
  evaluationDates: z.array(date),
  knowledgeDate: date,
  coverage: z.enum(["exact-universe", "declared-subset"]),
  measures: z.array(z.object({ id: token, sourceMeasureId: token, kind: z.enum(["paid", "incurred"]) }).strict()),
  maximumObservations: z.number().int().positive().safe(),
}).strict();

export const boundedGroupedDevelopmentResultSchema: z.ZodType<BoundedGroupedDevelopmentResult> = z.object({
  contractVersion: z.literal(BOUNDED_GROUPED_DEVELOPMENT_CONTRACT_VERSION),
  inputIdentity: token,
  boundaries: z.array(z.object({ origin: token, value: z.literal(0), evidence: z.literal("structural-origin-zero"), sourceObservations: z.number().int().nonnegative().safe() }).strict()),
  preparation: z.object({ revisionView: z.enum(["restated", "as-known"]), dispositions: z.object({ selected: z.number().int().nonnegative().safe(), duplicate: z.number().int().nonnegative().safe(), superseded: z.number().int().nonnegative().safe(), "after-knowledge-cutoff": z.number().int().nonnegative().safe(), deleted: z.number().int().nonnegative().safe(), conflict: z.number().int().nonnegative().safe() }).strict(), dispositionIdentity: token, findings: z.array(z.object({ code: z.enum(["missing-prerequisite", "unsupported-operation", "incompatible-scope", "resource-limit", "ambiguous-attribution", "incomplete-financial-population", "incompatible-semantics", "invalid-configuration"]), capability: z.enum(CUSTOMIZATION_CAPABILITIES), path: z.string(), message: z.string(), requiredFields: z.array(z.string()).optional() }).strict()) }).strict().nullable(),
  origins: z.array(z.object({ id: token, startDate: date, endDate: date }).strict()),
  ages: z.array(z.number().int().positive().safe()),
  ageConvention: z.enum(["calendar-period-end", "elapsed-months"]),
  knowledgeDate: date,
  selectedObservations: z.number().int().nonnegative().safe(),
  excludedObservations: z.number().int().nonnegative().safe(),
  exclusions: z.object({ outsideOriginUniverse: z.number().int().nonnegative().safe(), outsideDevelopmentGrid: z.number().int().nonnegative().safe() }).strict(),
  measures: z.array(z.object({
    id: token,
    kind: z.enum(["paid", "incurred"]),
    sourceMeasureId: token,
    unit: token,
    values: z.array(z.array(z.number().finite().nullable())),
    valuations: z.array(z.array(date.nullable())),
    reasons: z.array(z.array(z.enum(["NO_OBSERVATIONS", "MISSING_MEASURE", "PREPARATION_VALUATION_MISSING", "AFTER_PREPARED_VALUATIONS"]).nullable())),
  }).strict()),
}).strict();

export const BOUNDED_GROUPED_HISTORY_WORK_LIMITS = Object.freeze({
  maximumInputBytes: 256_000_000,
  maximumSpillBytes: 1_000_000_000,
  maximumElapsedMs: 300_000,
  maximumEvidenceBytes: 16_000_000,
  maximumFindings: 10_000,
});
export type BoundedGroupedHistoryWorkLimits = { readonly [K in keyof typeof BOUNDED_GROUPED_HISTORY_WORK_LIMITS]: number };
export const boundedGroupedHistoryWorkLimitsSchema: z.ZodType<BoundedGroupedHistoryWorkLimits> = z.object(Object.fromEntries(Object.entries(BOUNDED_GROUPED_HISTORY_WORK_LIMITS).map(([key, ceiling]) => [key, z.number().int().positive().safe().max(ceiling)]))).strict() as unknown as z.ZodType<BoundedGroupedHistoryWorkLimits>;
/** Complete bounded native finding evidence for a refused preparation; never a calculation receipt. */
export class BoundedGroupedHistoryPreparationError extends Error {
  readonly code = "EXTERNAL_STATE_FAILURE";
  readonly findings: NonNullable<BoundedGroupedDevelopmentResult["preparation"]>["findings"];
  constructor(findings: NonNullable<BoundedGroupedDevelopmentResult["preparation"]>["findings"]) {
    super(`Grouped history preparation is blocked (${findings.map(finding => finding.code).join(", ")})`);
    this.name = "BoundedGroupedHistoryPreparationError";
    this.findings = snapshotDiagnosticJson(findings);
    Object.freeze(this);
  }
}
export function parseBoundedGroupedDevelopmentRequest(input: unknown): BoundedGroupedDevelopmentRequest {
  const request = parseReviewBoundary(boundedGroupedDevelopmentRequestSchema, input);
  createBoundedGroupedDevelopmentAccumulator(request);
  return request;
}

/**
 * Executes SDK revision selection and origin/valuation accumulation without
 * materializing the complete claim history. This is calculation output, not
 * source authentication or an independently verified replay receipt.
 */
export async function projectStreamedHistoricalDevelopment(input: {
  readonly runId: string;
  readonly header: Omit<HistoricalDatasetInput, "records">;
  readonly records: AsyncIterable<unknown> | Iterable<unknown>;
  readonly request: unknown;
  readonly adapter: CustomizationExternalStateAdapter;
  readonly limits: CustomizationResourceLimits;
  readonly signal?: AbortSignal;
  readonly workLimits?: BoundedGroupedHistoryWorkLimits;
  /** Optional streaming evidence sink; serialized decisions never confer authority. */
  readonly onPreparationEvent?: (event: PreparedHistoryStreamEvent) => void | Promise<void>;
}): Promise<BoundedGroupedDevelopmentResult> {
  if (!isDiagnosticToken(input.runId))
    throw new CustomizationExecutionError("EXTERNAL_STATE_FAILURE", "Grouped development runId is invalid");
  const parsed = parseBoundedGroupedDevelopmentRequest(input.request);
  const workLimits = boundedGroupedHistoryWorkLimitsSchema.parse(input.workLimits ?? BOUNDED_GROUPED_HISTORY_WORK_LIMITS);
  const started = performance.now();
  const check = () => {
    if (input.signal?.aborted) throw new CustomizationExecutionError("CANCELLED", "Grouped development was cancelled");
    if (performance.now() - started > workLimits.maximumElapsedMs) throw new CustomizationExecutionError("RESOURCE_LIMIT", "Grouped development exceeds maximumElapsedMs");
  };
  let inputBytes = 0;
  const encoder = new TextEncoder();
  const limits = parseCustomizationResourceLimits(input.limits);
  if (parsed.maximumObservations > limits.maximumInputRecords)
    throw new CustomizationExecutionError("RESOURCE_LIMIT", "maximumObservations exceeds maximumInputRecords");
  const full = parseHistoricalDatasetInput({ ...input.header, records: [] });
  const { records: _empty, ...header } = full;
  if (header.historyMode !== "full-snapshots" || header.missingSnapshotRecord !== "unknown")
    throw new CustomizationExecutionError("EXTERNAL_STATE_FAILURE", "Grouped development currently requires observed full snapshots without carried-forward claims");
  if (parsed.recipe.steps[0]?.kind !== "prepare-history" || parsed.recipe.steps[0].historyContractId !== header.datasetId)
    throw new CustomizationExecutionError("EXTERNAL_STATE_FAILURE", "Grouped recipe history contract differs from the streamed dataset");
  if (header.evaluationDates && parsed.evaluationDates.some((date) => !header.evaluationDates!.includes(date)))
    throw new CustomizationExecutionError("EXTERNAL_STATE_FAILURE", "A requested evaluation is absent from the history header");
  const cutoff = header.revisionView.kind === "as-known"
    ? revisionInstantKey(header.revisionView.knowledgeCutoff)
    : null;
  if (header.revisionView.kind === "as-known" && (!cutoff || cutoff.slice(0, 10) > parsed.knowledgeDate))
    throw new CustomizationExecutionError("EXTERNAL_STATE_FAILURE", "Revision knowledge cutoff exceeds the grouped development knowledge date");
  async function* checkedRecords() {
    for await (const raw of input.records) {
      if (input.signal?.aborted)
        throw new CustomizationExecutionError("CANCELLED", "Grouped development was cancelled");
      const record = parseHistoricalObservationRecord(raw);
      const corrected = record.revision.correctedAt === undefined ? null : revisionInstantKey(record.revision.correctedAt);
      if (record.revision.correctedAt !== undefined && corrected === null)
        throw new CustomizationExecutionError("EXTERNAL_STATE_FAILURE", "Revision correction date is invalid");
      if (
        (header.revisionView.kind === "restated" || corrected === null || corrected <= cutoff!) &&
        [record.evaluationDate, record.effectiveDate, corrected?.slice(0, 10)].some(
          (date) => date !== undefined && date !== null && date > parsed.knowledgeDate,
        )
      )
        throw new CustomizationExecutionError("EXTERNAL_STATE_FAILURE", "Source history contains information after the grouped development knowledge date");
      inputBytes += encoder.encode(canonicalJson(record)).byteLength + 1;
      if (inputBytes > workLimits.maximumInputBytes) throw new CustomizationExecutionError("RESOURCE_LIMIT", "Grouped history exceeds maximumInputBytes");
      check();
      yield record;
    }
  }
  const accumulator = createBoundedGroupedDevelopmentAccumulator(parsed);
  const retainedCellBytes = parsed.origins.length * parsed.ages.length * parsed.measures.length * 64;
  const adapterWorkingBudget = Math.floor((limits.maximumInMemoryBytes - retainedCellBytes) / 2);
  const partitionLimits = { ...limits, maximumInMemoryBytes: limits.maximumInMemoryBytes - retainedCellBytes - adapterWorkingBudget };
  if (retainedCellBytes >= limits.maximumInMemoryBytes || adapterWorkingBudget < 1)
    throw new CustomizationExecutionError("RESOURCE_LIMIT", "Grouped accumulator exceeds maximumInMemoryBytes retained-cell budget");
  const dispositions = { selected: 0, duplicate: 0, superseded: 0, "after-knowledge-cutoff": 0, deleted: 0, conflict: 0 };
  let dispositionIdentity = "fnv1a64-jcs-v1:0000000000000000";
  const findings: NonNullable<BoundedGroupedDevelopmentResult["preparation"]>["findings"][number][] = [];
  const boundedAdapter: CustomizationExternalStateAdapter = { open: async <T extends JsonValue>(request: Parameters<CustomizationExternalStateAdapter["open"]>[0]): Promise<CustomizationExternalStateSession<T>> => {
    check();
    const session = await input.adapter.open<T>(request);
    if (!session.resourceUsage) { await session.dispose("failed"); throw new CustomizationExecutionError("RESOURCE_LIMIT", "Grouped history requires adapter resourceUsage accounting"); }
    const usage = () => {
      check();
      const value = session.resourceUsage!();
      if (![value.workingBytes, value.spillBytes].every(v => Number.isSafeInteger(v) && v >= 0) || value.workingBytes > adapterWorkingBudget || value.spillBytes > workLimits.maximumSpillBytes)
        throw new CustomizationExecutionError("RESOURCE_LIMIT", "Grouped external-state working memory or spill budget exceeded");
    };
    return { ...session, append: async (record: CustomizationExternalRecord<T>) => { usage(); await session.append(record); usage(); }, seal: async () => { usage(); await session.seal(); usage(); }, read: async function* () { for await (const record of session.read()) { usage(); yield record; } } };
  } };
  check();
  for await (const event of streamPreparedHistoricalDataset({
    runId: input.runId,
    header,
    records: checkedRecords(),
    adapter: boundedAdapter,
    limits: partitionLimits,
    ...(input.signal === undefined ? {} : { signal: input.signal }),
  })) {
    check();
    await input.onPreparationEvent?.(event);
    if (event.kind === "disposition") {
      dispositions[event.disposition.disposition] += 1;
      dispositionIdentity = `fnv1a64-jcs-v1:${fnv1a64(canonicalJson([dispositionIdentity, event.disposition]))}`;
    }
    if (event.kind === "finding") {
      findings.push(event.finding);
      if (findings.length > workLimits.maximumFindings) throw new CustomizationExecutionError("RESOURCE_LIMIT", "Grouped preparation exceeds maximumFindings");

    }
    if (event.kind === "observation" && findings.length === 0) accumulator.add(event.observation);
  }
  check();
  if (findings.length) throw new BoundedGroupedHistoryPreparationError(findings);
  const result = snapshotDiagnosticJson({ ...accumulator.complete(), preparation: { revisionView: header.revisionView.kind, dispositions, dispositionIdentity, findings } });
  if (encoder.encode(canonicalJson(result)).byteLength > workLimits.maximumEvidenceBytes) throw new CustomizationExecutionError("RESOURCE_LIMIT", "Grouped result exceeds maximumEvidenceBytes");
  boundedGroupedDevelopmentResultSchema.parse(result);
  check();
  return result;
}