import { sql, type SQL } from 'drizzle-orm'
import { canonicalContentMigrationHash } from './migrations/content-model.js'
import type { ContentSchemaValue } from './schema.js'
import type { ContentPrincipal } from './authz.js'
export type { ContentPrincipal } from './authz.js'
import {
  ContentDefinitionError,
  __contentRegistryInternals,
  defineContentType,
  normalizeContentTypeDefinition,
  principalScope,
  readDefinitionReceipt,
  resolveContentType,
  writeDefinitionReceipt,
  writeDefinitionVersion,
  type CodeContentType,
  type ContentDefinitionExecutor,
  type ContentFeature,
  type ContentTypeDefinition,
  type ContentTypePatch,
  type ResolvedContentType,
} from './registry.js'
import {
  __contentStatusInternals,
  defineContentStatus,
  normalizeContentStatusDefinition,
  resolveContentStatus,
  type CodeContentStatus,
  type ContentStatusChangeStrategy,
  type ContentStatusDefinition,
  type ContentStatusPatch,
  type ResolvedContentStatus,
} from './status.js'

export type ContentDefinitionAction =
  | 'createType'
  | 'createStatus'
  | 'changeType'
  | 'changeStatus'
  | 'reconcileCode'
  | 'restoreBackup'

export interface ContentDefinitionAuthorization {
  assert(
    tx: ContentDefinitionExecutor,
    principal: ContentPrincipal,
    action: ContentDefinitionAction,
    definitionKey?: string,
  ): Promise<{ policyVersion: string }>
}

export interface ContentDefinitionImpact {
  token: string
  definitionKey: string
  expectedVersion: number
  authorizationPolicyVersion: string
  corpusVersion: string
  corpusHash: string
  affectedCounts: Readonly<Record<string, number>>
  incompatibilities: readonly string[]
}

export type ContentTypeChangeStrategy =
  | { kind: 'reject' }
  | { kind: 'deactivate' }
  | { kind: 'mapStatus'; mappings: Readonly<Record<string, string>> }
  | { kind: 'detachTaxonomies'; taxonomyKeys: readonly string[] }
  | { kind: 'flattenHierarchy' }
  | { kind: 'retainDormantFeatureData'; features: readonly ContentFeature[] }
  | { kind: 'purgeFeatureData'; features: readonly ContentFeature[]; backupId: string }
  | { kind: 'reassignType'; targetTypeKey: string }

export type ContentDefinitionEvent =
  | {
      id: string
      version: 1
      kind: 'codeRegistryReconciled'
      operationId: string
      principalId: string
      fromVersion: string
      toVersion: string
      changedKeys: readonly string[]
      strategiesHash: string
      authorizationPolicyVersion: string
      corpusHashes: readonly string[]
      affectedCounts: Readonly<Record<string, number>>
      backupIds: readonly string[]
    }
  | {
      id: string
      version: 1
      kind: 'dbDefinitionCreated'
      operationId: string
      principalId: string
      definitionKind: 'type' | 'status'
      definitionKey: string
      definitionHash: string
      revision: number
      authorizationPolicyVersion: string
    }
  | {
      id: string
      version: 1
      kind: 'dbDefinitionChanged'
      operationId: string
      principalId: string
      definitionKind: 'type' | 'status'
      definitionKey: string
      action: 'update' | 'deactivate' | 'delete'
      strategyHash: string
      impactToken: string
      authorizationPolicyVersion: string
      corpusVersion: string
      corpusHash: string
      affectedCounts: Readonly<Record<string, number>>
      backupIds: readonly string[]
    }
  | {
      id: string
      version: 1
      kind: 'backupRestored'
      operationId: string
      principalId: string
      manifestId: string
      destinationCorpusHash: string
      restoredCounts: Readonly<Record<string, number>>
      authorizationPolicyVersion: string
    }

export interface ContentDefinitionAudit {
  record(tx: ContentDefinitionExecutor, event: ContentDefinitionEvent): Promise<void>
}
export interface ContentDefinitionOutbox {
  enqueue(tx: ContentDefinitionExecutor, event: ContentDefinitionEvent): Promise<void>
}
export interface ContentDefinitionEffects {
  audit: ContentDefinitionAudit
  outbox: ContentDefinitionOutbox
}

export interface ContentBackupManifest {
  id: string
  scopeHash: string
  corpusVersion: string
  corpusHash: string
  itemCounts: Readonly<Record<string, number>>
  byteCount: number
  immutable: true
}
export interface ContentBackupSnapshot {
  readonly manifest: ContentBackupManifest
  readonly payload: ContentSchemaValue
}
export interface ContentBackupStore {
  verify(
    tx: ContentDefinitionExecutor,
    backupId: string,
    scopeHash: string,
    corpusVersion: string,
    corpusHash: string,
    expectedCounts: Readonly<Record<string, number>>,
  ): Promise<ContentBackupManifest>
  read(tx: ContentDefinitionExecutor, manifest: ContentBackupManifest): Promise<ContentBackupSnapshot>
}
export interface ContentDestructiveDependencies { backups: ContentBackupStore }
export interface ContentBackupRestoreResult {
  operationId: string
  manifestId: string
  destinationCorpusHash: string
  restoredCounts: Readonly<Record<string, number>>
}

interface CorpusRow extends Record<string, unknown> {
  id: string
  status: string
  parentId: string | null
  menuOrder: number | string
  templateKey: string | null
  excerpt: string
  featuredMedia: unknown
  commentStatus: string
  pingStatus: string
  sticky: boolean | number
  format: string | null
  deletedAt: string | Date | null
  updatedAt: string | Date
}

interface StoredImpactPlan {
  readonly definitionKind: 'type' | 'status'
  readonly key: string
  readonly expectedVersion: number
  readonly policyVersion: string
  readonly corpusVersion: string
  readonly corpusHash: string
  readonly affectedCounts: Readonly<Record<string, number>>
  readonly incompatibilities: readonly string[]
  readonly request: ContentSchemaValue
  readonly strategyHash: string
}

function asSchemaValue(value: unknown): ContentSchemaValue {
  return JSON.parse(JSON.stringify(value)) as ContentSchemaValue
}

function parsePayload<T>(value: unknown): T {
  return (typeof value === 'string' ? JSON.parse(value) : value) as T
}

function countValue(value: unknown): number {
  if (typeof value === 'number') return value
  if (typeof value === 'bigint') return Number(value)
  if (typeof value === 'string' && /^\d+$/.test(value)) return Number(value)
  throw new ContentDefinitionError('invalid-receipt', 'adapter returned a non-numeric count')
}

async function typeCorpus(db: ContentDefinitionExecutor, key: string): Promise<readonly CorpusRow[]> {
  return db.execute<CorpusRow>(sql`SELECT id, status, parent_id AS "parentId", menu_order AS "menuOrder",
    template_key AS "templateKey", excerpt, featured_media AS "featuredMedia", comment_status AS "commentStatus",
    ping_status AS "pingStatus", sticky, format, deleted_at AS "deletedAt", updated_at AS "updatedAt"
    FROM content_entries WHERE type = ${key} ORDER BY id`)
}

async function statusCorpus(db: ContentDefinitionExecutor, key: string): Promise<readonly { id: string; type: string; updatedAt: string | Date }[]> {
  return db.execute<{ id: string; type: string; updatedAt: string | Date }>(sql`SELECT id, type, updated_at AS "updatedAt" FROM content_entries WHERE status = ${key} ORDER BY id`)
}

async function corpusEvidence(rows: readonly Record<string, unknown>[]): Promise<{ version: string; hash: string }> {
  const canonical = rows.map((row) => Object.fromEntries(Object.entries(row).map(([key, value]) => [key, value instanceof Date ? value.toISOString() : value])))
  const hash = await canonicalContentMigrationHash(canonical)
  return { version: `sha256:${hash}`, hash }
}

function removed<T>(before: readonly T[] | undefined, after: readonly T[] | undefined): T[] {
  const next = new Set(after ?? [])
  return (before ?? []).filter((item) => !next.has(item))
}

function validateStrategies(strategies: readonly ContentTypeChangeStrategy[]): void {
  if (strategies.length === 0) throw new ContentDefinitionError('strategy-conflict', 'at least one strategy is required')
  if (strategies.some((strategy) => strategy.kind === 'reject') && strategies.length !== 1) {
    throw new ContentDefinitionError('strategy-conflict', 'reject must be the sole strategy')
  }
  const singular = new Set<string>()
  for (const strategy of strategies) {
    const category = strategy.kind === 'mapStatus' ? 'status'
      : strategy.kind === 'flattenHierarchy' ? 'hierarchy'
        : strategy.kind === 'reassignType' || strategy.kind === 'deactivate' ? 'lifecycle'
          : undefined
    if (category && singular.has(category)) throw new ContentDefinitionError('strategy-conflict', `multiple ${category} strategies are not allowed`)
    if (category) singular.add(category)
  }
}

function typeImpactCounts(current: ResolvedContentType, next: ContentTypeDefinition, rows: readonly CorpusRow[]): { counts: Record<string, number>; incompatibilities: string[] } {
  const counts: Record<string, number> = { entries: rows.length }
  const incompatibilities: string[] = []
  for (const status of removed(current.statusKeys, next.statusKeys)) {
    const count = rows.filter((row) => row.status === status).length
    counts[`status:${status}`] = count
    if (count > 0) incompatibilities.push(`status:${status}`)
  }
  if (current.hierarchical && !next.hierarchical) {
    const count = rows.filter((row) => row.parentId !== null).length
    counts.hierarchy = count
    if (count > 0) incompatibilities.push('hierarchy')
  }
  for (const feature of removed(current.supports, next.supports)) {
    let count = 0
    if (feature === 'excerpt') count = rows.filter((row) => row.excerpt.length > 0).length
    else if (feature === 'featuredImage') count = rows.filter((row) => row.featuredMedia !== null).length
    else if (feature === 'pageAttributes') count = rows.filter((row) => row.parentId !== null || Number(row.menuOrder) !== 0 || row.templateKey !== null).length
    else if (feature === 'postFormats') count = rows.filter((row) => row.format !== null).length
    else count = rows.length
    counts[`feature:${feature}`] = count
    if (count > 0) incompatibilities.push(`feature:${feature}`)
  }
  const routeNarrowed = (current.publiclyQueryable && !next.publiclyQueryable)
    || (current.rest !== false && next.rest === false)
    || (current.hasArchive !== false && next.hasArchive === false)
    || (current.rewrite !== false && next.rewrite === false)
  if (routeNarrowed && rows.length > 0) {
    counts.routing = rows.length
    incompatibilities.push('routing')
  }
  return { counts, incompatibilities }
}

async function storeImpact(db: ContentDefinitionExecutor, plan: StoredImpactPlan): Promise<string> {
  const token = await canonicalContentMigrationHash(plan)
  const operationId = `impact:${token}`
  const existing = await db.execute<{ payload: unknown }>(sql`SELECT payload FROM content_lifecycle_journal WHERE operation_id = ${operationId}`)
  if (existing.length === 0) {
    const now = new Date().toISOString()
    await db.execute(sql`INSERT INTO content_lifecycle_journal (id, operation_id, kind, definition_kind, definition_key, definition_revision, state, payload, created_at, updated_at)
      VALUES (${crypto.randomUUID()}, ${operationId}, 'definition-impact', ${plan.definitionKind}, ${plan.key}, ${plan.expectedVersion}, 'prepared', ${JSON.stringify(plan)}, ${now}, ${now})`)
  }
  return token
}

async function readImpact(db: ContentDefinitionExecutor, token: string): Promise<StoredImpactPlan> {
  const rows = await db.execute<{ payload: unknown }>(sql`SELECT payload FROM content_lifecycle_journal WHERE operation_id = ${`impact:${token}`} AND kind = 'definition-impact'`)
  if (!rows[0]) throw new ContentDefinitionError('stale-impact', 'impact token is unknown or expired')
  const plan = parsePayload<StoredImpactPlan>(rows[0].payload)
  if (await canonicalContentMigrationHash(plan) !== token) throw new ContentDefinitionError('stale-impact', 'impact token bytes do not match stored plan')
  return plan
}

function strategyHash(strategies: unknown): Promise<string> {
  return canonicalContentMigrationHash(strategies)
}

export async function previewContentTypeChange(
  db: ContentDefinitionExecutor,
  principal: ContentPrincipal,
  input: {
    key: string
    expectedVersion: number
    patch?: ContentTypePatch
    deactivate?: boolean
    delete?: boolean
    strategies: readonly ContentTypeChangeStrategy[]
  },
  authorization: ContentDefinitionAuthorization,
  opts: { codeTypes?: readonly CodeContentType[] } = {},
): Promise<ContentDefinitionImpact> {
  validateStrategies(input.strategies)
  const { policyVersion } = await authorization.assert(db, principal, 'changeType', input.key)
  const current = await resolveContentType(db, input.key, opts)
  if (current.origin === 'code') throw new ContentDefinitionError('code-definition-owned', 'code definitions can only change through reconciliation')
  if (current.version !== input.expectedVersion) throw new ContentDefinitionError('stale-definition', 'expected version does not match current definition')
  const next = normalizeContentTypeDefinition({ ...current, ...(input.patch ?? {}), key: current.key, active: input.deactivate || input.delete ? false : current.active })
  const rows = await typeCorpus(db, input.key)
  const evidence = await corpusEvidence(rows as unknown as readonly Record<string, unknown>[])
  const impact = typeImpactCounts(current, next, rows)
  if ((input.deactivate || input.delete) && rows.length > 0) impact.incompatibilities.push(input.delete ? 'delete-with-entries' : 'deactivate-with-entries')
  const plan: StoredImpactPlan = {
    definitionKind: 'type', key: input.key, expectedVersion: input.expectedVersion, policyVersion,
    corpusVersion: evidence.version, corpusHash: evidence.hash,
    affectedCounts: Object.freeze({ ...impact.counts }), incompatibilities: Object.freeze([...new Set(impact.incompatibilities)].sort()),
    request: asSchemaValue({ patch: input.patch ?? null, deactivate: Boolean(input.deactivate), delete: Boolean(input.delete), strategies: input.strategies }),
    strategyHash: await strategyHash(input.strategies),
  }
  const token = await storeImpact(db, plan)
  return Object.freeze({
    token, definitionKey: input.key, expectedVersion: input.expectedVersion,
    authorizationPolicyVersion: policyVersion, corpusVersion: evidence.version, corpusHash: evidence.hash,
    affectedCounts: plan.affectedCounts, incompatibilities: plan.incompatibilities,
  })
}

function requestRecord(plan: StoredImpactPlan): {
  patch: ContentTypePatch | null
  deactivate: boolean
  delete: boolean
  strategies: readonly ContentTypeChangeStrategy[]
} {
  return plan.request as unknown as {
    patch: ContentTypePatch | null
    deactivate: boolean
    delete: boolean
    strategies: readonly ContentTypeChangeStrategy[]
  }
}

async function verifyBackups(
  db: ContentDefinitionExecutor,
  plan: StoredImpactPlan,
  strategies: readonly ContentTypeChangeStrategy[],
  destructive?: ContentDestructiveDependencies,
): Promise<readonly string[]> {
  const purges = strategies.filter((strategy): strategy is Extract<ContentTypeChangeStrategy, { kind: 'purgeFeatureData' }> => strategy.kind === 'purgeFeatureData')
  if (purges.length === 0) return Object.freeze([])
  if (!destructive) throw new ContentDefinitionError('backup-required', 'purge requires destructive backup dependencies')
  const ids: string[] = []
  for (const purge of purges) {
    const expectedCounts = Object.fromEntries(purge.features.map((feature) => [`feature:${feature}`, plan.affectedCounts[`feature:${feature}`] ?? 0]))
    const scopeHash = await canonicalContentMigrationHash({ definitionKey: plan.key, features: [...purge.features].sort(), policyVersion: plan.policyVersion, corpusVersion: plan.corpusVersion, corpusHash: plan.corpusHash })
    const manifest = await destructive.backups.verify(db, purge.backupId, scopeHash, plan.corpusVersion, plan.corpusHash, expectedCounts)
    if (!manifest.immutable || manifest.id !== purge.backupId || manifest.scopeHash !== scopeHash || manifest.corpusHash !== plan.corpusHash) {
      throw new ContentDefinitionError('backup-mismatch', 'backup manifest does not match the bound destructive scope')
    }
    ids.push(manifest.id)
  }
  return Object.freeze(ids.sort())
}

async function applyTypeStrategies(
  db: ContentDefinitionExecutor,
  key: string,
  strategies: readonly ContentTypeChangeStrategy[],
  opts: { codeTypes?: readonly CodeContentType[]; codeStatuses?: readonly CodeContentStatus[] },
): Promise<void> {
  for (const strategy of strategies) {
    if (strategy.kind === 'reject' || strategy.kind === 'deactivate' || strategy.kind === 'retainDormantFeatureData') continue
    if (strategy.kind === 'mapStatus') {
      for (const [from, to] of Object.entries(strategy.mappings).sort(([a], [b]) => a.localeCompare(b))) {
        const target = await resolveContentStatus(db, to, { codeStatuses: opts.codeStatuses })
        if (!target.active) throw new ContentDefinitionError('strategy-conflict', `target status ${to} is inactive`)
        await db.execute(sql`UPDATE content_entries SET status = ${to}, status_definition_revision = ${target.revision} WHERE type = ${key} AND status = ${from}`)
      }
    } else if (strategy.kind === 'flattenHierarchy') {
      await db.execute(sql`UPDATE content_entries SET parent_id = NULL WHERE type = ${key} AND parent_id IS NOT NULL`)
    } else if (strategy.kind === 'detachTaxonomies') {
      for (const taxonomy of [...new Set(strategy.taxonomyKeys)].sort()) {
        await db.execute(sql`DELETE FROM content_entry_terms WHERE entry_id IN (SELECT id FROM content_entries WHERE type = ${key}) AND term_id IN (SELECT id FROM content_terms WHERE taxonomy = ${taxonomy})`)
      }
    } else if (strategy.kind === 'purgeFeatureData') {
      for (const feature of strategy.features) {
        if (feature === 'excerpt') await db.execute(sql`UPDATE content_entries SET excerpt = '' WHERE type = ${key}`)
        else if (feature === 'featuredImage') await db.execute(sql`UPDATE content_entries SET featured_media = NULL WHERE type = ${key}`)
        else if (feature === 'pageAttributes') await db.execute(sql`UPDATE content_entries SET parent_id = NULL, menu_order = 0, template_key = NULL WHERE type = ${key}`)
        else if (feature === 'postFormats') await db.execute(sql`UPDATE content_entries SET format = NULL WHERE type = ${key}`)
      }
    } else if (strategy.kind === 'reassignType') {
      const target = await resolveContentType(db, strategy.targetTypeKey, { codeTypes: opts.codeTypes })
      if (!target.active) throw new ContentDefinitionError('strategy-conflict', 'target type is inactive')
      await db.execute(sql`UPDATE content_entries SET type = ${target.key}, type_definition_revision = ${target.revision} WHERE type = ${key}`)
    }
  }
}

export async function applyContentTypeChange(
  db: ContentDefinitionExecutor,
  principal: ContentPrincipal,
  input: { impactToken: string; operationId: string; strategies: readonly ContentTypeChangeStrategy[] },
  authorization: ContentDefinitionAuthorization,
  effects: ContentDefinitionEffects,
  opts: { codeTypes?: readonly CodeContentType[]; codeStatuses?: readonly CodeContentStatus[]; destructive?: ContentDestructiveDependencies } = {},
): Promise<ResolvedContentType | null> {
  validateStrategies(input.strategies)
  await authorization.assert(db, principal, 'changeType')
  const plan = await readImpact(db, input.impactToken)
  if (plan.definitionKind !== 'type') throw new ContentDefinitionError('stale-impact', 'impact token belongs to another definition kind')
  const { policyVersion } = await authorization.assert(db, principal, 'changeType', plan.key)
  if (policyVersion !== plan.policyVersion) throw new ContentDefinitionError('authorization-drift', 'authorization policy changed after preview')
  if (await strategyHash(input.strategies) !== plan.strategyHash) throw new ContentDefinitionError('stale-impact', 'strategies differ from preview')

  const scope = principalScope(principal)
  const requestHash = await canonicalContentMigrationHash({ impactToken: input.impactToken, strategies: input.strategies })
  const replay = await readDefinitionReceipt(db, input.operationId)
  if (replay) {
    if (replay.requestHash !== requestHash || replay.principalScope !== scope) throw new ContentDefinitionError('operation-conflict', 'operation id was already used for different bytes or scope')
    return replay.result === null ? null : Object.freeze(replay.result as unknown as ResolvedContentType)
  }

  const current = await resolveContentType(db, plan.key, { codeTypes: opts.codeTypes })
  if (current.origin === 'code') throw new ContentDefinitionError('code-definition-owned', 'code definition became effective after preview')
  if (current.version !== plan.expectedVersion) throw new ContentDefinitionError('stale-impact', 'definition version changed after preview')
  const rows = await typeCorpus(db, plan.key)
  const evidence = await corpusEvidence(rows as unknown as readonly Record<string, unknown>[])
  if (evidence.hash !== plan.corpusHash || evidence.version !== plan.corpusVersion) throw new ContentDefinitionError('stale-impact', 'affected corpus changed after preview')

  const request = requestRecord(plan)
  const backups = await verifyBackups(db, plan, input.strategies, opts.destructive)
  if (input.strategies[0]?.kind === 'reject' && plan.incompatibilities.length > 0) {
    throw new ContentDefinitionError('strategy-conflict', 'reject strategy cannot apply an incompatible change')
  }

  await applyTypeStrategies(db, plan.key, input.strategies, opts)
  const remaining = await db.execute<{ count: unknown }>(sql`SELECT COUNT(*) AS count FROM content_entries WHERE type = ${plan.key}`)
  const remainingCount = countValue(remaining[0]?.count ?? 0)
  if (request.delete && remainingCount > 0) throw new ContentDefinitionError('strategy-conflict', 'delete requires zero references or complete reassignment')

  const next = normalizeContentTypeDefinition({ ...current, ...(request.patch ?? {}), key: current.key, active: request.deactivate || request.delete ? false : current.active })
  const version = current.version + 1
  const revision = current.revision + 1
  const canonicalHash = await canonicalContentMigrationHash(next)
  await writeDefinitionVersion(db, { kind: 'type', key: current.key, revision, canonicalHash, definition: next as unknown as ContentSchemaValue, origin: 'db' })
  let result: ResolvedContentType | null
  if (request.delete) {
    await db.execute(sql`DELETE FROM content_type_definitions WHERE key = ${current.key} AND version = ${current.version}`)
    result = null
  } else {
    const now = new Date().toISOString()
    await db.execute(sql`UPDATE content_type_definitions SET version = ${version}, current_revision = ${revision}, active = ${next.active}, canonical_hash = ${canonicalHash}, definition = ${JSON.stringify(next)}, updated_at = ${now} WHERE key = ${current.key} AND version = ${current.version}`)
    result = Object.freeze({ ...next, origin: 'db', version, revision, canonicalHash })
  }

  await writeDefinitionReceipt(db, { operationId: input.operationId, kind: 'definition:change:type', payload: { requestHash, principalScope: scope, result: result as unknown as ContentSchemaValue } })
  const event: ContentDefinitionEvent = {
    id: crypto.randomUUID(), version: 1, kind: 'dbDefinitionChanged', operationId: input.operationId,
    principalId: principal.id, definitionKind: 'type', definitionKey: current.key,
    action: request.delete ? 'delete' : request.deactivate ? 'deactivate' : 'update', strategyHash: plan.strategyHash,
    impactToken: input.impactToken, authorizationPolicyVersion: policyVersion, corpusVersion: plan.corpusVersion,
    corpusHash: plan.corpusHash, affectedCounts: plan.affectedCounts, backupIds: backups,
  }
  await effects.audit.record(db, event)
  await effects.outbox.enqueue(db, event)
  return result
}

export async function previewContentStatusChange(
  db: ContentDefinitionExecutor,
  principal: ContentPrincipal,
  input: { key: string; expectedVersion: number; patch?: ContentStatusPatch; deactivate?: boolean; delete?: boolean; strategy: ContentStatusChangeStrategy },
  authorization: ContentDefinitionAuthorization,
  opts: { codeStatuses?: readonly CodeContentStatus[] } = {},
): Promise<ContentDefinitionImpact> {
  const { policyVersion } = await authorization.assert(db, principal, 'changeStatus', input.key)
  const current = await resolveContentStatus(db, input.key, opts)
  if (current.origin === 'code') throw new ContentDefinitionError('code-definition-owned', 'code statuses can only change through reconciliation')
  if (current.version !== input.expectedVersion) throw new ContentDefinitionError('stale-definition', 'expected version does not match current status')
  normalizeContentStatusDefinition({ ...current, ...(input.patch ?? {}), key: current.key })
  const rows = await statusCorpus(db, input.key)
  const evidence = await corpusEvidence(rows as unknown as readonly Record<string, unknown>[])
  const counts = Object.freeze({ entries: rows.length })
  const incompatibilities = Object.freeze(rows.length > 0 && (input.delete || input.deactivate || input.patch !== undefined) ? ['status-in-use'] : [])
  const plan: StoredImpactPlan = {
    definitionKind: 'status', key: input.key, expectedVersion: input.expectedVersion, policyVersion,
    corpusVersion: evidence.version, corpusHash: evidence.hash, affectedCounts: counts, incompatibilities,
    request: asSchemaValue({ patch: input.patch ?? null, deactivate: Boolean(input.deactivate), delete: Boolean(input.delete), strategy: input.strategy }),
    strategyHash: await strategyHash(input.strategy),
  }
  const token = await storeImpact(db, plan)
  return Object.freeze({ token, definitionKey: input.key, expectedVersion: input.expectedVersion, authorizationPolicyVersion: policyVersion, corpusVersion: evidence.version, corpusHash: evidence.hash, affectedCounts: counts, incompatibilities })
}

export async function applyContentStatusChange(
  db: ContentDefinitionExecutor,
  principal: ContentPrincipal,
  input: { impactToken: string; operationId: string; strategy: ContentStatusChangeStrategy },
  authorization: ContentDefinitionAuthorization,
  effects: ContentDefinitionEffects,
  opts: { codeStatuses?: readonly CodeContentStatus[] } = {},
): Promise<ResolvedContentStatus | null> {
  await authorization.assert(db, principal, 'changeStatus')
  const plan = await readImpact(db, input.impactToken)
  if (plan.definitionKind !== 'status') throw new ContentDefinitionError('stale-impact', 'impact token belongs to another definition kind')
  const { policyVersion } = await authorization.assert(db, principal, 'changeStatus', plan.key)
  if (policyVersion !== plan.policyVersion) throw new ContentDefinitionError('authorization-drift', 'authorization policy changed after preview')
  if (await strategyHash(input.strategy) !== plan.strategyHash) throw new ContentDefinitionError('stale-impact', 'strategy differs from preview')
  const scope = principalScope(principal)
  const requestHash = await canonicalContentMigrationHash({ impactToken: input.impactToken, strategy: input.strategy })
  const replay = await readDefinitionReceipt(db, input.operationId)
  if (replay) {
    if (replay.requestHash !== requestHash || replay.principalScope !== scope) throw new ContentDefinitionError('operation-conflict', 'operation id was already used for different bytes or scope')
    return replay.result === null ? null : Object.freeze(replay.result as unknown as ResolvedContentStatus)
  }
  const current = await resolveContentStatus(db, plan.key, opts)
  if (current.origin === 'code') throw new ContentDefinitionError('code-definition-owned', 'code status became effective after preview')
  if (current.version !== plan.expectedVersion) throw new ContentDefinitionError('stale-impact', 'status version changed after preview')
  const rows = await statusCorpus(db, plan.key)
  const evidence = await corpusEvidence(rows as unknown as readonly Record<string, unknown>[])
  if (evidence.hash !== plan.corpusHash) throw new ContentDefinitionError('stale-impact', 'status corpus changed after preview')
  const request = plan.request as unknown as { patch: ContentStatusPatch | null; deactivate: boolean; delete: boolean; strategy: ContentStatusChangeStrategy }

  if (rows.length > 0) {
    if (input.strategy.kind === 'reject') throw new ContentDefinitionError('strategy-conflict', 'status is still referenced')
    const target = await resolveContentStatus(db, input.strategy.targetStatusKey, opts)
    if (!target.active || target.key === current.key) throw new ContentDefinitionError('strategy-conflict', 'status mapping target must be another active status')
    await db.execute(sql`UPDATE content_entries SET status = ${target.key}, status_definition_revision = ${target.revision} WHERE status = ${current.key}`)
  }

  const next = normalizeContentStatusDefinition({ ...current, ...(request.patch ?? {}), key: current.key })
  const version = current.version + 1
  const revision = current.revision + 1
  const canonicalHash = await canonicalContentMigrationHash(next)
  await writeDefinitionVersion(db, { kind: 'status', key: current.key, revision, canonicalHash, definition: next as unknown as ContentSchemaValue, origin: 'db' })
  let result: ResolvedContentStatus | null
  if (request.delete) {
    await db.execute(sql`DELETE FROM content_status_definitions WHERE key = ${current.key} AND version = ${current.version}`)
    result = null
  } else {
    const active = request.deactivate ? false : current.active
    const persisted = { ...next, active }
    const now = new Date().toISOString()
    await db.execute(sql`UPDATE content_status_definitions SET version = ${version}, current_revision = ${revision}, active = ${active}, canonical_hash = ${canonicalHash}, definition = ${JSON.stringify(next)}, updated_at = ${now} WHERE key = ${current.key} AND version = ${current.version}`)
    result = Object.freeze({ ...persisted, origin: 'db', version, revision, canonicalHash })
  }
  await writeDefinitionReceipt(db, { operationId: input.operationId, kind: 'definition:change:status', payload: { requestHash, principalScope: scope, result: result as unknown as ContentSchemaValue } })
  const event: ContentDefinitionEvent = {
    id: crypto.randomUUID(), version: 1, kind: 'dbDefinitionChanged', operationId: input.operationId, principalId: principal.id,
    definitionKind: 'status', definitionKey: current.key, action: request.delete ? 'delete' : request.deactivate ? 'deactivate' : 'update',
    strategyHash: plan.strategyHash, impactToken: input.impactToken, authorizationPolicyVersion: policyVersion,
    corpusVersion: plan.corpusVersion, corpusHash: plan.corpusHash, affectedCounts: plan.affectedCounts, backupIds: [],
  }
  await effects.audit.record(db, event)
  await effects.outbox.enqueue(db, event)
  return result
}

export interface CodeContentRegistrySnapshot {
  version: string
  types: readonly CodeContentType[]
  statuses: readonly CodeContentStatus[]
}
export interface ContentCodeReconciliationPlan {
  token: string
  fromVersion: string
  nextCanonicalHash: string
  typeImpacts: readonly ContentDefinitionImpact[]
  statusImpacts: readonly ContentDefinitionImpact[]
}

interface StoredCodeReconciliationPlan {
  readonly currentVersion: string
  readonly nextCanonicalHash: string
  readonly policyVersion: string
  readonly next: CodeContentRegistrySnapshot
  readonly typeStrategies: Readonly<Record<string, readonly ContentTypeChangeStrategy[]>>
  readonly statusStrategies: Readonly<Record<string, ContentStatusChangeStrategy>>
  readonly typeImpacts: readonly ContentDefinitionImpact[]
  readonly statusImpacts: readonly ContentDefinitionImpact[]
}

async function currentCodeRegistryVersion(db: ContentDefinitionExecutor): Promise<string> {
  const types = (await __contentRegistryInternals.definitionRows(db, 'type'))
    .filter((row) => row.origin === 'code')
    .map((row) => ({ key: row.key, hash: row.canonicalHash, version: Number(row.version), active: __contentRegistryInternals.bool(row.active) }))
  const statuses = (await __contentStatusInternals.statusRows(db))
    .filter((row) => row.origin === 'code')
    .map((row) => ({ key: row.key, hash: row.canonicalHash, version: Number(row.version), active: __contentStatusInternals.bool(row.active) }))
  return canonicalContentMigrationHash({ types, statuses })
}

function normalizedCodeSnapshot(snapshot: CodeContentRegistrySnapshot): CodeContentRegistrySnapshot {
  const version = typeof snapshot.version === 'string' ? snapshot.version.trim() : ''
  if (!version || version.length > 256) throw new ContentDefinitionError('invalid-definition', 'code registry version must contain 1..256 characters', 'version')
  const typeKeys = new Set<string>()
  const statusKeys = new Set<string>()
  const types = snapshot.types.map((definition) => {
    const normalized = defineContentType(definition)
    if (typeKeys.has(normalized.key)) throw new ContentDefinitionError('definition-conflict', `duplicate code type ${normalized.key}`)
    typeKeys.add(normalized.key)
    return normalized
  })
  const statuses = snapshot.statuses.map((definition) => {
    const normalized = defineContentStatus(definition)
    if (statusKeys.has(normalized.key)) throw new ContentDefinitionError('definition-conflict', `duplicate code status ${normalized.key}`)
    statusKeys.add(normalized.key)
    return normalized
  })
  for (const definition of types) {
    for (const statusKey of definition.statusKeys ?? []) {
      if (!statusKeys.has(statusKey)) throw new ContentDefinitionError('invalid-definition', `code type ${definition.key} references missing status ${statusKey}`, 'statusKeys')
    }
  }
  return Object.freeze({ version, types: Object.freeze(types), statuses: Object.freeze(statuses) })
}

function resolvedTypeRow(row: Awaited<ReturnType<typeof __contentRegistryInternals.definitionRows>>[number]): ResolvedContentType {
  const definition = __contentRegistryInternals.parseDefinition(row.definition)
  return Object.freeze({
    ...definition,
    active: __contentRegistryInternals.bool(row.active),
    origin: row.origin === 'code' ? 'code' : 'db',
    version: Number(row.version),
    revision: Number(row.currentRevision),
    canonicalHash: row.canonicalHash,
    ...(row.shadowedDbVersion === null ? {} : { shadowedDbVersion: Number(row.shadowedDbVersion) }),
  })
}

function resolvedStatusRow(row: Awaited<ReturnType<typeof __contentStatusInternals.statusRows>>[number]): ResolvedContentStatus {
  const definition = __contentStatusInternals.parseStatus(row.definition)
  return Object.freeze({
    ...definition,
    active: __contentStatusInternals.bool(row.active),
    origin: row.origin === 'code' ? 'code' : 'db',
    version: Number(row.version),
    revision: Number(row.currentRevision),
    canonicalHash: row.canonicalHash,
    ...(row.shadowedDbVersion === null ? {} : { shadowedDbVersion: Number(row.shadowedDbVersion) }),
  })
}

async function codeTypeImpact(
  db: ContentDefinitionExecutor,
  current: ResolvedContentType,
  next: ContentTypeDefinition | null,
  strategies: readonly ContentTypeChangeStrategy[] | undefined,
  policyVersion: string,
): Promise<ContentDefinitionImpact> {
  const rows = await typeCorpus(db, current.key)
  const nextDefinition = next ?? normalizeContentTypeDefinition({ ...current, active: false })
  const impact = typeImpactCounts(current, nextDefinition, rows)
  if (next === null && rows.length > 0) impact.incompatibilities.push('deactivate-with-entries')
  const chosen = strategies ?? [{ kind: 'reject' } as const]
  validateStrategies(chosen)
  if (impact.incompatibilities.length > 0 && strategies === undefined) {
    throw new ContentDefinitionError('strategy-conflict', `code type ${current.key} requires explicit narrowing strategies`)
  }
  const evidence = await corpusEvidence(rows as unknown as readonly Record<string, unknown>[])
  const plan: StoredImpactPlan = {
    definitionKind: 'type',
    key: current.key,
    expectedVersion: current.version,
    policyVersion,
    corpusVersion: evidence.version,
    corpusHash: evidence.hash,
    affectedCounts: Object.freeze({ ...impact.counts }),
    incompatibilities: Object.freeze([...new Set(impact.incompatibilities)].sort()),
    request: asSchemaValue({ codeReconciliation: true, nextDefinition, remove: next === null, strategies: chosen }),
    strategyHash: await strategyHash(chosen),
  }
  const token = await storeImpact(db, plan)
  return Object.freeze({ token, definitionKey: current.key, expectedVersion: current.version, authorizationPolicyVersion: policyVersion, corpusVersion: evidence.version, corpusHash: evidence.hash, affectedCounts: plan.affectedCounts, incompatibilities: plan.incompatibilities })
}

async function codeStatusImpact(
  db: ContentDefinitionExecutor,
  current: ResolvedContentStatus,
  next: ContentStatusDefinition | null,
  strategy: ContentStatusChangeStrategy | undefined,
  policyVersion: string,
): Promise<ContentDefinitionImpact> {
  const rows = await statusCorpus(db, current.key)
  const chosen = strategy ?? { kind: 'reject' as const }
  if (rows.length > 0 && strategy === undefined) {
    throw new ContentDefinitionError('strategy-conflict', `code status ${current.key} requires an explicit mapping/reject strategy`)
  }
  const evidence = await corpusEvidence(rows as unknown as readonly Record<string, unknown>[])
  const incompatibilities = rows.length > 0 ? Object.freeze(['status-in-use']) : Object.freeze([] as string[])
  const plan: StoredImpactPlan = {
    definitionKind: 'status', key: current.key, expectedVersion: current.version, policyVersion,
    corpusVersion: evidence.version, corpusHash: evidence.hash, affectedCounts: Object.freeze({ entries: rows.length }), incompatibilities,
    request: asSchemaValue({ codeReconciliation: true, nextDefinition: next, remove: next === null, strategy: chosen }),
    strategyHash: await strategyHash(chosen),
  }
  const token = await storeImpact(db, plan)
  return Object.freeze({ token, definitionKey: current.key, expectedVersion: current.version, authorizationPolicyVersion: policyVersion, corpusVersion: evidence.version, corpusHash: evidence.hash, affectedCounts: plan.affectedCounts, incompatibilities })
}

export async function previewContentCodeReconciliation(
  db: ContentDefinitionExecutor,
  principal: ContentPrincipal,
  input: {
    expectedCurrentVersion: string
    next: CodeContentRegistrySnapshot
    typeStrategies: Readonly<Record<string, readonly ContentTypeChangeStrategy[]>>
    statusStrategies: Readonly<Record<string, ContentStatusChangeStrategy>>
  },
  authorization: ContentDefinitionAuthorization,
): Promise<ContentCodeReconciliationPlan> {
  const { policyVersion } = await authorization.assert(db, principal, 'reconcileCode')
  const currentVersion = await currentCodeRegistryVersion(db)
  if (input.expectedCurrentVersion !== currentVersion) throw new ContentDefinitionError('stale-definition', 'code registry version changed')
  const next = normalizedCodeSnapshot(input.next)
  const nextCanonicalHash = await canonicalContentMigrationHash(next)
  const currentTypeRows = await __contentRegistryInternals.definitionRows(db, 'type')
  const currentStatusRows = await __contentStatusInternals.statusRows(db)
  const nextTypes = new Map(next.types.map((definition) => [definition.key, definition]))
  const nextStatuses = new Map(next.statuses.map((definition) => [definition.key, definition]))
  const typeImpacts: ContentDefinitionImpact[] = []
  const statusImpacts: ContentDefinitionImpact[] = []

  for (const row of currentTypeRows) {
    const nextDefinition = nextTypes.get(row.key)
    const current = resolvedTypeRow(row)
    const nextHash = nextDefinition ? await canonicalContentMigrationHash(nextDefinition) : undefined
    const needsImpact = (row.origin === 'code' && nextDefinition === undefined)
      || (nextDefinition !== undefined && (row.canonicalHash !== nextHash || row.origin !== 'code'))
    if (!needsImpact) continue
    typeImpacts.push(await codeTypeImpact(db, current, nextDefinition ?? null, input.typeStrategies[row.key], policyVersion))
  }

  for (const row of currentStatusRows) {
    const nextDefinition = nextStatuses.get(row.key)
    const current = resolvedStatusRow(row)
    const nextHash = nextDefinition ? await canonicalContentMigrationHash(nextDefinition) : undefined
    const needsImpact = (row.origin === 'code' && nextDefinition === undefined)
      || (nextDefinition !== undefined && (row.canonicalHash !== nextHash || row.origin !== 'code'))
    if (!needsImpact) continue
    statusImpacts.push(await codeStatusImpact(db, current, nextDefinition ?? null, input.statusStrategies[row.key], policyVersion))
  }

  const stored: StoredCodeReconciliationPlan = Object.freeze({
    currentVersion,
    nextCanonicalHash,
    policyVersion,
    next,
    typeStrategies: Object.freeze({ ...input.typeStrategies }),
    statusStrategies: Object.freeze({ ...input.statusStrategies }),
    typeImpacts: Object.freeze(typeImpacts),
    statusImpacts: Object.freeze(statusImpacts),
  })
  const token = await canonicalContentMigrationHash(stored)
  const now = new Date().toISOString()
  const rows = await db.execute<{ operationId: string }>(sql`SELECT operation_id AS "operationId" FROM content_lifecycle_journal WHERE operation_id = ${`code-impact:${token}`}`)
  if (rows.length === 0) {
    await db.execute(sql`INSERT INTO content_lifecycle_journal (id, operation_id, kind, state, payload, created_at, updated_at) VALUES (${crypto.randomUUID()}, ${`code-impact:${token}`}, 'code-registry-impact', 'prepared', ${JSON.stringify(stored)}, ${now}, ${now})`)
  }
  return Object.freeze({ token, fromVersion: currentVersion, nextCanonicalHash, typeImpacts: stored.typeImpacts, statusImpacts: stored.statusImpacts })
}

async function readCodePlan(db: ContentDefinitionExecutor, token: string): Promise<StoredCodeReconciliationPlan> {
  const rows = await db.execute<{ payload: unknown }>(sql`SELECT payload FROM content_lifecycle_journal WHERE operation_id = ${`code-impact:${token}`} AND kind = 'code-registry-impact'`)
  if (!rows[0]) throw new ContentDefinitionError('stale-impact', 'code reconciliation token is unknown')
  const plan = parsePayload<StoredCodeReconciliationPlan>(rows[0].payload)
  if (await canonicalContentMigrationHash(plan) !== token) throw new ContentDefinitionError('stale-impact', 'code reconciliation token bytes do not match stored plan')
  return plan
}

async function verifyCodeImpact(
  db: ContentDefinitionExecutor,
  impact: ContentDefinitionImpact,
  kind: 'type' | 'status',
): Promise<StoredImpactPlan> {
  const stored = await readImpact(db, impact.token)
  if (stored.definitionKind !== kind || stored.key !== impact.definitionKey || stored.expectedVersion !== impact.expectedVersion) {
    throw new ContentDefinitionError('stale-impact', 'code impact identity changed')
  }
  if (kind === 'type') {
    const current = await resolveContentType(db, stored.key)
    if (current.version !== stored.expectedVersion || current.canonicalHash !== (await __contentRegistryInternals.definitionRows(db, 'type', stored.key))[0]?.canonicalHash) {
      throw new ContentDefinitionError('stale-impact', 'code type definition changed after preview')
    }
    const rows = await typeCorpus(db, stored.key)
    const evidence = await corpusEvidence(rows as unknown as readonly Record<string, unknown>[])
    if (evidence.hash !== stored.corpusHash || evidence.version !== stored.corpusVersion) throw new ContentDefinitionError('stale-impact', 'code type corpus changed after preview')
  } else {
    const current = await resolveContentStatus(db, stored.key)
    if (current.version !== stored.expectedVersion) throw new ContentDefinitionError('stale-impact', 'code status definition changed after preview')
    const rows = await statusCorpus(db, stored.key)
    const evidence = await corpusEvidence(rows as unknown as readonly Record<string, unknown>[])
    if (evidence.hash !== stored.corpusHash || evidence.version !== stored.corpusVersion) throw new ContentDefinitionError('stale-impact', 'code status corpus changed after preview')
  }
  return stored
}

export async function applyContentCodeReconciliation(
  db: ContentDefinitionExecutor,
  principal: ContentPrincipal,
  input: {
    planToken: string
    operationId: string
    next: CodeContentRegistrySnapshot
    typeStrategies: Readonly<Record<string, readonly ContentTypeChangeStrategy[]>>
    statusStrategies: Readonly<Record<string, ContentStatusChangeStrategy>>
  },
  authorization: ContentDefinitionAuthorization,
  effects: ContentDefinitionEffects,
  destructive?: ContentDestructiveDependencies,
): Promise<CodeContentRegistrySnapshot> {
  const { policyVersion } = await authorization.assert(db, principal, 'reconcileCode')
  const plan = await readCodePlan(db, input.planToken)
  if (plan.policyVersion !== policyVersion) throw new ContentDefinitionError('authorization-drift', 'authorization policy changed after code preview')
  const next = normalizedCodeSnapshot(input.next)
  if (await canonicalContentMigrationHash(next) !== plan.nextCanonicalHash) throw new ContentDefinitionError('stale-impact', 'next code registry bytes differ from preview')
  if (await canonicalContentMigrationHash(input.typeStrategies) !== await canonicalContentMigrationHash(plan.typeStrategies)
    || await canonicalContentMigrationHash(input.statusStrategies) !== await canonicalContentMigrationHash(plan.statusStrategies)) {
    throw new ContentDefinitionError('stale-impact', 'code reconciliation strategies differ from preview')
  }
  const scope = principalScope(principal)
  const requestHash = await canonicalContentMigrationHash({ planToken: input.planToken, nextCanonicalHash: plan.nextCanonicalHash, typeStrategies: input.typeStrategies, statusStrategies: input.statusStrategies })
  const replay = await readDefinitionReceipt(db, input.operationId)
  if (replay) {
    if (replay.requestHash !== requestHash || replay.principalScope !== scope) throw new ContentDefinitionError('operation-conflict', 'operation id was already used for different code bytes or scope')
    return Object.freeze(replay.result as unknown as CodeContentRegistrySnapshot)
  }
  if (await currentCodeRegistryVersion(db) !== plan.currentVersion) throw new ContentDefinitionError('stale-impact', 'code registry changed after preview')

  const nextTypeMap = new Map(next.types.map((definition) => [definition.key, definition]))
  const nextStatusMap = new Map(next.statuses.map((definition) => [definition.key, definition]))
  const verifiedBackups: string[] = []
  const affectedCounts: Record<string, number> = {}
  const corpusHashes: string[] = []
  const verifiedTypeImpacts: Array<{ stored: StoredImpactPlan; strategies: readonly ContentTypeChangeStrategy[] }> = []
  const verifiedStatusImpacts: Array<{ stored: StoredImpactPlan; strategy: ContentStatusChangeStrategy }> = []

  // Verify the entire reconciliation plan before the first mutation. A type strategy can
  // legitimately change a status corpus (for example review -> draft); verifying status
  // impacts after that mutation would make a valid, already-bound plan invalidate itself.
  for (const impact of plan.typeImpacts) {
    const stored = await verifyCodeImpact(db, impact, 'type')
    const strategies = input.typeStrategies[stored.key] ?? [{ kind: 'reject' } as const]
    if (await strategyHash(strategies) !== stored.strategyHash) throw new ContentDefinitionError('stale-impact', `type strategies for ${stored.key} differ from preview`)
    if (strategies[0]?.kind === 'reject' && stored.incompatibilities.length > 0) throw new ContentDefinitionError('strategy-conflict', `code type ${stored.key} has unresolved incompatibilities`)
    verifiedBackups.push(...await verifyBackups(db, stored, strategies, destructive))
    verifiedTypeImpacts.push({ stored, strategies })
    corpusHashes.push(stored.corpusHash)
    for (const [metric, count] of Object.entries(stored.affectedCounts)) affectedCounts[`type:${stored.key}:${metric}`] = count
  }

  for (const impact of plan.statusImpacts) {
    const stored = await verifyCodeImpact(db, impact, 'status')
    const strategy = input.statusStrategies[stored.key] ?? { kind: 'reject' as const }
    if (await strategyHash(strategy) !== stored.strategyHash) throw new ContentDefinitionError('stale-impact', `status strategy for ${stored.key} differs from preview`)
    const affectedEntries = stored.affectedCounts.entries ?? 0
    if (affectedEntries > 0 && strategy.kind === 'reject') throw new ContentDefinitionError('strategy-conflict', `code status ${stored.key} is still referenced`)
    if (affectedEntries > 0 && strategy.kind === 'map') {
      const target = nextStatusMap.get(strategy.targetStatusKey) ?? await resolveContentStatus(db, strategy.targetStatusKey, { codeStatuses: next.statuses })
      if (!target || target.key === stored.key) throw new ContentDefinitionError('strategy-conflict', 'code status mapping target must be another active status')
    }
    verifiedStatusImpacts.push({ stored, strategy })
    corpusHashes.push(stored.corpusHash)
    for (const [metric, count] of Object.entries(stored.affectedCounts)) affectedCounts[`status:${stored.key}:${metric}`] = count
  }

  for (const { stored, strategies } of verifiedTypeImpacts) {
    await applyTypeStrategies(db, stored.key, strategies, { codeTypes: next.types, codeStatuses: next.statuses })
  }

  for (const { stored, strategy } of verifiedStatusImpacts) {
    if ((stored.affectedCounts.entries ?? 0) === 0 || strategy.kind === 'reject') continue
    const target = nextStatusMap.get(strategy.targetStatusKey) ?? await resolveContentStatus(db, strategy.targetStatusKey, { codeStatuses: next.statuses })
    const revision = 'revision' in target ? target.revision : 1
    await db.execute(sql`UPDATE content_entries SET status = ${target.key}, status_definition_revision = ${revision} WHERE status = ${stored.key}`)
  }

  const currentTypeRows = await __contentRegistryInternals.definitionRows(db, 'type')
  const currentStatusRows = await __contentStatusInternals.statusRows(db)
  const changedKeys: string[] = []

  for (const definition of next.types) {
    const normalized = normalizeContentTypeDefinition(definition)
    const hash = await canonicalContentMigrationHash(normalized)
    const existing = currentTypeRows.find((row) => row.key === normalized.key)
    const now = new Date().toISOString()
    if (!existing) {
      await db.execute(sql`INSERT INTO content_type_definitions (key, origin, version, active, current_revision, canonical_hash, definition, created_at, updated_at) VALUES (${normalized.key}, 'code', 1, ${true}, 1, ${hash}, ${JSON.stringify(normalized)}, ${now}, ${now})`)
      await writeDefinitionVersion(db, { kind: 'type', key: normalized.key, revision: 1, canonicalHash: hash, definition: normalized as unknown as ContentSchemaValue, origin: 'code' })
      changedKeys.push(`type:${normalized.key}`)
    } else if (existing.canonicalHash !== hash || existing.origin !== 'code' || !__contentRegistryInternals.bool(existing.active)) {
      const version = Number(existing.version) + 1
      const revision = Number(existing.currentRevision) + 1
      await db.execute(sql`UPDATE content_type_definitions SET origin = 'code', version = ${version}, active = ${true}, current_revision = ${revision}, canonical_hash = ${hash}, definition = ${JSON.stringify(normalized)}, shadowed_db_version = ${existing.origin === 'db' ? Number(existing.version) : existing.shadowedDbVersion}, updated_at = ${now} WHERE key = ${normalized.key}`)
      await writeDefinitionVersion(db, { kind: 'type', key: normalized.key, revision, canonicalHash: hash, definition: normalized as unknown as ContentSchemaValue, origin: 'code' })
      changedKeys.push(`type:${normalized.key}`)
    }
  }
  for (const existing of currentTypeRows.filter((row) => row.origin === 'code' && !nextTypeMap.has(row.key) && __contentRegistryInternals.bool(row.active))) {
    const current = resolvedTypeRow(existing)
    const inactive = normalizeContentTypeDefinition({ ...current, active: false })
    const version = current.version + 1
    const revision = current.revision + 1
    const hash = await canonicalContentMigrationHash(inactive)
    const now = new Date().toISOString()
    await db.execute(sql`UPDATE content_type_definitions SET version = ${version}, active = ${false}, current_revision = ${revision}, canonical_hash = ${hash}, definition = ${JSON.stringify(inactive)}, updated_at = ${now} WHERE key = ${existing.key}`)
    await writeDefinitionVersion(db, { kind: 'type', key: existing.key, revision, canonicalHash: hash, definition: inactive as unknown as ContentSchemaValue, origin: 'code' })
    changedKeys.push(`type:${existing.key}`)
  }

  for (const definition of next.statuses) {
    const normalized = normalizeContentStatusDefinition(definition)
    const hash = await canonicalContentMigrationHash(normalized)
    const existing = currentStatusRows.find((row) => row.key === normalized.key)
    const now = new Date().toISOString()
    if (!existing) {
      await db.execute(sql`INSERT INTO content_status_definitions (key, origin, version, active, current_revision, canonical_hash, definition, created_at, updated_at) VALUES (${normalized.key}, 'code', 1, ${true}, 1, ${hash}, ${JSON.stringify(normalized)}, ${now}, ${now})`)
      await writeDefinitionVersion(db, { kind: 'status', key: normalized.key, revision: 1, canonicalHash: hash, definition: normalized as unknown as ContentSchemaValue, origin: 'code' })
      changedKeys.push(`status:${normalized.key}`)
    } else if (existing.canonicalHash !== hash || existing.origin !== 'code' || !__contentStatusInternals.bool(existing.active)) {
      const version = Number(existing.version) + 1
      const revision = Number(existing.currentRevision) + 1
      await db.execute(sql`UPDATE content_status_definitions SET origin = 'code', version = ${version}, active = ${true}, current_revision = ${revision}, canonical_hash = ${hash}, definition = ${JSON.stringify(normalized)}, shadowed_db_version = ${existing.origin === 'db' ? Number(existing.version) : existing.shadowedDbVersion}, updated_at = ${now} WHERE key = ${normalized.key}`)
      await writeDefinitionVersion(db, { kind: 'status', key: normalized.key, revision, canonicalHash: hash, definition: normalized as unknown as ContentSchemaValue, origin: 'code' })
      changedKeys.push(`status:${normalized.key}`)
    }
  }
  for (const existing of currentStatusRows.filter((row) => row.origin === 'code' && !nextStatusMap.has(row.key) && __contentStatusInternals.bool(row.active))) {
    const current = resolvedStatusRow(existing)
    const version = current.version + 1
    const revision = current.revision + 1
    const nextDefinition = normalizeContentStatusDefinition(current)
    const hash = await canonicalContentMigrationHash(nextDefinition)
    const now = new Date().toISOString()
    await db.execute(sql`UPDATE content_status_definitions SET version = ${version}, active = ${false}, current_revision = ${revision}, canonical_hash = ${hash}, definition = ${JSON.stringify(nextDefinition)}, updated_at = ${now} WHERE key = ${existing.key}`)
    await writeDefinitionVersion(db, { kind: 'status', key: existing.key, revision, canonicalHash: hash, definition: nextDefinition as unknown as ContentSchemaValue, origin: 'code' })
    changedKeys.push(`status:${existing.key}`)
  }

  const result = next
  await writeDefinitionReceipt(db, { operationId: input.operationId, kind: 'definition:reconcile:code', payload: { requestHash, principalScope: scope, result: result as unknown as ContentSchemaValue } })
  const event: ContentDefinitionEvent = {
    id: crypto.randomUUID(), version: 1, kind: 'codeRegistryReconciled', operationId: input.operationId,
    principalId: principal.id, fromVersion: plan.currentVersion, toVersion: next.version,
    changedKeys: Object.freeze(changedKeys.sort()), strategiesHash: await canonicalContentMigrationHash({ typeStrategies: input.typeStrategies, statusStrategies: input.statusStrategies }),
    authorizationPolicyVersion: policyVersion, corpusHashes: Object.freeze([...new Set(corpusHashes)].sort()),
    affectedCounts: Object.freeze(affectedCounts), backupIds: Object.freeze([...new Set(verifiedBackups)].sort()),
  }
  await effects.audit.record(db, event)
  await effects.outbox.enqueue(db, event)
  return result
}

export interface ContentBackupEntrySnapshot {
  readonly id: string
  readonly excerpt?: string
  readonly featuredMedia?: ContentSchemaValue | null
  readonly parentId?: string | null
  readonly menuOrder?: number
  readonly templateKey?: string | null
  readonly format?: string | null
}

export interface ContentBackupTypeDefinitionSnapshot {
  readonly key: string
  readonly definition: ContentTypeDefinition
}

export interface ContentBackupStatusDefinitionSnapshot {
  readonly key: string
  readonly definition: ContentStatusDefinition
  readonly active: boolean
}

export interface ContentBackupPayload {
  readonly version: 1
  readonly typeDefinitions?: readonly ContentBackupTypeDefinitionSnapshot[]
  readonly statusDefinitions?: readonly ContentBackupStatusDefinitionSnapshot[]
  readonly entries?: readonly ContentBackupEntrySnapshot[]
}

function own(record: object, key: PropertyKey): boolean {
  return Object.prototype.hasOwnProperty.call(record, key)
}

function parseBackupPayload(value: ContentSchemaValue): ContentBackupPayload {
  if (typeof value !== 'object' || value === null || Array.isArray(value)) {
    throw new ContentDefinitionError('backup-mismatch', 'backup payload must be an object')
  }
  const record = value as Record<string, unknown>
  if (record.version !== 1) throw new ContentDefinitionError('backup-mismatch', 'backup payload version must be 1')
  for (const section of ['entries', 'typeDefinitions', 'statusDefinitions'] as const) {
    if (record[section] !== undefined && !Array.isArray(record[section])) {
      throw new ContentDefinitionError('backup-mismatch', `${section} must be an array`)
    }
  }
  const payload = record as unknown as ContentBackupPayload
  const entryIds = new Set<string>()
  for (const entry of payload.entries ?? []) {
    if (typeof entry !== 'object' || entry === null || typeof entry.id !== 'string' || entry.id.length === 0) {
      throw new ContentDefinitionError('backup-mismatch', 'backup entry snapshots require a non-empty id')
    }
    if (entryIds.has(entry.id)) throw new ContentDefinitionError('backup-mismatch', `duplicate backup entry ${entry.id}`)
    entryIds.add(entry.id)
    if (own(entry, 'excerpt') && typeof entry.excerpt !== 'string') throw new ContentDefinitionError('backup-mismatch', `entry ${entry.id} excerpt must be a string`)
    if (own(entry, 'parentId') && entry.parentId !== null && typeof entry.parentId !== 'string') throw new ContentDefinitionError('backup-mismatch', `entry ${entry.id} parentId must be string|null`)
    if (own(entry, 'menuOrder') && !Number.isSafeInteger(entry.menuOrder)) throw new ContentDefinitionError('backup-mismatch', `entry ${entry.id} menuOrder must be a safe integer`)
    if (own(entry, 'templateKey') && entry.templateKey !== null && typeof entry.templateKey !== 'string') throw new ContentDefinitionError('backup-mismatch', `entry ${entry.id} templateKey must be string|null`)
    if (own(entry, 'format') && entry.format !== null && typeof entry.format !== 'string') throw new ContentDefinitionError('backup-mismatch', `entry ${entry.id} format must be string|null`)
    if (!['excerpt', 'featuredMedia', 'parentId', 'menuOrder', 'templateKey', 'format'].some((key) => own(entry, key))) {
      throw new ContentDefinitionError('backup-mismatch', `entry ${entry.id} snapshot contains no restorable fields`)
    }
  }
  const typeKeys = new Set<string>()
  for (const item of payload.typeDefinitions ?? []) {
    if (typeof item !== 'object' || item === null || typeof item.key !== 'string') throw new ContentDefinitionError('backup-mismatch', 'type definition snapshot is malformed')
    if (typeKeys.has(item.key)) throw new ContentDefinitionError('backup-mismatch', `duplicate type definition ${item.key}`)
    typeKeys.add(item.key)
    if (normalizeContentTypeDefinition(item.definition).key !== item.key) throw new ContentDefinitionError('backup-mismatch', `type definition key mismatch for ${item.key}`)
  }
  const statusKeys = new Set<string>()
  for (const item of payload.statusDefinitions ?? []) {
    if (typeof item !== 'object' || item === null || typeof item.key !== 'string' || typeof item.active !== 'boolean') throw new ContentDefinitionError('backup-mismatch', 'status definition snapshot is malformed')
    if (statusKeys.has(item.key)) throw new ContentDefinitionError('backup-mismatch', `duplicate status definition ${item.key}`)
    statusKeys.add(item.key)
    if (normalizeContentStatusDefinition(item.definition).key !== item.key) throw new ContentDefinitionError('backup-mismatch', `status definition key mismatch for ${item.key}`)
  }
  return Object.freeze({
    version: 1,
    ...(payload.entries ? { entries: Object.freeze([...payload.entries]) } : {}),
    ...(payload.typeDefinitions ? { typeDefinitions: Object.freeze([...payload.typeDefinitions]) } : {}),
    ...(payload.statusDefinitions ? { statusDefinitions: Object.freeze([...payload.statusDefinitions]) } : {}),
  })
}

function normalizeJsonColumn(value: unknown): ContentSchemaValue | null {
  if (value === null) return null
  if (typeof value === 'string') {
    try { return JSON.parse(value) as ContentSchemaValue } catch { return value }
  }
  return value as ContentSchemaValue
}

async function contentBackupDestinationState(db: ContentDefinitionExecutor, payload: ContentBackupPayload): Promise<ContentSchemaValue> {
  const entries: ContentSchemaValue[] = []
  for (const entry of [...(payload.entries ?? [])].sort((a, b) => a.id.localeCompare(b.id))) {
    const rows = await db.execute<Record<string, unknown>>(sql`SELECT id, excerpt, featured_media AS "featuredMedia", parent_id AS "parentId", menu_order AS "menuOrder", template_key AS "templateKey", format FROM content_entries WHERE id = ${entry.id}`)
    const row = rows[0]
    if (!row || rows.length !== 1) throw new ContentDefinitionError('backup-mismatch', `restore destination entry ${entry.id} is missing`)
    const state: Record<string, ContentSchemaValue> = { id: entry.id }
    if (own(entry, 'excerpt')) state.excerpt = String(row.excerpt)
    if (own(entry, 'featuredMedia')) state.featuredMedia = normalizeJsonColumn(row.featuredMedia)
    if (own(entry, 'parentId')) state.parentId = row.parentId === null ? null : String(row.parentId)
    if (own(entry, 'menuOrder')) state.menuOrder = Number(row.menuOrder)
    if (own(entry, 'templateKey')) state.templateKey = row.templateKey === null ? null : String(row.templateKey)
    if (own(entry, 'format')) state.format = row.format === null ? null : String(row.format)
    entries.push(state)
  }

  const types: ContentSchemaValue[] = []
  for (const item of [...(payload.typeDefinitions ?? [])].sort((a, b) => a.key.localeCompare(b.key))) {
    const current = await resolveContentType(db, item.key)
    const refs = await db.execute<{ id: string; revision: number | string }>(sql`SELECT id, type_definition_revision AS revision FROM content_entries WHERE type = ${item.key} ORDER BY id`)
    types.push({ key: item.key, origin: current.origin, version: current.version, revision: current.revision, active: current.active, canonicalHash: current.canonicalHash, references: refs.map((row) => ({ id: row.id, revision: Number(row.revision) })) })
  }

  const statuses: ContentSchemaValue[] = []
  for (const item of [...(payload.statusDefinitions ?? [])].sort((a, b) => a.key.localeCompare(b.key))) {
    const current = await resolveContentStatus(db, item.key)
    const refs = await db.execute<{ id: string; revision: number | string }>(sql`SELECT id, status_definition_revision AS revision FROM content_entries WHERE status = ${item.key} ORDER BY id`)
    statuses.push({ key: item.key, origin: current.origin, version: current.version, revision: current.revision, active: current.active, canonicalHash: current.canonicalHash, references: refs.map((row) => ({ id: row.id, revision: Number(row.revision) })) })
  }
  return { version: 1, entries, typeDefinitions: types, statusDefinitions: statuses }
}

export async function contentBackupDestinationHash(db: ContentDefinitionExecutor, payload: ContentBackupPayload): Promise<string> {
  return canonicalContentMigrationHash(await contentBackupDestinationState(db, payload))
}

function manifestsMatch(left: ContentBackupManifest, right: ContentBackupManifest): boolean {
  return left.id === right.id
    && left.scopeHash === right.scopeHash
    && left.corpusVersion === right.corpusVersion
    && left.corpusHash === right.corpusHash
    && left.byteCount === right.byteCount
    && left.immutable === true
    && right.immutable === true
}

export async function restoreContentBackup(
  db: ContentDefinitionExecutor,
  principal: ContentPrincipal,
  input: { manifest: ContentBackupManifest; destinationCorpusHash: string; operationId: string },
  authorization: ContentDefinitionAuthorization,
  effects: ContentDefinitionEffects,
  deps: ContentDestructiveDependencies,
): Promise<ContentBackupRestoreResult> {
  const { policyVersion } = await authorization.assert(db, principal, 'restoreBackup')
  const scope = principalScope(principal)
  const requestHash = await canonicalContentMigrationHash({ manifestId: input.manifest.id, destinationCorpusHash: input.destinationCorpusHash })
  const replay = await readDefinitionReceipt(db, input.operationId)
  if (replay) {
    if (replay.requestHash !== requestHash || replay.principalScope !== scope) throw new ContentDefinitionError('operation-conflict', 'restore operation id was reused with different identity')
    return Object.freeze(replay.result as unknown as ContentBackupRestoreResult)
  }
  if (!input.manifest.immutable) throw new ContentDefinitionError('backup-mismatch', 'restore requires an immutable backup')

  const verified = await deps.backups.verify(db, input.manifest.id, input.manifest.scopeHash, input.manifest.corpusVersion, input.manifest.corpusHash, input.manifest.itemCounts)
  if (!manifestsMatch(verified, input.manifest)
    || await canonicalContentMigrationHash(verified.itemCounts) !== await canonicalContentMigrationHash(input.manifest.itemCounts)) {
    throw new ContentDefinitionError('backup-mismatch', 'restore manifest verification changed identity')
  }
  const snapshot = await deps.backups.read(db, verified)
  if (!manifestsMatch(snapshot.manifest, verified)
    || await canonicalContentMigrationHash(snapshot.manifest.itemCounts) !== await canonicalContentMigrationHash(verified.itemCounts)) {
    throw new ContentDefinitionError('backup-mismatch', 'backup store returned a different snapshot identity')
  }
  const payload = parseBackupPayload(snapshot.payload)

  const affectedTypeKeys = new Set((payload.typeDefinitions ?? []).map((item) => item.key))
  for (const entry of payload.entries ?? []) {
    const rows = await db.execute<{ type: string }>(sql`SELECT type FROM content_entries WHERE id = ${entry.id}`)
    if (rows.length !== 1) throw new ContentDefinitionError('backup-mismatch', `restore destination entry ${entry.id} is missing`)
    affectedTypeKeys.add(rows[0]!.type)
  }
  for (const item of payload.statusDefinitions ?? []) {
    const rows = await db.execute<{ type: string }>(sql`SELECT DISTINCT type FROM content_entries WHERE status = ${item.key} ORDER BY type`)
    for (const row of rows) affectedTypeKeys.add(row.type)
  }
  for (const key of [...affectedTypeKeys].sort()) {
    const assertion = await authorization.assert(db, principal, 'restoreBackup', key)
    if (assertion.policyVersion !== policyVersion) throw new ContentDefinitionError('authorization-drift', 'authorization policy changed while authorizing restore scope')
  }

  const destinationHash = await contentBackupDestinationHash(db, payload)
  if (destinationHash !== input.destinationCorpusHash) throw new ContentDefinitionError('stale-impact', 'restore destination corpus changed')

  const typeCurrents = new Map<string, ResolvedContentType>()
  for (const item of payload.typeDefinitions ?? []) {
    const current = await resolveContentType(db, item.key)
    if (current.origin !== 'db') throw new ContentDefinitionError('code-definition-owned', `cannot restore deploy-owned type ${item.key}`)
    typeCurrents.set(item.key, current)
  }
  const statusCurrents = new Map<string, ResolvedContentStatus>()
  for (const item of payload.statusDefinitions ?? []) {
    const current = await resolveContentStatus(db, item.key)
    if (current.origin !== 'db') throw new ContentDefinitionError('code-definition-owned', `cannot restore deploy-owned status ${item.key}`)
    statusCurrents.set(item.key, current)
  }

  const restoredCounts: Record<string, number> = {}
  const now = new Date().toISOString()
  for (const entry of payload.entries ?? []) {
    const assignments: SQL[] = []
    if (own(entry, 'excerpt')) assignments.push(sql`excerpt = ${entry.excerpt}`)
    if (own(entry, 'featuredMedia')) assignments.push(sql`featured_media = ${entry.featuredMedia === null ? null : JSON.stringify(entry.featuredMedia)}`)
    if (own(entry, 'parentId')) assignments.push(sql`parent_id = ${entry.parentId ?? null}`)
    if (own(entry, 'menuOrder')) assignments.push(sql`menu_order = ${entry.menuOrder}`)
    if (own(entry, 'templateKey')) assignments.push(sql`template_key = ${entry.templateKey ?? null}`)
    if (own(entry, 'format')) assignments.push(sql`format = ${entry.format ?? null}`)
    const updated = await db.execute<{ id: string }>(sql`UPDATE content_entries SET ${sql.join(assignments, sql`, `)}, updated_at = ${now} WHERE id = ${entry.id} RETURNING id`)
    if (updated.length !== 1) throw new ContentDefinitionError('stale-impact', `restore destination entry ${entry.id} changed or disappeared`)
  }
  if ((payload.entries ?? []).length > 0) restoredCounts.entries = payload.entries!.length

  for (const item of payload.typeDefinitions ?? []) {
    const current = typeCurrents.get(item.key)!
    const next = normalizeContentTypeDefinition(item.definition)
    const revision = current.revision + 1
    const version = current.version + 1
    const hash = await canonicalContentMigrationHash(next)
    await writeDefinitionVersion(db, { kind: 'type', key: item.key, revision, canonicalHash: hash, definition: next as unknown as ContentSchemaValue, origin: 'db' })
    const updated = await db.execute<{ key: string }>(sql`UPDATE content_type_definitions SET version = ${version}, current_revision = ${revision}, active = ${next.active}, canonical_hash = ${hash}, definition = ${JSON.stringify(next)}, updated_at = ${now} WHERE key = ${item.key} AND version = ${current.version} RETURNING key`)
    if (updated.length !== 1) throw new ContentDefinitionError('stale-impact', `type definition ${item.key} changed during restore`)
    await db.execute(sql`UPDATE content_entries SET type_definition_revision = ${revision} WHERE type = ${item.key}`)
  }
  if ((payload.typeDefinitions ?? []).length > 0) restoredCounts.typeDefinitions = payload.typeDefinitions!.length

  for (const item of payload.statusDefinitions ?? []) {
    const current = statusCurrents.get(item.key)!
    const next = normalizeContentStatusDefinition(item.definition)
    const revision = current.revision + 1
    const version = current.version + 1
    const hash = await canonicalContentMigrationHash(next)
    await writeDefinitionVersion(db, { kind: 'status', key: item.key, revision, canonicalHash: hash, definition: next as unknown as ContentSchemaValue, origin: 'db' })
    const updated = await db.execute<{ key: string }>(sql`UPDATE content_status_definitions SET version = ${version}, current_revision = ${revision}, active = ${item.active}, canonical_hash = ${hash}, definition = ${JSON.stringify(next)}, updated_at = ${now} WHERE key = ${item.key} AND version = ${current.version} RETURNING key`)
    if (updated.length !== 1) throw new ContentDefinitionError('stale-impact', `status definition ${item.key} changed during restore`)
    await db.execute(sql`UPDATE content_entries SET status_definition_revision = ${revision} WHERE status = ${item.key}`)
  }
  if ((payload.statusDefinitions ?? []).length > 0) restoredCounts.statusDefinitions = payload.statusDefinitions!.length

  const result: ContentBackupRestoreResult = Object.freeze({ operationId: input.operationId, manifestId: verified.id, destinationCorpusHash: input.destinationCorpusHash, restoredCounts: Object.freeze(restoredCounts) })
  await writeDefinitionReceipt(db, { operationId: input.operationId, kind: 'definition:restore', payload: { requestHash, principalScope: scope, result: result as unknown as ContentSchemaValue } })
  const event: ContentDefinitionEvent = {
    id: crypto.randomUUID(), version: 1, kind: 'backupRestored', operationId: input.operationId,
    principalId: principal.id, manifestId: verified.id, destinationCorpusHash: input.destinationCorpusHash,
    restoredCounts: result.restoredCounts, authorizationPolicyVersion: policyVersion,
  }
  await effects.audit.record(db, event)
  await effects.outbox.enqueue(db, event)
  return result
}