import { sql } from 'drizzle-orm'
import { Miniflare } from 'miniflare'
import { describe, expect, it } from 'vitest'
import { createD1Client } from '@platform-modules/db/sqlite/d1'
import type { FieldGroup as LegacyFieldGroup } from './model.js'
import { makePgHarness } from './pg-harness.js'
import {
  canonicalFieldsMigrationHash,
  escapeFieldPathSegment,
  fieldsRecursiveEavForwardMigrationSql,
  fieldsRecursiveEavReverseMigrationSql,
  parseLegacyMultipleValueJson,
  planFieldsRecursiveEavMigration,
  snapshotFlatFieldsCorpus,
  type FieldsRecursiveEavMigrationPlan,
  type MigratedFieldValueNode,
} from './migrations/recursive-eav.js'
import { fieldsSchema, type FlatFieldValue } from './schema.js'
import { readFlatFieldValues } from './store.js'

const IDS = {
  title: '11111111-1111-4111-8111-111111111111',
  count: '22222222-2222-4222-8222-222222222222',
  related: '33333333-3333-4333-8333-333333333333',
  size40: '44444444-4444-4444-8444-444444444444',
  size41: '55555555-5555-4555-8555-555555555555',
} as const
const CREATED = new Date('2026-08-01T12:00:00.000Z')
const UPDATED = new Date('2026-08-07T12:00:00.000Z')
const NOW = '2026-08-08T15:00:00.000Z'

const group = {
  key: 'legacy_group', label: 'Legacy group', location: { entityType: 'content' },
  fields: [
    { type: 'text', key: 'title', label: 'Title' },
    { type: 'number', key: 'count', label: 'Count' },
    { type: 'relationship', key: 'related', label: 'Related', targetEntityType: 'content' },
    { type: 'select', key: 'sizes', label: 'Sizes', multiple: true, options: [
      { value: '40', label: '40' }, { value: '41', label: '41' },
    ] },
  ],
} satisfies LegacyFieldGroup

const corpus: readonly FlatFieldValue[] = Object.freeze([
  Object.freeze({ id: IDS.title, entityType: 'content', entityId: 'entry-1', groupId: 'legacy_group', fieldKey: 'title', ordinal: 0, valueText: 'Preserved', valueNum: null, valueBool: null, valueDate: null, refType: null, refId: null, refMeta: null, createdAt: CREATED, updatedAt: UPDATED }),
  Object.freeze({ id: IDS.count, entityType: 'content', entityId: 'entry-1', groupId: 'legacy_group', fieldKey: 'count', ordinal: 0, valueText: null, valueNum: '42.50', valueBool: null, valueDate: null, refType: null, refId: null, refMeta: null, createdAt: CREATED, updatedAt: UPDATED }),
  Object.freeze({ id: IDS.related, entityType: 'content', entityId: 'entry-1', groupId: 'legacy_group', fieldKey: 'related', ordinal: 0, valueText: null, valueNum: null, valueBool: null, valueDate: null, refType: 'content', refId: 'entry-2', refMeta: { label: 'Related entry' }, createdAt: CREATED, updatedAt: UPDATED }),
  Object.freeze({ id: IDS.size40, entityType: 'content', entityId: 'entry-1', groupId: 'legacy_group', fieldKey: 'sizes', ordinal: 0, valueText: '40', valueNum: null, valueBool: null, valueDate: null, refType: null, refId: null, refMeta: null, createdAt: CREATED, updatedAt: UPDATED }),
  Object.freeze({ id: IDS.size41, entityType: 'content', entityId: 'entry-1', groupId: 'legacy_group', fieldKey: 'sizes', ordinal: 1, valueText: '41', valueNum: null, valueBool: null, valueDate: null, refType: null, refId: null, refMeta: null, createdAt: CREATED, updatedAt: UPDATED }),
])

interface D1Statement { bind(...values: unknown[]): D1Statement; run(): Promise<unknown>; all(): Promise<{ results: unknown[] }> }
interface D1Binding { exec(query: string): Promise<unknown>; prepare(query: string): D1Statement }

function execD1(binding: D1Binding, statement: string): Promise<unknown> {
  return binding.exec(statement.replace(/\s+/g, ' ').trim())
}

const D1_DDL = `CREATE TABLE field_values (
  id TEXT PRIMARY KEY, entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, group_id TEXT NOT NULL,
  field_key TEXT NOT NULL, ordinal INTEGER NOT NULL, value_text TEXT, value_num TEXT, value_bool INTEGER,
  value_date TEXT, ref_type TEXT, ref_id TEXT, ref_meta TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL
);`
const PG_DDL = `CREATE TABLE field_values (
  id uuid PRIMARY KEY, entity_type text NOT NULL, entity_id text NOT NULL, group_id text NOT NULL,
  field_key text NOT NULL, ordinal integer NOT NULL, value_text text, value_num numeric, value_bool boolean,
  value_date timestamptz(3), ref_type text, ref_id text, ref_meta jsonb,
  created_at timestamptz(3) NOT NULL, updated_at timestamptz(3) NOT NULL
);`

function rowArgs(row: FlatFieldValue): unknown[] {
  return [row.id, row.entityType, row.entityId, row.groupId, row.fieldKey, row.ordinal, row.valueText, row.valueNum,
    row.valueBool, row.valueDate?.toISOString() ?? null, row.refType, row.refId,
    row.refMeta === null ? null : JSON.stringify(row.refMeta), row.createdAt.toISOString(), row.updatedAt.toISOString()]
}

function normalizeNodeRow(row: Record<string, unknown>): MigratedFieldValueNode {
  const sourceIds = typeof row.legacySourceIds === 'string' ? JSON.parse(row.legacySourceIds) as string[] : row.legacySourceIds as string[]
  return Object.freeze({
    nodeId: String(row.nodeId), entityType: String(row.entityType), entityId: String(row.entityId), groupKey: String(row.groupKey),
    definitionRevision: Number(row.definitionRevision), fieldKey: String(row.fieldKey), path: String(row.path),
    parentNodeId: row.parentNodeId === null ? null : String(row.parentNodeId), rowId: row.rowId === null ? null : String(row.rowId),
    layoutKey: row.layoutKey === null ? null : String(row.layoutKey), nodeKind: String(row.nodeKind) as MigratedFieldValueNode['nodeKind'],
    ordinal: Number(row.ordinal), isNull: row.isNull === true || row.isNull === 1,
    valueText: row.valueText === null ? null : String(row.valueText), valueNumber: row.valueNumber === null ? null : Number(row.valueNumber),
    valueBoolean: row.valueBoolean === null ? null : row.valueBoolean === true || row.valueBoolean === 1,
    valueDateTime: row.valueDateTime === null ? null : String(row.valueDateTime), valueRef: row.valueRef === null ? null : String(row.valueRef),
    valueJson: row.valueJson === null ? null : String(row.valueJson), legacySourceIds: Object.freeze(sourceIds),
    legacySourceHash: String(row.legacySourceHash),
  })
}

const NODE_SELECT = `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", legacy_source_ids AS "legacySourceIds", legacy_source_hash AS "legacySourceHash"
  FROM field_value_nodes ORDER BY entity_type, entity_id, group_key, ordinal`

async function persistD1Plan(binding: D1Binding, plan: FieldsRecursiveEavMigrationPlan): Promise<void> {
  for (const definition of plan.definitions) {
    await binding.prepare(`INSERT INTO field_group_definitions
      (key, origin, version, active, current_revision, canonical_hash, definition, created_at, updated_at)
      VALUES (?, ?, 1, 1, 1, ?, ?, ?, ?)`)
      .bind(definition.groupKey, definition.origin, definition.canonicalHash, JSON.stringify(definition.definition), NOW, NOW).run()
    await binding.prepare(`INSERT INTO field_definition_versions
      (group_key, revision, canonical_hash, definition, origin, created_at) VALUES (?, 1, ?, ?, ?, ?)`)
      .bind(definition.groupKey, definition.canonicalHash, JSON.stringify(definition.definition), definition.origin, NOW).run()
  }
  for (const node of plan.nodes) {
    await binding.prepare(`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, legacy_source_ids, legacy_source_hash, created_at, updated_at)
      VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
      .bind(node.nodeId, node.entityType, node.entityId, node.groupKey, node.definitionRevision, node.fieldKey, node.path,
        node.parentNodeId, node.rowId, node.layoutKey, node.nodeKind, node.ordinal, node.isNull ? 1 : 0,
        node.valueText, node.valueNumber, node.valueBoolean === null ? null : node.valueBoolean ? 1 : 0,
        node.valueDateTime, node.valueRef, node.valueJson, JSON.stringify(node.legacySourceIds), node.legacySourceHash, NOW, NOW).run()
  }
}

async function persistPgPlan(db: Awaited<ReturnType<typeof makePgHarness>>['db'], plan: FieldsRecursiveEavMigrationPlan): Promise<void> {
  for (const definition of plan.definitions) {
    const json = JSON.stringify(definition.definition).replaceAll("'", "''")
    await db.execute(sql.raw(`INSERT INTO field_group_definitions
      (key, origin, version, active, current_revision, canonical_hash, definition, created_at, updated_at)
      VALUES ('${definition.groupKey}', '${definition.origin}', 1, true, 1, '${definition.canonicalHash}', '${json}'::jsonb, '${NOW}', '${NOW}')`))
    await db.execute(sql.raw(`INSERT INTO field_definition_versions
      (group_key, revision, canonical_hash, definition, origin, created_at)
      VALUES ('${definition.groupKey}', 1, '${definition.canonicalHash}', '${json}'::jsonb, '${definition.origin}', '${NOW}')`))
  }
  for (const node of plan.nodes) {
    await db.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, legacy_source_ids, legacy_source_hash, created_at, updated_at)
      VALUES (${node.nodeId}::uuid, ${node.entityType}, ${node.entityId}, ${node.groupKey}, ${node.definitionRevision},
       ${node.fieldKey}, ${node.path}, ${node.parentNodeId}::uuid, ${node.rowId}, ${node.layoutKey}, ${node.nodeKind},
       ${node.ordinal}, ${node.isNull}, ${node.valueText}, ${node.valueNumber}, ${node.valueBoolean},
       ${node.valueDateTime}, ${node.valueRef}, ${node.valueJson}, ${JSON.stringify(node.legacySourceIds)}::jsonb,
       ${node.legacySourceHash}, ${NOW}::timestamptz, ${NOW}::timestamptz)`)
  }
}

async function d1Evidence(plan: FieldsRecursiveEavMigrationPlan) {
  const mf = new Miniflare({ modules: true, script: `export default { fetch() { return new Response('ok') } }`, d1Databases: { DB: '00000000-0000-4000-8000-000000000022' } })
  try {
    const binding = await mf.getD1Database('DB') as unknown as D1Binding
    await execD1(binding, D1_DDL)
    for (const row of corpus) {
      await binding.prepare(`INSERT INTO field_values
        (id,entity_type,entity_id,group_id,field_key,ordinal,value_text,value_num,value_bool,value_date,ref_type,ref_id,ref_meta,created_at,updated_at)
        VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`).bind(...rowArgs(row)).run()
    }
    const client = createD1Client(binding as never, { schema: fieldsSchema })
    const before = await readFlatFieldValues({ adapter: 'd1', execute: client.execute })
    expect(await snapshotFlatFieldsCorpus(before)).toEqual(plan.source)
    for (const statement of fieldsRecursiveEavForwardMigrationSql('d1')) await execD1(binding, statement)
    await persistD1Plan(binding, plan)
    const nodeResult = await binding.prepare(NODE_SELECT).all()
    const nodes = nodeResult.results.map((row) => normalizeNodeRow(row as Record<string, unknown>))
    expect(await canonicalFieldsMigrationHash(nodes)).toBe(plan.target.hash)
    await expect(binding.prepare(`INSERT INTO field_value_nodes
      (node_id,entity_type,entity_id,group_key,definition_revision,field_key,path,node_kind,ordinal,is_null,value_text,value_number,created_at,updated_at)
      VALUES ('77777777-7777-4777-8777-777777777777','content','bad','legacy_group',1,'bad','/f:bad','leaf',0,0,'x','1',?,?)`).bind(NOW, NOW).run()).rejects.toThrow()
    await binding.prepare(`UPDATE field_values SET value_text = 'Changed' WHERE id = ?`).bind(IDS.title).run()
    await binding.prepare(`UPDATE field_values SET value_text = 'Preserved' WHERE id = ?`).bind(IDS.title).run()
    const journal = await binding.prepare(`SELECT seq, operation, value_id AS valueId FROM fields_migration_write_journal ORDER BY seq`).all()
    expect(journal.results).toHaveLength(2)
    const highWaterMark = String((journal.results.at(-1) as Record<string, unknown>).seq)
    for (const statement of fieldsRecursiveEavReverseMigrationSql('d1')) await execD1(binding, statement)
    const reversed = await readFlatFieldValues({ adapter: 'd1', execute: client.execute })
    return { source: await snapshotFlatFieldsCorpus(reversed), highWaterMark, nodes }
  } finally {
    await mf.dispose()
  }
}

async function pgEvidence(plan: FieldsRecursiveEavMigrationPlan) {
  const pg = await makePgHarness()
  try {
    await pg.db.execute(sql.raw(PG_DDL))
    for (const row of corpus) {
      const refMeta = row.refMeta === null ? null : JSON.stringify(row.refMeta)
      await pg.db.execute(sql`INSERT INTO field_values
        (id,entity_type,entity_id,group_id,field_key,ordinal,value_text,value_num,value_bool,value_date,ref_type,ref_id,ref_meta,created_at,updated_at)
        VALUES (${row.id}::uuid,${row.entityType},${row.entityId},${row.groupId},${row.fieldKey},${row.ordinal},${row.valueText},${row.valueNum},${row.valueBool},${row.valueDate},${row.refType},${row.refId},${refMeta}::jsonb,${row.createdAt},${row.updatedAt})`)
    }
    const before = await pg.db.transaction((tx) => readFlatFieldValues({ adapter: 'postgres', execute: tx.execute }))
    expect(await snapshotFlatFieldsCorpus(before)).toEqual(plan.source)
    for (const statement of fieldsRecursiveEavForwardMigrationSql('postgres')) await pg.db.execute(sql.raw(statement))
    await persistPgPlan(pg.db, plan)
    const rawNodes = await pg.db.transaction((tx) => tx.execute(sql.raw(NODE_SELECT))) as unknown as Record<string, unknown>[]
    const nodes = rawNodes.map(normalizeNodeRow)
    expect(await canonicalFieldsMigrationHash(nodes)).toBe(plan.target.hash)
    await expect(pg.db.execute(sql.raw(`INSERT INTO field_value_nodes
      (node_id,entity_type,entity_id,group_key,definition_revision,field_key,path,node_kind,ordinal,is_null,value_text,value_number)
      VALUES ('77777777-7777-4777-8777-777777777777','content','bad','legacy_group',1,'bad','/f:bad','leaf',0,false,'x',1)`))).rejects.toThrow()
    await pg.db.execute(sql`UPDATE field_values SET value_text = 'Changed' WHERE id = ${IDS.title}::uuid`)
    await pg.db.execute(sql`UPDATE field_values SET value_text = 'Preserved' WHERE id = ${IDS.title}::uuid`)
    const journal = await pg.db.transaction((tx) => tx.execute(sql.raw(`SELECT seq, operation, value_id AS "valueId" FROM fields_migration_write_journal ORDER BY seq`))) as unknown as Record<string, unknown>[]
    expect(journal).toHaveLength(2)
    const highWaterMark = String(journal.at(-1)?.seq)
    for (const statement of fieldsRecursiveEavReverseMigrationSql('postgres')) await pg.db.execute(sql.raw(statement))
    const reversed = await pg.db.transaction((tx) => readFlatFieldValues({ adapter: 'postgres', execute: tx.execute }))
    return { source: await snapshotFlatFieldsCorpus(reversed), highWaterMark, nodes }
  } finally {
    await pg.teardown()
  }
}

describe('recursive fields EAV migration', () => {
  it('escapes canonical field paths and collapses legal legacy multi-values into one typed JSON leaf', async () => {
    expect(escapeFieldPathSegment('a~/b')).toBe('a~0~1b')
    const plan = await planFieldsRecursiveEavMigration(corpus, [group])
    expect(plan.source.count).toBe(5)
    expect(plan.target.count).toBe(4)
    expect(plan.definitionRefs).toEqual([{ groupKey: 'legacy_group', revision: 1 }])
    const sizes = plan.nodes.find((node) => node.fieldKey === 'sizes')!
    expect(sizes.path).toBe('/f:sizes')
    expect(sizes.ordinal).toBe(3)
    expect(sizes.legacySourceIds).toEqual([IDS.size40, IDS.size41])
    expect(parseLegacyMultipleValueJson(sizes.valueJson!)).toEqual([
      { legacyId: IDS.size40, ordinal: 0, lane: 'text', value: '40' },
      { legacyId: IDS.size41, ordinal: 1, lane: 'text', value: '41' },
    ])
    expect(plan.nodes.find((node) => node.fieldKey === 'count')).toMatchObject({ valueNumber: 42.5 })
    expect(plan.nodes.find((node) => node.fieldKey === 'related')?.valueRef).toContain('entry-2')
  })

  it('fails closed when a value cannot resolve its historical group/field definition', async () => {
    await expect(planFieldsRecursiveEavMigration(corpus, [])).rejects.toThrow('references missing group')
    const incomplete = { ...group, fields: group.fields.filter((field) => field.key !== 'title') }
    await expect(planFieldsRecursiveEavMigration(corpus, [incomplete])).rejects.toThrow('references missing historical field')
  })

  it('proves equivalent D1/Postgres nodes, journal high-water marks, constraints, and reverse legacy parity', async () => {
    const plan = await planFieldsRecursiveEavMigration(corpus, [group])
    const [d1, pg] = await Promise.all([d1Evidence(plan), pgEvidence(plan)])
    expect(d1.nodes).toEqual(plan.nodes)
    expect(pg.nodes).toEqual(plan.nodes)
    expect(d1.source).toEqual(plan.source)
    expect(pg.source).toEqual(plan.source)
    expect(Number(d1.highWaterMark)).toBeGreaterThan(0)
    expect(Number(pg.highWaterMark)).toBeGreaterThan(0)
  }, 90_000)
})