import { sql } from 'drizzle-orm'
import { Miniflare } from 'miniflare'
import { describe, expect, it } from 'vitest'
import { createD1Client } from '@platform-modules/db/sqlite/d1'
import { startPg } from './pg-harness.js'
import {
  canonicalContentMigrationHash,
  contentModelForwardMigrationSql,
  contentModelReverseMigrationSql,
  planContentModelMigration,
  snapshotFlatContentCorpus,
  type ContentMigrationEntryProjection,
  type ContentModelMigrationPlan,
} from './migrations/content-model.js'
import { contentSchema, type FlatContentEntry } from './schema.js'
import { readFlatContentEntries } from './store.js'


interface TestD1Statement {
  bind(...values: unknown[]): TestD1Statement
  run(): Promise<unknown>
  all(): Promise<{ results: unknown[] }>
}

interface TestD1Binding {
  prepare(query: string): TestD1Statement
  exec(query: string): Promise<unknown>
}

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

const ENTRY_A = '11111111-1111-4111-8111-111111111111'
const ENTRY_B = '22222222-2222-4222-8222-222222222222'
const NOW = '2026-08-08T15:00:00.000Z'

const CORPUS: readonly FlatContentEntry[] = Object.freeze([
  Object.freeze({
    id: ENTRY_A,
    slug: 'legacy-story',
    type: 'legacy_story',
    title: 'Legacy story',
    body: '<p>Preserve — exactly.</p>',
    status: 'legacy_review',
    visibility: 'members',
    publishedAt: null,
    author: 'author-a',
    createdAt: new Date('2026-08-01T12:00:00.000Z'),
    updatedAt: new Date('2026-08-07T12:00:00.000Z'),
  }),
  Object.freeze({
    id: ENTRY_B,
    slug: 'event-one',
    type: 'custom-event',
    title: 'Event one',
    body: '<p>Published body</p>',
    status: 'published',
    visibility: 'public',
    publishedAt: new Date('2026-08-06T09:30:00.000Z'),
    author: 'author-b',
    createdAt: new Date('2026-08-02T12:00:00.000Z'),
    updatedAt: new Date('2026-08-06T09:30:00.000Z'),
  }),
])

const LEGACY_D1_DDL = `CREATE TABLE content_entries (
  id TEXT PRIMARY KEY, slug TEXT NOT NULL, type TEXT NOT NULL, title TEXT NOT NULL,
  body TEXT NOT NULL, status TEXT NOT NULL, visibility TEXT NOT NULL,
  published_at TEXT, author TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL
)`

const LEGACY_PG_DDL = `CREATE TABLE content_entries (
  id uuid PRIMARY KEY, slug text NOT NULL, type text NOT NULL, title text NOT NULL,
  body text NOT NULL, status text NOT NULL, visibility text NOT NULL,
  published_at timestamptz(3), author text NOT NULL,
  created_at timestamptz(3) NOT NULL, updated_at timestamptz(3) NOT NULL
)`

function targetFromFinalRows(
  legacy: readonly FlatContentEntry[],
  finalRows: readonly Record<string, unknown>[],
): readonly ContentMigrationEntryProjection[] {
  const byId = new Map(finalRows.map((row) => [String(row.id), row]))
  return Object.freeze([...legacy]
    .sort((a, b) => a.id.localeCompare(b.id))
    .map((entry) => {
      const row = byId.get(entry.id)
      if (!row) throw new Error(`missing final row ${entry.id}`)
      return Object.freeze({
        ...entry,
        parentId: row.parentId === null ? null : String(row.parentId),
        menuOrder: Number(row.menuOrder),
        templateKey: row.templateKey === null ? null : String(row.templateKey),
        excerpt: String(row.excerpt),
        featuredMedia: row.featuredMedia === null ? null : row.featuredMedia as { id: string; kind?: string },
        commentStatus: String(row.commentStatus) as 'open' | 'closed',
        pingStatus: String(row.pingStatus) as 'open' | 'closed',
        sticky: row.sticky === true || row.sticky === 1,
        format: row.format === null ? null : String(row.format),
        deletedAt: row.deletedAt === null ? null : new Date(String(row.deletedAt)),
        lastEditedBy: String(row.lastEditedBy),
        typeDefinitionRevision: Number(row.typeDefinitionRevision),
        statusDefinitionRevision: Number(row.statusDefinitionRevision),
      })
    }))
}

async function targetHash(entries: readonly ContentMigrationEntryProjection[]): Promise<string> {
  return canonicalContentMigrationHash(entries.map((entry) => ({
    id: entry.id,
    slug: entry.slug,
    type: entry.type,
    title: entry.title,
    body: entry.body,
    status: entry.status,
    visibility: entry.visibility,
    publishedAt: entry.publishedAt,
    author: entry.author,
    createdAt: entry.createdAt,
    updatedAt: entry.updatedAt,
    parentId: entry.parentId,
    menuOrder: entry.menuOrder,
    templateKey: entry.templateKey,
    excerpt: entry.excerpt,
    featuredMedia: entry.featuredMedia,
    commentStatus: entry.commentStatus,
    pingStatus: entry.pingStatus,
    sticky: entry.sticky,
    format: entry.format,
    deletedAt: entry.deletedAt,
    lastEditedBy: entry.lastEditedBy,
    typeDefinitionRevision: entry.typeDefinitionRevision,
    statusDefinitionRevision: entry.statusDefinitionRevision,
  })))
}

async function persistD1Plan(binding: TestD1Binding, plan: ContentModelMigrationPlan): Promise<void> {
  for (const definition of plan.definitions) {
    const table = definition.kind === 'type' ? 'content_type_definitions' : 'content_status_definitions'
    await binding.prepare(`INSERT INTO ${table}
      (key, origin, version, active, current_revision, canonical_hash, definition, created_at, updated_at)
      VALUES (?, ?, 1, 1, 1, ?, ?, ?, ?)`)
      .bind(definition.key, definition.origin, definition.canonicalHash, JSON.stringify(definition.definition), NOW, NOW)
      .run()
    await binding.prepare(`INSERT INTO content_definition_versions
      (definition_kind, definition_key, revision, canonical_hash, definition, origin, created_at)
      VALUES (?, ?, 1, ?, ?, ?, ?)`)
      .bind(definition.kind, definition.key, definition.canonicalHash, JSON.stringify(definition.definition), definition.origin, NOW)
      .run()
  }
}

async function persistPgPlan(db: Awaited<ReturnType<typeof startPg>>['db'], plan: ContentModelMigrationPlan): Promise<void> {
  for (const definition of plan.definitions) {
    const table = definition.kind === 'type' ? 'content_type_definitions' : 'content_status_definitions'
    const payload = JSON.stringify(definition.definition)
    await db.execute(sql.raw(`INSERT INTO ${table}
      (key, origin, version, active, current_revision, canonical_hash, definition, created_at, updated_at)
      VALUES ('${definition.key.replaceAll("'", "''")}', '${definition.origin}', 1, true, 1,
        '${definition.canonicalHash}', '${payload.replaceAll("'", "''")}'::jsonb, '${NOW}', '${NOW}')`))
    await db.execute(sql.raw(`INSERT INTO content_definition_versions
      (definition_kind, definition_key, revision, canonical_hash, definition, origin, created_at)
      VALUES ('${definition.kind}', '${definition.key.replaceAll("'", "''")}', 1, '${definition.canonicalHash}',
        '${payload.replaceAll("'", "''")}'::jsonb, '${definition.origin}', '${NOW}')`))
  }
}

async function d1Evidence(plan: ContentModelMigrationPlan) {
  const mf = new Miniflare({
    modules: true,
    script: `export default { fetch() { return new Response('ok') } }`,
    d1Databases: { DB: '00000000-0000-4000-8000-000000000021' },
  })
  try {
    const binding = await mf.getD1Database('DB') as unknown as TestD1Binding
    await execD1(binding, LEGACY_D1_DDL)
    for (const entry of CORPUS) {
      await binding.prepare(`INSERT INTO content_entries
        (id, slug, type, title, body, status, visibility, published_at, author, created_at, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
        .bind(
          entry.id, entry.slug, entry.type, entry.title, entry.body, entry.status, entry.visibility,
          entry.publishedAt?.toISOString() ?? null, entry.author, entry.createdAt.toISOString(), entry.updatedAt.toISOString(),
        ).run()
    }
    const client = createD1Client(binding, { schema: contentSchema })
    const before = await readFlatContentEntries({ adapter: 'd1', execute: client.execute })
    expect(await snapshotFlatContentCorpus(before)).toEqual(plan.source)

    for (const statement of contentModelForwardMigrationSql('d1')) await execD1(binding, statement)
    await persistD1Plan(binding, plan)

    const final = await binding.prepare(`SELECT id,
      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, last_edited_by AS lastEditedBy,
      type_definition_revision AS typeDefinitionRevision,
      status_definition_revision AS statusDefinitionRevision
      FROM content_entries ORDER BY id`).all()
    const finalRows = final.results as Record<string, unknown>[]
    expect(await targetHash(targetFromFinalRows(before, finalRows))).toBe(plan.target.hash)

    await binding.prepare(`UPDATE content_entries SET title = ? WHERE id = ?`).bind('Journalled change', ENTRY_A).run()
    await binding.prepare(`UPDATE content_entries SET title = ? WHERE id = ?`).bind('Legacy story', ENTRY_A).run()
    const journal = await binding.prepare(`SELECT seq, operation, entry_id AS entryId, before_value AS beforeValue,
      after_value AS afterValue FROM content_migration_write_journal ORDER BY seq`).all()
    const journalRows = journal.results as Array<Record<string, unknown>>
    expect(journalRows).toHaveLength(2)
    expect(journalRows[0]).toMatchObject({ operation: 'update', entryId: ENTRY_A })
    expect(JSON.parse(String(journalRows[0]?.afterValue))).toMatchObject({ title: 'Journalled change', type: 'legacy_story' })
    const highWaterMark = String(journalRows.at(-1)?.seq)

    const definitions = await binding.prepare(`SELECT definition_kind AS kind, definition_key AS key, canonical_hash AS canonicalHash
      FROM content_definition_versions ORDER BY definition_kind, definition_key`).all()

    for (const statement of contentModelReverseMigrationSql('d1')) await execD1(binding, statement)
    const reversed = await readFlatContentEntries({ adapter: 'd1', execute: client.execute })
    const reverseSnapshot = await snapshotFlatContentCorpus(reversed)
    expect(reverseSnapshot).toEqual(plan.source)

    return {
      ids: reversed.map((entry) => entry.id),
      sourceHash: plan.source.hash,
      targetHash: plan.target.hash,
      reverseHash: reverseSnapshot.hash,
      highWaterMark,
      definitions: definitions.results,
    }
  } finally {
    await mf.dispose()
  }
}

async function pgEvidence(plan: ContentModelMigrationPlan) {
  const pg = await startPg()
  try {
    await pg.db.execute(sql.raw(LEGACY_PG_DDL))
    for (const entry of CORPUS) {
      await pg.db.execute(sql`INSERT INTO content_entries
        (id, slug, type, title, body, status, visibility, published_at, author, created_at, updated_at)
        VALUES (${entry.id}::uuid, ${entry.slug}, ${entry.type}, ${entry.title}, ${entry.body}, ${entry.status}, ${entry.visibility},
          ${entry.publishedAt}, ${entry.author}, ${entry.createdAt}, ${entry.updatedAt})`)
    }
    const before = await pg.db.transaction((tx) => readFlatContentEntries({ adapter: 'postgres', execute: tx.execute }))
    expect(await snapshotFlatContentCorpus(before)).toEqual(plan.source)

    for (const statement of contentModelForwardMigrationSql('postgres')) await pg.db.execute(sql.raw(statement))
    await persistPgPlan(pg.db, plan)

    const finalRows = await pg.db.transaction((tx) => tx.execute(sql.raw(`SELECT id::text AS id,
      parent_id::text 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", last_edited_by AS "lastEditedBy",
      type_definition_revision AS "typeDefinitionRevision",
      status_definition_revision AS "statusDefinitionRevision"
      FROM content_entries ORDER BY id`))) as unknown as Record<string, unknown>[]
    expect(await targetHash(targetFromFinalRows(before, finalRows))).toBe(plan.target.hash)

    await pg.db.execute(sql`UPDATE content_entries SET title = 'Journalled change' WHERE id = ${ENTRY_A}::uuid`)
    await pg.db.execute(sql`UPDATE content_entries SET title = 'Legacy story' WHERE id = ${ENTRY_A}::uuid`)
    const journalRows = await pg.db.transaction((tx) => tx.execute(sql.raw(`SELECT seq, operation, entry_id AS "entryId", before_value AS "beforeValue",
      after_value AS "afterValue" FROM content_migration_write_journal ORDER BY seq`))) as unknown as Array<Record<string, unknown>>
    expect(journalRows).toHaveLength(2)
    expect(journalRows[0]).toMatchObject({ operation: 'update', entryId: ENTRY_A })
    expect(journalRows[0]?.afterValue).toMatchObject({ title: 'Journalled change', type: 'legacy_story' })
    const highWaterMark = String(journalRows.at(-1)?.seq)

    const definitions = await pg.db.transaction((tx) => tx.execute(sql.raw(`SELECT definition_kind AS kind, definition_key AS key, canonical_hash AS "canonicalHash"
      FROM content_definition_versions ORDER BY definition_kind, definition_key`))) as unknown as Array<Record<string, unknown>>

    for (const statement of contentModelReverseMigrationSql('postgres')) await pg.db.execute(sql.raw(statement))
    const reversed = await pg.db.transaction((tx) => readFlatContentEntries({ adapter: 'postgres', execute: tx.execute }))
    const reverseSnapshot = await snapshotFlatContentCorpus(reversed)
    expect(reverseSnapshot).toEqual(plan.source)

    return {
      ids: reversed.map((entry) => entry.id),
      sourceHash: plan.source.hash,
      targetHash: plan.target.hash,
      reverseHash: reverseSnapshot.hash,
      highWaterMark,
      definitions,
    }
  } finally {
    await pg.stop()
  }
}

describe('content-model lossless migration', () => {
  it('materializes every arbitrary type/status without coercion and assigns immutable revision 1', async () => {
    const plan = await planContentModelMigration(CORPUS, { defaultTemplateKey: 'generic' })
    expect(plan.source.count).toBe(2)
    expect(plan.target.count).toBe(2)
    expect(plan.definitions).toEqual(expect.arrayContaining([
      expect.objectContaining({ kind: 'type', key: 'legacy_story', origin: 'db', revision: 1 }),
      expect.objectContaining({ kind: 'type', key: 'custom-event', origin: 'db', revision: 1 }),
      expect.objectContaining({ kind: 'status', key: 'legacy_review', origin: 'db', revision: 1 }),
      expect.objectContaining({ kind: 'status', key: 'pending', origin: 'code', revision: 1 }),
    ]))
    expect(plan.entries.find((entry) => entry.id === ENTRY_A)).toMatchObject({
      type: 'legacy_story',
      status: 'legacy_review',
      lastEditedBy: 'author-a',
      typeDefinitionRevision: 1,
      statusDefinitionRevision: 1,
    })
    expect(plan.definitions.every((definition) => /^[a-f0-9]{64}$/.test(definition.canonicalHash))).toBe(true)
  })

  it('proves equivalent D1/Postgres forward + journal high-water + reverse parity', async () => {
    const plan = await planContentModelMigration(CORPUS, { defaultTemplateKey: 'generic' })
    const [d1, pg] = await Promise.all([d1Evidence(plan), pgEvidence(plan)])
    expect(d1.ids).toEqual([ENTRY_A, ENTRY_B])
    expect(pg.ids).toEqual(d1.ids)
    expect(pg.sourceHash).toBe(d1.sourceHash)
    expect(pg.targetHash).toBe(d1.targetHash)
    expect(pg.reverseHash).toBe(d1.reverseHash)
    expect(d1.reverseHash).toBe(plan.source.hash)
    expect(Number(d1.highWaterMark)).toBeGreaterThan(0)
    expect(Number(pg.highWaterMark)).toBeGreaterThan(0)
    expect(pg.definitions).toEqual(d1.definitions)
  }, 90_000)
})