import { sql } from 'drizzle-orm'
import type { Querier, TransactionIdentity } from '@platform-modules/db'
import { resolveFieldDefinitionVersion } from './definitions.js'
import type { EntityRef } from './model.js'
import type {
  AnyFieldDefinition,
  FieldStorageValue,
  FieldValueNodeRow,
  FieldsSchema,
  FieldsTransaction,
  StoredBlockDocument,
} from './schema.js'
import {
  assertActiveFieldsTransaction,
  assertFieldsAuthorized,
  assertReceiptReplay,
  fieldsRequestHash,
  readFieldsReceipt,
  writeFieldsReceipt,
  FieldsLifecycleError,
} from './definition-lifecycle.js'
import { createFieldTypeRegistry, type FieldsContext } from './types.js'
import {
  decodeFieldValueNodes,
  encodeFieldValueNodes,
  redactFieldValueForRevision,
  type FieldValueMap,
} from './values.js'

export interface FieldsRevision {
  readonly id: string
  readonly ref: EntityRef
  readonly parentRevisionId: string
  readonly parentAutosaveId?: string
  readonly kind: 'revision' | 'autosave'
  readonly retentionClass?: string
  readonly definitionVersions: Readonly<Record<string, number>>
  readonly values: Readonly<Record<string, FieldStorageValue>>
  readonly blockDocument?: StoredBlockDocument
  readonly createdAt: Date
  readonly createdBy: string
  readonly sourceVersion: string
}
export interface FieldsRevisionCursor { readonly createdAt: Date; readonly id: string }
export interface FieldsRevisionPage { readonly items: readonly FieldsRevision[]; readonly nextCursor: FieldsRevisionCursor | null }
export interface FieldsRevisionDiff {
  readonly valuePaths: readonly { path: readonly (string | number)[]; kind: 'added' | 'removed' | 'changed' }[]
  readonly blockNodes: readonly { nodeId: string; kind: 'added' | 'removed' | 'moved' | 'changed' }[]
}
export interface FieldsRevisionMutation {
  readonly expectedVersion: string
  readonly operationId: string
  readonly expectedTransactionIdentity: TransactionIdentity
  readonly parentRevisionId: string
}
export interface FieldsAutosaveMutation extends FieldsRevisionMutation {
  readonly parentAutosaveId: string
  readonly retentionClass: string
}
export interface FieldsAutosaveRetentionPolicy {
  readonly maxPerParent: number
  readonly maxAgeMs: number
  readonly classes: Readonly<Record<string, { maxPerParent: number; maxAgeMs: number }>>
}

type FieldsRevisionReader = Querier<FieldsSchema> | FieldsTransaction

function rowsOf<Row extends Record<string, unknown>>(result: unknown): readonly Row[] {
  if (Array.isArray(result)) return result as Row[]
  if (typeof result === 'object' && result !== null) {
    const rows=(result as { rows?: unknown }).rows
    if (Array.isArray(rows)) return rows as Row[]
  }
  throw new FieldsLifecycleError('integrity', 'database adapter returned an invalid row result')
}

async function executeRows<Row extends Record<string, unknown>>(db: FieldsRevisionReader, query: Parameters<FieldsTransaction['execute']>[0]): Promise<readonly Row[]> {
  return rowsOf<Row>(await db.execute(query))
}

interface NodeRow extends Record<string, unknown> {
  nodeId: string
  entityType: string
  entityId: string
  groupKey: string
  definitionRevision: number
  fieldKey: string
  path: string
  parentNodeId: string | null
  rowId: string | null
  layoutKey: string | null
  nodeKind: FieldValueNodeRow['nodeKind']
  ordinal: number
  isNull: boolean | number
  valueText: string | null
  valueNumber: string | number | null
  valueBoolean: boolean | number | null
  valueDateTime: string | null
  valueRef: string | null
  valueJson: string | null
}
interface BlockDocumentRow extends Record<string, unknown> {
  version: number
  document: unknown
  definitionVersions: unknown
}
interface RevisionRow extends Record<string, unknown> {
  id: string
  entityType: string
  entityId: string
  parentRevisionId: string
  parentAutosaveId: string | null
  kind: 'revision' | 'autosave'
  retentionClass: string | null
  definitionVersions: unknown
  values: unknown
  blockDocument: unknown | null
  createdAt: Date | string
  createdBy: string
  sourceVersion: string
}

export function fieldsRuntimeMigrationSql(adapter: 'postgres' | 'd1'): readonly string[] {
  if (adapter === 'postgres') return Object.freeze([
    `CREATE TABLE IF NOT EXISTS field_block_documents (entity_type text NOT NULL, entity_id text NOT NULL, version integer NOT NULL DEFAULT 1 CHECK (version > 0), document jsonb NOT NULL, definition_versions jsonb NOT NULL, parent_revision_id text NOT NULL, host_session_id text, updated_by text NOT NULL, updated_at timestamptz(3) NOT NULL DEFAULT NOW(), PRIMARY KEY (entity_type, entity_id))`,
    `CREATE TABLE IF NOT EXISTS field_reusable_blocks (id text PRIMARY KEY, version integer NOT NULL DEFAULT 1 CHECK (version > 0), document jsonb NOT NULL, definition_versions jsonb NOT NULL, deleted_at timestamptz(3), host_session_id text, updated_by text NOT NULL, updated_at timestamptz(3) NOT NULL DEFAULT NOW())`,
    `CREATE TABLE IF NOT EXISTS field_revisions (id uuid PRIMARY KEY, entity_type text NOT NULL, entity_id text NOT NULL, parent_revision_id text NOT NULL, parent_autosave_id text, kind text NOT NULL CHECK (kind IN ('revision','autosave')), retention_class text, definition_versions jsonb NOT NULL, values jsonb NOT NULL, block_document jsonb, created_at timestamptz(3) NOT NULL DEFAULT NOW(), created_by text NOT NULL, source_version text NOT NULL, host_session_id text)`,
    `CREATE INDEX IF NOT EXISTS field_revisions_entity_created_idx ON field_revisions (entity_type, entity_id, created_at, id)`,
    `CREATE INDEX IF NOT EXISTS field_revisions_parent_idx ON field_revisions (parent_revision_id, kind)`,
    `CREATE TABLE IF NOT EXISTS fields_operation_receipts (operation_id text PRIMARY KEY, kind text NOT NULL, principal_scope text NOT NULL, request_hash text NOT NULL, result jsonb NOT NULL, created_at timestamptz(3) NOT NULL DEFAULT NOW())`,
    `CREATE INDEX IF NOT EXISTS fields_operation_receipts_kind_idx ON fields_operation_receipts (kind, created_at)`,
  ])
  return Object.freeze([
    `CREATE TABLE IF NOT EXISTS field_block_documents (entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, version INTEGER NOT NULL DEFAULT 1 CHECK (version > 0), document TEXT NOT NULL, definition_versions TEXT NOT NULL, parent_revision_id TEXT NOT NULL, host_session_id TEXT, updated_by TEXT NOT NULL, updated_at TEXT NOT NULL, PRIMARY KEY (entity_type, entity_id))`,
    `CREATE TABLE IF NOT EXISTS field_reusable_blocks (id TEXT PRIMARY KEY, version INTEGER NOT NULL DEFAULT 1 CHECK (version > 0), document TEXT NOT NULL, definition_versions TEXT NOT NULL, deleted_at TEXT, host_session_id TEXT, updated_by TEXT NOT NULL, updated_at TEXT NOT NULL)`,
    `CREATE TABLE IF NOT EXISTS field_revisions (id TEXT PRIMARY KEY, entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, parent_revision_id TEXT NOT NULL, parent_autosave_id TEXT, kind TEXT NOT NULL CHECK (kind IN ('revision','autosave')), retention_class TEXT, definition_versions TEXT NOT NULL, values TEXT NOT NULL, block_document TEXT, created_at TEXT NOT NULL, created_by TEXT NOT NULL, source_version TEXT NOT NULL, host_session_id TEXT)`,
    `CREATE INDEX IF NOT EXISTS field_revisions_entity_created_idx ON field_revisions (entity_type, entity_id, created_at, id)`,
    `CREATE INDEX IF NOT EXISTS field_revisions_parent_idx ON field_revisions (parent_revision_id, kind)`,
    `CREATE TABLE IF NOT EXISTS fields_operation_receipts (operation_id TEXT PRIMARY KEY, kind TEXT NOT NULL, principal_scope TEXT NOT NULL, request_hash TEXT NOT NULL, result TEXT NOT NULL, created_at TEXT NOT NULL)`,
    `CREATE INDEX IF NOT EXISTS fields_operation_receipts_kind_idx ON fields_operation_receipts (kind, created_at)`,
  ])
}

function parseJson(value: unknown, field: string): unknown {
  if (typeof value !== 'string') return value
  try { return JSON.parse(value) as unknown }
  catch { throw new FieldsLifecycleError('integrity', `${field} contains invalid JSON`) }
}
function storage(value: unknown, depth = 0): FieldStorageValue {
  if (depth > 24) throw new FieldsLifecycleError('integrity', 'snapshot exceeds maximum depth')
  if (value === null || typeof value === 'string' || typeof value === 'boolean') return value
  if (typeof value === 'number') {
    if (!Number.isFinite(value)) throw new FieldsLifecycleError('integrity', 'snapshot contains a non-finite number')
    return value
  }
  if (Array.isArray(value)) return Object.freeze(value.map((item) => storage(item, depth + 1)))
  if (typeof value === 'object') {
    const out: Record<string, FieldStorageValue> = {}
    for (const key of Object.keys(value).sort()) {
      if (key === '__proto__' || key === 'prototype' || key === 'constructor') throw new FieldsLifecycleError('integrity', 'snapshot contains a forbidden key')
      const child = (value as Record<string, unknown>)[key]
      if (child !== undefined) out[key] = storage(child, depth + 1)
    }
    return Object.freeze(out)
  }
  throw new FieldsLifecycleError('integrity', `snapshot contains unsupported ${typeof value}`)
}
function record(value: unknown, field: string): Record<string, unknown> {
  const parsed = parseJson(value, field)
  if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) throw new FieldsLifecycleError('integrity', `${field} is malformed`)
  return parsed as Record<string, unknown>
}
function numericRecord(value: unknown, field: string): Readonly<Record<string, number>> {
  const input = record(value, field), out: Record<string, number> = {}
  for (const [key, raw] of Object.entries(input)) {
    const number = Number(raw)
    if (!Number.isSafeInteger(number) || number < 1) throw new FieldsLifecycleError('integrity', `${field}.${key} must be a positive revision`)
    out[key] = number
  }
  return Object.freeze(out)
}
function storedBlock(value: unknown): StoredBlockDocument {
  const input = record(value, 'blockDocument')
  if (input.version !== 1 || !Array.isArray(input.roots)) throw new FieldsLifecycleError('integrity', 'block document is malformed')
  return Object.freeze({ version: 1, roots: Object.freeze(input.roots.map(storage)) })
}
function nodeFromRow(row: NodeRow): FieldValueNodeRow {
  return Object.freeze({
    nodeId: row.nodeId,
    entityType: row.entityType,
    entityId: row.entityId,
    groupKey: row.groupKey,
    definitionRevision: Number(row.definitionRevision),
    fieldKey: row.fieldKey,
    path: row.path,
    parentNodeId: row.parentNodeId,
    rowId: row.rowId,
    layoutKey: row.layoutKey,
    nodeKind: row.nodeKind,
    ordinal: Number(row.ordinal),
    isNull: row.isNull === true || row.isNull === 1,
    valueText: row.valueText,
    valueNumber: row.valueNumber === null ? null : Number(row.valueNumber),
    valueBoolean: row.valueBoolean === null ? null : row.valueBoolean === true || row.valueBoolean === 1,
    valueDateTime: row.valueDateTime,
    valueRef: row.valueRef,
    valueJson: row.valueJson,
  })
}

function recursiveRedact(
  definitions: readonly AnyFieldDefinition[],
  values: FieldValueMap,
  context: FieldsContext,
): Readonly<Record<string, FieldStorageValue>> {
  const registry = context.fieldTypeRegistry ?? createFieldTypeRegistry()
  const out: Record<string, FieldStorageValue> = {}
  for (const definition of definitions) {
    const value = values[definition.key]
    if (value === undefined) continue
    if (definition.type === 'password') {
      out[definition.key] = storage({ configured: Boolean((value as { configured?: unknown })?.configured) })
      continue
    }
    if (definition.type === 'message' || definition.type === 'accordion' || definition.type === 'tab') continue
    if (definition.type === 'group') {
      out[definition.key] = storage(recursiveRedact((definition.settings as { fields: readonly AnyFieldDefinition[] }).fields, (value ?? {}) as FieldValueMap, context))
      continue
    }
    if (definition.type === 'repeater') {
      const fields = (definition.settings as { fields: readonly AnyFieldDefinition[] }).fields
      out[definition.key] = storage((value as readonly { rowId: string; values: FieldValueMap }[]).map((row) => ({ rowId: row.rowId, values: recursiveRedact(fields, row.values, context) })))
      continue
    }
    if (definition.type === 'flexible') {
      const layouts = (definition.settings as { layouts: readonly { key: string; fields: readonly AnyFieldDefinition[] }[] }).layouts
      out[definition.key] = storage((value as readonly { rowId: string; layoutKey: string; values: FieldValueMap }[]).map((row) => {
        const layout = layouts.find((item) => item.key === row.layoutKey)
        if (!layout) throw new FieldsLifecycleError('integrity', 'snapshot references unknown flexible layout')
        return { rowId: row.rowId, layoutKey: row.layoutKey, values: recursiveRedact(layout.fields, row.values, context) }
      }))
      continue
    }
    if (definition.type === 'clone') {
      if (!context.cloneResolver) throw new FieldsLifecycleError('capability-unavailable', 'clone resolver is required to snapshot clone values')
      out[definition.key] = storage(recursiveRedact(context.cloneResolver.resolve(definition), (value ?? {}) as FieldValueMap, context))
      continue
    }
    const redacted = redactFieldValueForRevision(definition, value, registry)
    if (redacted !== 'omit') out[definition.key] = storage(redacted)
  }
  return Object.freeze(out)
}

interface EntitySnapshot {
  readonly version: string
  readonly definitionVersions: Readonly<Record<string, number>>
  readonly values: Readonly<Record<string, FieldStorageValue>>
  readonly blockDocument?: StoredBlockDocument
}
async function snapshotEntity(db: FieldsRevisionReader, ref: EntityRef, context: FieldsContext): Promise<EntitySnapshot> {
  const rows = await executeRows<NodeRow>(db, sql`
    SELECT node_id AS "nodeId", entity_type AS "entityType", entity_id AS "entityId",
      group_key AS "groupKey", definition_revision AS "definitionRevision", field_key AS "fieldKey",
      path, parent_node_id AS "parentNodeId", row_id AS "rowId", layout_key AS "layoutKey",
      node_kind AS "nodeKind", ordinal, is_null AS "isNull", value_text AS "valueText",
      value_number AS "valueNumber", value_boolean AS "valueBoolean", value_date_time AS "valueDateTime",
      value_ref AS "valueRef", value_json AS "valueJson"
    FROM field_value_nodes
    WHERE entity_type = ${ref.entityType} AND entity_id = ${ref.entityId}
    ORDER BY group_key, path
  `)
  const byGroup = new Map<string, NodeRow[]>()
  for (const row of rows) {
    const bucket = byGroup.get(row.groupKey) ?? []
    bucket.push(row)
    byGroup.set(row.groupKey, bucket)
  }
  const definitionVersions: Record<string, number> = {}
  const values: Record<string, FieldStorageValue> = {}
  for (const [groupKey, bucket] of [...byGroup].sort(([left], [right]) => left.localeCompare(right))) {
    const revisions = [...new Set(bucket.map((row) => Number(row.definitionRevision)))]
    if (revisions.length !== 1) throw new FieldsLifecycleError('integrity', `current group ${groupKey} mixes immutable definition revisions`)
    const revision = revisions[0]!
    const version = await resolveFieldDefinitionVersion(db, groupKey, revision)
    const decoded = decodeFieldValueNodes(version.definition, bucket.map(nodeFromRow), {
      definitionRevision: revision,
      ...(context.cloneResolver ? { cloneResolver: { resolve: (definition) => context.cloneResolver!.resolve(definition) } } : {}),
    })
    definitionVersions[groupKey] = revision
    values[groupKey] = storage(recursiveRedact(version.definition.fields, decoded, context))
  }
  const blockRows = await executeRows<BlockDocumentRow>(db, sql`
    SELECT version, document, definition_versions AS "definitionVersions"
    FROM field_block_documents
    WHERE entity_type = ${ref.entityType} AND entity_id = ${ref.entityId}
    LIMIT 1
  `)
  const blockDocument = blockRows[0] ? storedBlock(parseJson(blockRows[0].document, 'blockDocument')) : undefined
  const version = await fieldsRequestHash({ definitionVersions, values, blockDocument: blockDocument ?? null })
  return Object.freeze({
    version,
    definitionVersions: Object.freeze(definitionVersions),
    values: Object.freeze(values),
    ...(blockDocument ? { blockDocument } : {}),
  })
}

function revisionFromRow(row: RevisionRow): FieldsRevision {
  const definitionVersions = numericRecord(row.definitionVersions, 'definitionVersions')
  const valuesObject = record(row.values, 'values')
  const values: Record<string, FieldStorageValue> = {}
  for (const [key, value] of Object.entries(valuesObject)) values[key] = storage(value)
  const blockDocument = row.blockDocument === null ? undefined : storedBlock(parseJson(row.blockDocument, 'blockDocument'))
  return Object.freeze({
    id: row.id,
    ref: Object.freeze({ entityType: row.entityType, entityId: row.entityId }),
    parentRevisionId: row.parentRevisionId,
    ...(row.parentAutosaveId ? { parentAutosaveId: row.parentAutosaveId } : {}),
    kind: row.kind,
    ...(row.retentionClass ? { retentionClass: row.retentionClass } : {}),
    definitionVersions,
    values: Object.freeze(values),
    ...(blockDocument ? { blockDocument } : {}),
    createdAt: row.createdAt instanceof Date ? new Date(row.createdAt) : new Date(row.createdAt),
    createdBy: row.createdBy,
    sourceVersion: row.sourceVersion,
  })
}

async function requireRevision(db: FieldsRevisionReader, ref: EntityRef, id: string): Promise<FieldsRevision> {
  const rows = await executeRows<RevisionRow>(db, sql`
    SELECT id, entity_type AS "entityType", entity_id AS "entityId",
      parent_revision_id AS "parentRevisionId", parent_autosave_id AS "parentAutosaveId",
      kind, retention_class AS "retentionClass", definition_versions AS "definitionVersions",
      values, block_document AS "blockDocument", created_at AS "createdAt",
      created_by AS "createdBy", source_version AS "sourceVersion"
    FROM field_revisions
    WHERE id = ${id} AND entity_type = ${ref.entityType} AND entity_id = ${ref.entityId}
    LIMIT 1
  `)
  const row = rows[0]
  if (!row) throw new FieldsLifecycleError('not-found', 'revision is unavailable')
  return revisionFromRow(row)
}

async function insertRevision(
  tx: FieldsTransaction,
  input: {
    id: string
    ref: EntityRef
    parentRevisionId: string
    parentAutosaveId?: string
    kind: 'revision' | 'autosave'
    retentionClass?: string
    snapshot: EntitySnapshot
    createdBy: string
  },
): Promise<FieldsRevision> {
  const createdAt = new Date()
  await tx.execute(sql`
    INSERT INTO field_revisions (
      id, entity_type, entity_id, parent_revision_id, parent_autosave_id, kind,
      retention_class, definition_versions, values, block_document,
      created_at, created_by, source_version
    ) VALUES (
      ${input.id}, ${input.ref.entityType}, ${input.ref.entityId}, ${input.parentRevisionId},
      ${input.parentAutosaveId ?? null}, ${input.kind}, ${input.retentionClass ?? null},
      ${JSON.stringify(input.snapshot.definitionVersions)}, ${JSON.stringify(input.snapshot.values)},
      ${input.snapshot.blockDocument ? JSON.stringify(input.snapshot.blockDocument) : null},
      ${createdAt.toISOString()}, ${input.createdBy}, ${input.snapshot.version}
    )
  `)
  return Object.freeze({
    id: input.id,
    ref: Object.freeze({ ...input.ref }),
    parentRevisionId: input.parentRevisionId,
    ...(input.parentAutosaveId ? { parentAutosaveId: input.parentAutosaveId } : {}),
    kind: input.kind,
    ...(input.retentionClass ? { retentionClass: input.retentionClass } : {}),
    definitionVersions: input.snapshot.definitionVersions,
    values: input.snapshot.values,
    ...(input.snapshot.blockDocument ? { blockDocument: input.snapshot.blockDocument } : {}),
    createdAt,
    createdBy: input.createdBy,
    sourceVersion: input.snapshot.version,
  })
}

async function capture(
  tx: FieldsTransaction,
  ref: EntityRef,
  mutation: FieldsRevisionMutation | FieldsAutosaveMutation,
  context: FieldsContext,
  kind: 'revision' | 'autosave',
): Promise<FieldsRevision> {
  assertActiveFieldsTransaction(tx, mutation.expectedTransactionIdentity)
  const authorized = await assertFieldsAuthorized(context, 'write', ref)
  if (!mutation.parentRevisionId || mutation.parentRevisionId.length > 256) throw new FieldsLifecycleError('validation', 'parentRevisionId is required', 'parentRevisionId')
  const autosave = kind === 'autosave' ? mutation as FieldsAutosaveMutation : undefined
  if (autosave && (!autosave.parentAutosaveId || autosave.parentAutosaveId.length > 256)) throw new FieldsLifecycleError('validation', 'parentAutosaveId is required', 'parentAutosaveId')
  if (autosave && (!autosave.retentionClass || autosave.retentionClass.length > 128)) throw new FieldsLifecycleError('validation', 'retentionClass is required', 'retentionClass')

  const current = await snapshotEntity(tx, ref, context)
  if (current.version !== mutation.expectedVersion) throw new FieldsLifecycleError('stale-version', 'current fields version changed')
  const requestHash = await fieldsRequestHash({
    ref,
    kind,
    expectedVersion: mutation.expectedVersion,
    parentRevisionId: mutation.parentRevisionId,
    parentAutosaveId: autosave?.parentAutosaveId ?? null,
    retentionClass: autosave?.retentionClass ?? null,
  })
  const existing = await readFieldsReceipt(tx, mutation.operationId)
  if (existing) {
    assertReceiptReplay(existing, { kind: `fields:${kind}`, context: authorized, requestHash })
    const result = record(existing.result, 'receipt')
    if (typeof result.id !== 'string') throw new FieldsLifecycleError('integrity', 'revision receipt is malformed')
    return requireRevision(tx, ref, result.id)
  }

  const revision = await insertRevision(tx, {
    id: crypto.randomUUID(),
    ref,
    parentRevisionId: mutation.parentRevisionId,
    ...(autosave ? { parentAutosaveId: autosave.parentAutosaveId, retentionClass: autosave.retentionClass } : {}),
    kind,
    snapshot: current,
    createdBy: authorized.principal.id,
  })
  await writeFieldsReceipt(tx, {
    operationId: mutation.operationId,
    kind: `fields:${kind}`,
    context: authorized,
    requestHash,
    result: storage({ id: revision.id }),
  })
  return revision
}

export async function captureFieldsRevision(tx: FieldsTransaction, ref: EntityRef, mutation: FieldsRevisionMutation, context: FieldsContext): Promise<FieldsRevision> {
  return capture(tx, ref, mutation, context, 'revision')
}
export async function createFieldsAutosave(tx: FieldsTransaction, ref: EntityRef, mutation: FieldsAutosaveMutation, context: FieldsContext): Promise<FieldsRevision> {
  return capture(tx, ref, mutation, context, 'autosave')
}
export async function getFieldsRevision(db: Querier<FieldsSchema>, ref: EntityRef, revisionId: string, context: FieldsContext): Promise<FieldsRevision | null> {
  await assertFieldsAuthorized(context, 'read', ref)
  try { return await requireRevision(db, ref, revisionId) }
  catch (error) {
    if (error instanceof FieldsLifecycleError && error.code === 'not-found') return null
    throw error
  }
}
export async function listFieldsRevisions(
  db: Querier<FieldsSchema>,
  ref: EntityRef,
  input: { cursor?: FieldsRevisionCursor; limit: number; kind?: 'revision' | 'autosave' },
  context: FieldsContext,
): Promise<FieldsRevisionPage> {
  await assertFieldsAuthorized(context, 'read', ref)
  const limit = Math.min(Math.max(Number.isSafeInteger(input.limit) ? input.limit : 20, 1), 100)
  if (input.cursor && (!(input.cursor.createdAt instanceof Date) || Number.isNaN(input.cursor.createdAt.getTime()) || !input.cursor.id)) throw new FieldsLifecycleError('validation', 'cursor is invalid', 'cursor')
  const kindClause = input.kind ? sql`AND kind = ${input.kind}` : sql``
  const cursorClause = input.cursor
    ? sql`AND (created_at < ${input.cursor.createdAt.toISOString()} OR (created_at = ${input.cursor.createdAt.toISOString()} AND id < ${input.cursor.id}))`
    : sql``
  const rows = await executeRows<RevisionRow>(db, sql`
    SELECT id, entity_type AS "entityType", entity_id AS "entityId",
      parent_revision_id AS "parentRevisionId", parent_autosave_id AS "parentAutosaveId",
      kind, retention_class AS "retentionClass", definition_versions AS "definitionVersions",
      values, block_document AS "blockDocument", created_at AS "createdAt",
      created_by AS "createdBy", source_version AS "sourceVersion"
    FROM field_revisions
    WHERE entity_type = ${ref.entityType} AND entity_id = ${ref.entityId}
      ${kindClause} ${cursorClause}
    ORDER BY created_at DESC, id DESC
    LIMIT ${limit + 1}
  `)
  const items = rows.slice(0, limit).map(revisionFromRow)
  const last = items.at(-1)
  return Object.freeze({
    items: Object.freeze(items),
    nextCursor: rows.length > limit && last ? Object.freeze({ createdAt: last.createdAt, id: last.id }) : null,
  })
}

function flatten(value: unknown, path: (string | number)[] = [], out = new Map<string, string>()): Map<string, string> {
  if (value === null || typeof value !== 'object') { out.set(JSON.stringify(path), JSON.stringify(value)); return out }
  if (Array.isArray(value)) { value.forEach((item, index) => flatten(item, [...path, index], out)); return out }
  for (const key of Object.keys(value as Record<string, unknown>).sort()) flatten((value as Record<string, unknown>)[key], [...path, key], out)
  return out
}
function blockNodeState(document: StoredBlockDocument | undefined): Map<string, { path: string; hash: string }> {
  const out = new Map<string, { path: string; hash: string }>()
  const walk = (node: unknown, path: string) => {
    if (!node || typeof node !== 'object' || Array.isArray(node)) return
    const value = node as Record<string, unknown>
    if (typeof value.id === 'string') out.set(value.id, { path, hash: JSON.stringify(storage(value)) })
    if (Array.isArray(value.children)) value.children.forEach((child, index) => walk(child, `${path}/${index}`))
  }
  document?.roots.forEach((node, index) => walk(node, String(index)))
  return out
}
export async function diffFieldsRevisions(db: Querier<FieldsSchema>, ref: EntityRef, leftRevisionId: string, rightRevisionId: string, context: FieldsContext): Promise<FieldsRevisionDiff> {
  await assertFieldsAuthorized(context, 'read', ref)
  const [left, right] = await Promise.all([requireRevision(db, ref, leftRevisionId), requireRevision(db, ref, rightRevisionId)])
  const before = flatten(left.values), after = flatten(right.values)
  const valuePaths: { path: (string | number)[]; kind: 'added' | 'removed' | 'changed' }[] = []
  for (const key of [...new Set([...before.keys(), ...after.keys()])].sort()) {
    const leftValue = before.get(key), rightValue = after.get(key)
    if (leftValue === rightValue) continue
    valuePaths.push({ path: JSON.parse(key) as (string | number)[], kind: leftValue === undefined ? 'added' : rightValue === undefined ? 'removed' : 'changed' })
  }
  const leftBlocks = blockNodeState(left.blockDocument), rightBlocks = blockNodeState(right.blockDocument)
  const blockNodes: { nodeId: string; kind: 'added' | 'removed' | 'moved' | 'changed' }[] = []
  for (const id of [...new Set([...leftBlocks.keys(), ...rightBlocks.keys()])].sort()) {
    const leftNode = leftBlocks.get(id), rightNode = rightBlocks.get(id)
    if (!leftNode) blockNodes.push({ nodeId: id, kind: 'added' })
    else if (!rightNode) blockNodes.push({ nodeId: id, kind: 'removed' })
    else if (leftNode.path !== rightNode.path) blockNodes.push({ nodeId: id, kind: 'moved' })
    else if (leftNode.hash !== rightNode.hash) blockNodes.push({ nodeId: id, kind: 'changed' })
  }
  return Object.freeze({ valuePaths: Object.freeze(valuePaths), blockNodes: Object.freeze(blockNodes) })
}

function preservePasswordValues(
  definitions: readonly AnyFieldDefinition[],
  revision: Record<string, unknown>,
  current: Record<string, unknown>,
  context: FieldsContext,
): Record<string, unknown> {
  const out: Record<string, unknown> = { ...revision }
  for (const definition of definitions) {
    if (definition.type === 'password') {
      if (Object.prototype.hasOwnProperty.call(current, definition.key)) out[definition.key] = current[definition.key]
      else delete out[definition.key]
      continue
    }
    const revisionValue = revision[definition.key], currentValue = current[definition.key]
    if (definition.type === 'group' && revisionValue && typeof revisionValue === 'object' && !Array.isArray(revisionValue)) {
      out[definition.key] = preservePasswordValues((definition.settings as { fields: readonly AnyFieldDefinition[] }).fields, revisionValue as Record<string, unknown>, (currentValue ?? {}) as Record<string, unknown>, context)
      continue
    }
    if (definition.type === 'repeater' && Array.isArray(revisionValue)) {
      const fields = (definition.settings as { fields: readonly AnyFieldDefinition[] }).fields
      const currentRows = Array.isArray(currentValue) ? currentValue as Array<{ rowId?: string; values?: Record<string, unknown> }> : []
      out[definition.key] = revisionValue.map((raw) => {
        const row = raw as { rowId: string; values: Record<string, unknown> }
        const currentRow = currentRows.find((candidate) => candidate.rowId === row.rowId)
        return { ...row, values: preservePasswordValues(fields, row.values, currentRow?.values ?? {}, context) }
      })
      continue
    }
    if (definition.type === 'flexible' && Array.isArray(revisionValue)) {
      const layouts = (definition.settings as { layouts: readonly { key: string; fields: readonly AnyFieldDefinition[] }[] }).layouts
      const currentRows = Array.isArray(currentValue) ? currentValue as Array<{ rowId?: string; values?: Record<string, unknown> }> : []
      out[definition.key] = revisionValue.map((raw) => {
        const row = raw as { rowId: string; layoutKey: string; values: Record<string, unknown> }
        const layout = layouts.find((candidate) => candidate.key === row.layoutKey)
        if (!layout) throw new FieldsLifecycleError('integrity', 'revision references unknown flexible layout')
        const currentRow = currentRows.find((candidate) => candidate.rowId === row.rowId)
        return { ...row, values: preservePasswordValues(layout.fields, row.values, currentRow?.values ?? {}, context) }
      })
      continue
    }
    if (definition.type === 'clone' && revisionValue && typeof revisionValue === 'object' && !Array.isArray(revisionValue)) {
      if (!context.cloneResolver) throw new FieldsLifecycleError('capability-unavailable', 'clone resolver is required to restore clone values')
      out[definition.key] = preservePasswordValues(context.cloneResolver.resolve(definition), revisionValue as Record<string, unknown>, (currentValue ?? {}) as Record<string, unknown>, context)
    }
  }
  return out
}

async function writeValueNodes(tx: FieldsTransaction, ref: EntityRef, rows: readonly FieldValueNodeRow[]): Promise<void> {
  await tx.execute(sql`DELETE FROM field_value_nodes WHERE entity_type = ${ref.entityType} AND entity_id = ${ref.entityId}`)
  for (const row of rows) {
    await tx.execute(sql`
      INSERT INTO field_value_nodes (
        node_id, entity_type, entity_id, group_key, definition_revision, field_key, path,
        parent_node_id, row_id, layout_key, node_kind, ordinal, is_null,
        value_text, value_number, value_boolean, value_date_time, value_ref, value_json,
        created_at, updated_at
      ) VALUES (
        ${row.nodeId}, ${row.entityType}, ${row.entityId}, ${row.groupKey}, ${row.definitionRevision},
        ${row.fieldKey}, ${row.path}, ${row.parentNodeId}, ${row.rowId}, ${row.layoutKey}, ${row.nodeKind},
        ${row.ordinal}, ${row.isNull}, ${row.valueText}, ${row.valueNumber}, ${row.valueBoolean},
        ${row.valueDateTime}, ${row.valueRef}, ${row.valueJson}, ${new Date().toISOString()}, ${new Date().toISOString()}
      )
    `)
  }
}

async function writeBlockSnapshot(
  tx: FieldsTransaction,
  ref: EntityRef,
  document: StoredBlockDocument,
  definitionVersions: Readonly<Record<string, number>>,
  parentRevisionId: string,
  principalId: string,
): Promise<void> {
  const current = await tx.execute<{ version: number } & Record<string, unknown>>(sql`
    SELECT version FROM field_block_documents
    WHERE entity_type = ${ref.entityType} AND entity_id = ${ref.entityId}
    LIMIT 1
  `)
  const version = Number(current[0]?.version ?? 0) + 1
  await tx.execute(sql`
    INSERT INTO field_block_documents (
      entity_type, entity_id, version, document, definition_versions,
      parent_revision_id, updated_by, updated_at
    ) VALUES (
      ${ref.entityType}, ${ref.entityId}, ${version}, ${JSON.stringify(document)},
      ${JSON.stringify(definitionVersions)}, ${parentRevisionId}, ${principalId}, ${new Date().toISOString()}
    )
    ON CONFLICT (entity_type, entity_id) DO UPDATE SET
      version = EXCLUDED.version,
      document = EXCLUDED.document,
      definition_versions = EXCLUDED.definition_versions,
      parent_revision_id = EXCLUDED.parent_revision_id,
      updated_by = EXCLUDED.updated_by,
      updated_at = EXCLUDED.updated_at
  `)
}

export async function restoreFieldsRevision(
  tx: FieldsTransaction,
  ref: EntityRef,
  revisionId: string,
  mutation: FieldsRevisionMutation,
  context: FieldsContext,
): Promise<{ version: string; revision: FieldsRevision }> {
  assertActiveFieldsTransaction(tx, mutation.expectedTransactionIdentity)
  const authorized = await assertFieldsAuthorized(context, 'write', ref)
  const source = await requireRevision(tx, ref, revisionId)
  const current = await snapshotEntity(tx, ref, context)
  if (current.version !== mutation.expectedVersion) throw new FieldsLifecycleError('stale-version', 'current fields version changed')
  const requestHash = await fieldsRequestHash({ ref, revisionId, expectedVersion: mutation.expectedVersion, parentRevisionId: mutation.parentRevisionId })
  const existing = await readFieldsReceipt(tx, mutation.operationId)
  if (existing) {
    assertReceiptReplay(existing, { kind: 'fields:restore-revision', context: authorized, requestHash })
    const result = record(existing.result, 'restore receipt')
    if (typeof result.version !== 'string' || typeof result.revisionId !== 'string') throw new FieldsLifecycleError('integrity', 'restore receipt is malformed')
    return Object.freeze({ version: result.version, revision: await requireRevision(tx, ref, result.revisionId) })
  }

  const nodes: FieldValueNodeRow[] = []
  for (const [groupKey, definitionRevision] of Object.entries(source.definitionVersions).sort(([left], [right]) => left.localeCompare(right))) {
    const definition = await resolveFieldDefinitionVersion(tx, groupKey, definitionRevision)
    const revisionValues = record(source.values[groupKey] ?? {}, `revision.values.${groupKey}`)
    const currentValues = record(current.values[groupKey] ?? {}, `current.values.${groupKey}`)
    const restoredValues = preservePasswordValues(definition.definition.fields, revisionValues, currentValues, context)
    const ids = new Map<string, string>()
    const nodeIdForPath = (path: string) => {
      let id = ids.get(path)
      if (!id) { id = crypto.randomUUID(); ids.set(path, id) }
      return id
    }
    nodes.push(...encodeFieldValueNodes(definition.definition, restoredValues, {
      entityType: ref.entityType,
      entityId: ref.entityId,
      definitionRevision,
      nodeIdForPath,
      ...(context.cloneResolver ? { cloneResolver: { resolve: (field) => context.cloneResolver!.resolve(field) } } : {}),
    }))
  }
  await writeValueNodes(tx, ref, nodes)
  if (source.blockDocument) await writeBlockSnapshot(tx, ref, source.blockDocument, source.definitionVersions, mutation.parentRevisionId, authorized.principal.id)

  const restored = await snapshotEntity(tx, ref, context)
  const restoreRevision = await insertRevision(tx, {
    id: crypto.randomUUID(),
    ref,
    parentRevisionId: mutation.parentRevisionId,
    kind: 'revision',
    snapshot: restored,
    createdBy: authorized.principal.id,
  })
  await writeFieldsReceipt(tx, {
    operationId: mutation.operationId,
    kind: 'fields:restore-revision',
    context: authorized,
    requestHash,
    result: storage({ version: restored.version, revisionId: restoreRevision.id }),
  })
  return Object.freeze({ version: restored.version, revision: restoreRevision })
}

interface AutosaveRow extends Record<string, unknown> {
  id: string
  parentRevisionId: string
  parentAutosaveId: string | null
  retentionClass: string | null
  createdAt: Date | string
}
export async function pruneFieldsAutosaves(
  tx: FieldsTransaction,
  ref: EntityRef,
  policy: FieldsAutosaveRetentionPolicy,
  operationId: string,
  context: FieldsContext,
): Promise<{ deleted: number }> {
  assertActiveFieldsTransaction(tx)
  const authorized = await assertFieldsAuthorized(context, 'delete', ref)
  if (!Number.isSafeInteger(policy.maxPerParent) || policy.maxPerParent < 0 || policy.maxPerParent > 500 || !Number.isSafeInteger(policy.maxAgeMs) || policy.maxAgeMs < 0) throw new FieldsLifecycleError('validation', 'autosave retention policy is invalid', 'policy')
  for (const [key, value] of Object.entries(policy.classes)) {
    if (!key || key.length > 128 || !Number.isSafeInteger(value.maxPerParent) || value.maxPerParent < 0 || value.maxPerParent > 500 || !Number.isSafeInteger(value.maxAgeMs) || value.maxAgeMs < 0) throw new FieldsLifecycleError('validation', 'autosave retention class is invalid', `policy.classes.${key}`)
  }
  const requestHash = await fieldsRequestHash({ ref, policy })
  const replay = await readFieldsReceipt(tx, operationId)
  if (replay) {
    assertReceiptReplay(replay, { kind: 'fields:prune-autosaves', context: authorized, requestHash })
    const result = record(replay.result, 'prune receipt')
    return Object.freeze({ deleted: Number(result.deleted ?? 0) })
  }

  const rows = await tx.execute<AutosaveRow>(sql`
    SELECT id, parent_revision_id AS "parentRevisionId", parent_autosave_id AS "parentAutosaveId",
      retention_class AS "retentionClass", created_at AS "createdAt"
    FROM field_revisions
    WHERE entity_type = ${ref.entityType} AND entity_id = ${ref.entityId} AND kind = 'autosave'
    ORDER BY created_at DESC, id DESC
  `)
  const referenced = new Set(rows.map((row) => row.parentAutosaveId).filter((id): id is string => Boolean(id)))
  const now = Date.now(), perParent = new Map<string, number>(), deleteIds: string[] = []
  for (const row of rows) {
    const classPolicy = row.retentionClass ? policy.classes[row.retentionClass] : undefined
    const maxPerParent = classPolicy?.maxPerParent ?? policy.maxPerParent
    const maxAgeMs = classPolicy?.maxAgeMs ?? policy.maxAgeMs
    const count = perParent.get(row.parentRevisionId) ?? 0
    perParent.set(row.parentRevisionId, count + 1)
    const createdAt = row.createdAt instanceof Date ? row.createdAt : new Date(row.createdAt)
    if (Number.isNaN(createdAt.getTime())) throw new FieldsLifecycleError('integrity', 'autosave timestamp is invalid')
    if (!referenced.has(row.id) && (count >= maxPerParent || now - createdAt.getTime() > maxAgeMs)) deleteIds.push(row.id)
  }
  for (const id of deleteIds) await tx.execute(sql`DELETE FROM field_revisions WHERE id = ${id} AND kind = 'autosave'`)
  await writeFieldsReceipt(tx, {
    operationId,
    kind: 'fields:prune-autosaves',
    context: authorized,
    requestHash,
    result: storage({ deleted: deleteIds.length }),
  })
  return Object.freeze({ deleted: deleteIds.length })
}