import { afterEach, describe, expect, it } from 'vitest'
import { sql, type SQLWrapper } from 'drizzle-orm'
import type { D1BatchItem, D1Client } from '@platform-modules/db/sqlite/d1'
import { startPg } from './pg-harness.js'
import { contentModelForwardMigrationSql, canonicalContentMigrationHash } from './migrations/content-model.js'
import { contentRevisionsMigrationSql } from './revisions.js'
import { contentTaxonomyMigrationSql } from './taxonomy.js'
import { bookType, admin } from './lifecycle.test-helpers.js'
import { normalizeContentStatusDefinition, type ContentStatusDefinition } from './status.js'
import { normalizeContentTypeDefinition } from './registry.js'
import type { ContentSchema } from './schema.js'
import {
  createDbContentImportJournal,
  parseContentImport,
  planContentImport,
  startContentImport,
  prepareStartContentImportD1,
  prepareResumeContentImportD1,
  prepareStagedContentImportGuardD1,
  preparePublishContentImportD1,
  prepareRollbackContentImportD1,
  prepareReversePublishedContentImportD1,
  resumeContentImport,
  publishContentImport,
  predictPublishedContentImportCorpus,
  rollbackContentImport,
  planPublishedContentImportReverse,
  reversePublishedContentImport,
  type ContentExportBundle,
  type ContentImportAuthorization,
  type ContentImportCommand,
  type ContentImportLifecycleEffects,
  type ContentImportLimits,
  type ContentImportPlan,
  type ContentImportSession,
} from './index.js'

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
)`
const fixed='2026-08-22T00:00:00.000Z'
const entryId='11111111-1111-4111-8111-111111111111'
const revisionId='22222222-2222-4222-8222-222222222222'
const limits:ContentImportLimits=Object.freeze({maxBytes:2_000_000,maxDefinitions:100,maxEntries:100,maxRevisions:100,batchSize:10})
const authorization:ContentImportAuthorization=Object.freeze({assert:async()=>({policyVersion:'policy-1'})})
const effects:ContentImportLifecycleEffects=Object.freeze({audit:{record:async()=>{}},outbox:{enqueue:async()=>{}}})
let stop:(()=>Promise<void>)|undefined
afterEach(async()=>{await stop?.();stop=undefined})

function draftStatus():ContentStatusDefinition{return {key:'draft',label:'Draft',published:false,internal:false,excludeFromSearch:true,publiclyQueryable:false,showInAdminAll:true,showInAdminStatusFilter:true,dateLabel:'lastModified',transitionInput:'none'}}
async function bundle():Promise<ContentExportBundle>{
  const type=normalizeContentTypeDefinition({...bookType(),statusKeys:['draft'],taxonomies:[]}),status=normalizeContentStatusDefinition(draftStatus()),typeHash=await canonicalContentMigrationHash(type),statusHash=await canonicalContentMigrationHash(status)
  const raw={version:1,generatedAt:fixed,types:[{key:'book',origin:'db',version:1,revision:1,active:true,canonicalHash:typeHash,definition:type,versions:[{revision:1,canonicalHash:typeHash,definition:type,origin:'db',createdAt:fixed}]}],statuses:[{key:'draft',origin:'db',version:1,revision:1,active:true,canonicalHash:statusHash,definition:status,versions:[{revision:1,canonicalHash:statusHash,definition:status,origin:'db',createdAt:fixed}]}],taxonomies:[],terms:[],entries:[{id:entryId,slug:'first-book',type:'book',title:'First Book',body:'Body',status:'draft',visibility:'public',publishedAt:null,author:'admin',createdAt:fixed,updatedAt:fixed,parentId:null,menuOrder:0,templateKey:'default',excerpt:'Excerpt',featuredMedia:null,commentStatus:'open',pingStatus:'closed',sticky:false,format:null,deletedAt:null,lastEditedBy:'admin',typeDefinitionRevision:1,statusDefinitionRevision:1,termIds:[]}],revisions:[{id:revisionId,entryId,seq:1,title:'First Book',body:'Body',slug:'first-book',type:'book',termIds:[],editor:'admin',createdAt:fixed}]}
  return JSON.parse(JSON.stringify(raw)) as ContentExportBundle
}
async function fixture(){const pg=await startPg();stop=pg.stop;await pg.db.execute(sql.raw(LEGACY_PG_DDL));for(const statement of contentModelForwardMigrationSql('postgres'))await pg.db.execute(sql.raw(statement));for(const statement of contentRevisionsMigrationSql().split(';').map((item)=>item.trim()).filter(Boolean))await pg.db.execute(sql.raw(statement));for(const statement of contentTaxonomyMigrationSql().split(';').map((item)=>item.trim()).filter(Boolean))await pg.db.execute(sql.raw(statement));return pg}
function command(plan:ContentImportPlan,session:ContentImportSession,operationId:string):ContentImportCommand{return Object.freeze({operationId,expectedPlanHash:plan.planHash,expectedManifestHash:plan.manifestHash,scope:plan.scope,requiredAuthority:plan.requiredAuthority,expectedAuthorizationPolicyVersion:plan.authorizationPolicyVersion,expectedJournalVersion:session.journalVersion,expectedSection:session.nextSection,expectedOffset:session.nextOffset})}
async function prepare(){const pg=await fixture(),manifest=await bundle(),journal=createDbContentImportJournal(),plan=await pg.db.transaction((tx)=>planContentImport(tx,authorization,admin,manifest,{scope:{tenantKey:'tenant-1',typeKeys:['book']},conflictPolicy:'mappedReplacement'}));expect(plan.conflicts).toEqual([]);const session=await pg.db.transaction((tx)=>startContentImport(tx,journal,authorization,admin,manifest,{plan,staging:{hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa1',visibility:'hiddenUntilPublish'},operationId:'start:1'}));expect(session.id).toBe('aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa1');return {pg,manifest,journal,plan,session}}
async function stageAll(pg:Awaited<ReturnType<typeof startPg>>,manifest:ContentExportBundle,journal:ReturnType<typeof createDbContentImportJournal>,plan:ContentImportPlan,start:ContentImportSession){let session=start,index=0;while(session.state!=='staged'){const current=session;session=await pg.db.transaction((tx)=>resumeContentImport(tx,journal,authorization,admin,manifest,current.id,command(plan,current,`resume:${++index}`)))}return session}
function oneWinner(results:PromiseSettledResult<ContentImportSession>[]):void{expect(results.filter((result)=>result.status==='fulfilled')).toHaveLength(1);expect(results.filter((result)=>result.status==='rejected')).toHaveLength(1)}
function firstRow<Row>(value:unknown):Row{const rows=Array.isArray(value)?value:(value as {rows?:unknown})?.rows;if(!Array.isArray(rows)||rows.length===0)throw new Error('expected a database row');return rows[0] as Row}

describe('native content import',()=>{
  it('prepares a D1 start without publishing a result or mutating before the host batch',async()=>{
    const pg=await fixture(),manifest=await bundle(),plan=await pg.db.transaction((tx)=>planContentImport(tx,authorization,admin,manifest,{scope:{tenantKey:'tenant-1',typeKeys:['book']},conflictPolicy:'mappedReplacement'})),prepared:SQLWrapper[]=[]
    const db={
      execute:<Row extends Record<string,unknown>>(query:SQLWrapper)=>pg.db.execute<Row>(query),
      prepare:<Row=unknown>(query:SQLWrapper)=>{prepared.push(query);return Object.freeze({}) as D1BatchItem<Row>},
    } as unknown as D1Client<ContentSchema>
    const contribution=await prepareStartContentImportD1(db,authorization,admin,manifest,{plan,staging:{hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa5',visibility:'hiddenUntilPublish'},operationId:'d1:start:1'})
    expect(contribution.items).toHaveLength(6);expect(prepared).toHaveLength(6);expect(contribution.result).toMatchObject({id:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa5',state:'validated',journalVersion:'0',nextSection:'type',nextOffset:0})
    expect(firstRow<{count:number}>(await pg.db.execute(sql`SELECT COUNT(*)::int AS count FROM content_import_journal`))).toMatchObject({count:0})
    expect(firstRow<{count:number}>(await pg.db.execute(sql`SELECT COUNT(*)::int AS count FROM content_lifecycle_journal`))).toMatchObject({count:0})
  })

  it('prepares a D1 resume as an all-or-nothing hidden-ledger contribution',async()=>{
    const {pg,manifest,plan,session}=await prepare(),stored=firstRow<{payload:unknown}>(await pg.db.execute(sql`SELECT payload FROM content_import_journal WHERE import_id=${session.id}`))!,payload=typeof stored.payload==='string'?JSON.parse(stored.payload):stored.payload,prepared:SQLWrapper[]=[]
    await pg.db.execute(sql`UPDATE content_import_journal SET payload=${JSON.stringify({...payload,plan})} WHERE import_id=${session.id}`)
    const db={execute:<Row extends Record<string,unknown>>(query:SQLWrapper)=>pg.db.execute<Row>(query),prepare:<Row=unknown>(query:SQLWrapper)=>{prepared.push(query);return Object.freeze({}) as D1BatchItem<Row>}} as unknown as D1Client<ContentSchema>
    const contribution=await prepareResumeContentImportD1(db,authorization,admin,manifest,session.id,command(plan,session,'d1:resume:1'))
    expect(contribution.items).toHaveLength(8);expect(prepared).toHaveLength(8);expect(contribution.result).toMatchObject({id:session.id,state:'applying',journalVersion:'1',appliedBatches:1,nextSection:'status',nextOffset:0})
    expect(firstRow<{version:number;state:string;count:number}>(await pg.db.execute(sql`SELECT version,state,(SELECT COUNT(*) FROM content_lifecycle_journal)::int AS count FROM content_import_journal WHERE import_id=${session.id}`))).toMatchObject({version:0,state:'validated',count:1})
    await expect(prepareResumeContentImportD1(db,authorization,admin,manifest,session.id,{...command(plan,session,'d1:resume:stale'),expectedOffset:1})).rejects.toMatchObject({code:'stale-journal'})
    expect(firstRow<{version:number;state:string}>(await pg.db.execute(sql`SELECT version,state FROM content_import_journal WHERE import_id=${session.id}`))).toMatchObject({version:0,state:'validated'})
  })

  it('prepares a deterministic read-only staged guard and rejects stale retained evidence',async()=>{
    const {pg,manifest,journal,plan,session:start}=await prepare(),initial=firstRow<{payload:unknown}>(await pg.db.execute(sql`SELECT payload FROM content_import_journal WHERE import_id=${start.id}`))!,initialPayload=typeof initial.payload==='string'?JSON.parse(initial.payload):initial.payload
    await pg.db.execute(sql`UPDATE content_import_journal SET payload=${JSON.stringify({...initialPayload,plan})} WHERE import_id=${start.id}`)
    const staged=await stageAll(pg,manifest,journal,plan,start),prepared:SQLWrapper[]=[]
    const db={execute:<Row extends Record<string,unknown>>(query:SQLWrapper)=>pg.db.execute<Row>(query),prepare:<Row=unknown>(query:SQLWrapper)=>{prepared.push(query);return Object.freeze({}) as D1BatchItem<Row>}} as unknown as D1Client<ContentSchema>
    const stagedCommand=command(plan,staged,'d1:staged-guard:unused')
    const first=await prepareStagedContentImportGuardD1(db,authorization,admin,staged.id,stagedCommand),second=await prepareStagedContentImportGuardD1(db,authorization,admin,staged.id,stagedCommand)
    expect(first.items).toHaveLength(1);expect(second.items).toHaveLength(1);expect(prepared).toHaveLength(2);expect(first.result).toEqual(staged);expect(second.result).toEqual(first.result)
    expect(firstRow<{state:string;version:number;receipts:number}>(await pg.db.execute(sql`SELECT state,version,(SELECT COUNT(*) FROM content_lifecycle_journal WHERE operation_id='d1:staged-guard:unused')::int AS receipts FROM content_import_journal WHERE import_id=${staged.id}`))).toMatchObject({state:'staged',version:Number(staged.journalVersion),receipts:0})
    const retained=firstRow<{payload:unknown}>(await pg.db.execute(sql`SELECT payload FROM content_import_journal WHERE import_id=${staged.id}`)),retainedPayload=typeof retained.payload==='string'?JSON.parse(retained.payload):retained.payload
    await pg.db.execute(sql`UPDATE content_import_journal SET payload=${JSON.stringify({...retainedPayload,plan:{...plan,manifestHash:'drifted'}})} WHERE import_id=${staged.id}`)
    await expect(prepareStagedContentImportGuardD1(db,authorization,admin,staged.id,stagedCommand)).rejects.toMatchObject({code:'integrity'})
  })

  it('authorizes a staged guard before journal lookup',async()=>{
    let reads=0
    const denied:ContentImportAuthorization=Object.freeze({assert:async()=>{throw new Error('denied')}})
    const db={execute:async()=>{reads++;return []},prepare:<Row=unknown>()=>Object.freeze({}) as D1BatchItem<Row>} as unknown as D1Client<ContentSchema>
    const fakePlan={planHash:'plan',manifestHash:'manifest',authorizationPolicyVersion:'policy-1',scope:{tenantKey:'tenant-1',typeKeys:['book']},requiredAuthority:[]} as unknown as ContentImportPlan
    const fakeSession={journalVersion:'1',nextSection:null,nextOffset:0} as ContentImportSession
    await expect(prepareStagedContentImportGuardD1(db,denied,admin,'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa9',command(fakePlan,fakeSession,'guard:denied'))).rejects.toMatchObject({code:'authorization'})
    expect(reads).toBe(0)
  })

  it('prepares D1 publication without exposing success or mutating before the host batch',async()=>{
    const {pg,manifest,journal,plan,session:start}=await prepare(),staged=await stageAll(pg,manifest,journal,plan,start),prepared:SQLWrapper[]=[]
    const db={execute:<Row extends Record<string,unknown>>(query:SQLWrapper)=>pg.db.execute<Row>(query),prepare:<Row=unknown>(query:SQLWrapper)=>{prepared.push(query);return Object.freeze({}) as D1BatchItem<Row>}} as unknown as D1Client<ContentSchema>
    const contribution=await preparePublishContentImportD1(db,authorization,admin,staged.id,command(plan,staged,'d1:publish:1'))
    expect(contribution.items.length).toBeGreaterThan(8);expect(prepared).toHaveLength(contribution.items.length);expect(contribution.result).toMatchObject({id:staged.id,state:'published',journalVersion:String(Number(staged.journalVersion)+1)});expect(contribution.result.publishedCorpusVersion).toMatch(/^[0-9a-f]{64}$/)
    expect(firstRow<{state:string;version:number;entries:number;receipts:number}>(await pg.db.execute(sql`SELECT state,version,(SELECT COUNT(*) FROM content_entries)::int AS entries,(SELECT COUNT(*) FROM content_lifecycle_journal WHERE operation_id='d1:publish:1')::int AS receipts FROM content_import_journal WHERE import_id=${staged.id}`))).toMatchObject({state:'staged',version:Number(staged.journalVersion),entries:0,receipts:0})
    await expect(preparePublishContentImportD1(db,authorization,admin,staged.id,{...command(plan,staged,'d1:publish:stale'),expectedJournalVersion:'999'})).rejects.toMatchObject({code:'stale-journal'})
  })

  it('prepares published reverse as an atomic retryable contribution without exposing success before commit',async()=>{
    const {pg,manifest,journal,plan,session:start}=await prepare(),staged=await stageAll(pg,manifest,journal,plan,start),published=await pg.db.transaction((tx)=>publishContentImport(tx,journal,authorization,effects,admin,staged.id,command(plan,staged,'d1:reverse:publish'))),reversePlan=await pg.db.transaction((tx)=>planPublishedContentImportReverse(tx,journal,authorization,admin,published.id,{scope:plan.scope,requiredAuthority:plan.requiredAuthority,expectedPolicyVersion:plan.authorizationPolicyVersion})),reverseCommand={...command(plan,published,'d1:reverse:1'),expectedReversePlanHash:reversePlan.reversePlanHash,expectedPublishedCorpusVersion:reversePlan.publishedCorpusVersion},prepared:SQLWrapper[]=[]
    const db={execute:<Row extends Record<string,unknown>>(query:SQLWrapper)=>pg.db.execute<Row>(query),prepare:<Row=unknown>(query:SQLWrapper)=>{prepared.push(query);return Object.freeze({}) as D1BatchItem<Row>}} as unknown as D1Client<ContentSchema>
    const contribution=await prepareReversePublishedContentImportD1(db,authorization,admin,reversePlan,reverseCommand)
    expect(contribution.items.length).toBeGreaterThan(10);expect(prepared).toHaveLength(contribution.items.length);expect(contribution.result).toMatchObject({id:published.id,state:'reversed',journalVersion:String(Number(published.journalVersion)+2),publishedCorpusVersion:published.publishedCorpusVersion})
    const interpretAfterHostBatch=async():Promise<ContentImportSession>=>{await Promise.reject(new Error('injected final participant failure'));return contribution.result}
    await expect(interpretAfterHostBatch()).rejects.toThrow('injected final participant failure')
    expect(firstRow<{state:string;version:number;entries:number;receipts:number}>(await pg.db.execute(sql`SELECT state,version,(SELECT COUNT(*) FROM content_entries)::int AS entries,(SELECT COUNT(*) FROM content_lifecycle_journal WHERE operation_id='d1:reverse:1')::int AS receipts FROM content_import_journal WHERE import_id=${published.id}`))).toMatchObject({state:'published',version:Number(published.journalVersion),entries:1,receipts:0})
    await expect(prepareReversePublishedContentImportD1(db,authorization,admin,reversePlan,{...reverseCommand,expectedPublishedCorpusVersion:'changed'})).rejects.toMatchObject({code:'conflict'})
    expect(firstRow<{state:string;entries:number;receipts:number}>(await pg.db.execute(sql`SELECT state,(SELECT COUNT(*) FROM content_entries)::int AS entries,(SELECT COUNT(*) FROM content_lifecycle_journal WHERE operation_id='d1:reverse:1')::int AS receipts FROM content_import_journal WHERE import_id=${published.id}`))).toMatchObject({state:'published',entries:1,receipts:0})
  })

  it('prepares D1 rollback as an atomic terminal result without exposing it before commit',async()=>{
    const {pg,manifest,journal,plan,session:start}=await prepare(),applying=await pg.db.transaction((tx)=>resumeContentImport(tx,journal,authorization,admin,manifest,start.id,command(plan,start,'d1:rollback:resume'))),prepared:SQLWrapper[]=[]
    const db={execute:<Row extends Record<string,unknown>>(query:SQLWrapper)=>pg.db.execute<Row>(query),prepare:<Row=unknown>(query:SQLWrapper)=>{prepared.push(query);return Object.freeze({}) as D1BatchItem<Row>}} as unknown as D1Client<ContentSchema>
    await pg.db.execute(sql`INSERT INTO content_status_definitions (key,origin,version,active,current_revision,canonical_hash,definition,created_at,updated_at) VALUES ('cleanup-drift','code',1,true,1,${'c'.repeat(64)},${JSON.stringify({key:'cleanup-drift'})},${new Date().toISOString()},${new Date().toISOString()})`)
    const contribution=await prepareRollbackContentImportD1(db,authorization,admin,applying.id,command(plan,applying,'d1:rollback:1'))
    expect(contribution.items).toHaveLength(8);expect(prepared).toHaveLength(8);expect(contribution.result).toMatchObject({id:applying.id,state:'rolledBack',journalVersion:String(Number(applying.journalVersion)+1),nextSection:null,nextOffset:0})
    expect(firstRow<{state:string;version:number;batches:number;receipts:number;entries:number}>(await pg.db.execute(sql`SELECT state,version,jsonb_array_length(payload->'batches')::int AS batches,(SELECT COUNT(*) FROM content_lifecycle_journal WHERE operation_id='d1:rollback:1')::int AS receipts,(SELECT COUNT(*) FROM content_entries)::int AS entries FROM content_import_journal WHERE import_id=${applying.id}`))).toMatchObject({state:'applying',version:Number(applying.journalVersion),batches:1,receipts:0,entries:0})
    await expect(prepareRollbackContentImportD1(db,authorization,admin,applying.id,{...command(plan,applying,'d1:rollback:stale'),expectedJournalVersion:'999'})).rejects.toMatchObject({code:'stale-journal'})
    expect(firstRow<{state:string;version:number;batches:number}>(await pg.db.execute(sql`SELECT state,version,jsonb_array_length(payload->'batches')::int AS batches FROM content_import_journal WHERE import_id=${applying.id}`))).toMatchObject({state:'applying',version:Number(applying.journalVersion),batches:1})
  })

  it('parses bounded bytes and rejects password material or incomplete definition dependencies',async()=>{const manifest=await bundle(),bytes=new TextEncoder().encode(JSON.stringify(manifest));await expect(parseContentImport(bytes,limits)).resolves.toEqual(manifest);const hostile={...manifest,entries:[{...(manifest.entries[0] as Record<string,unknown>),passwordHash:'secret'}]};await expect(parseContentImport(new TextEncoder().encode(JSON.stringify(hostile)),limits)).rejects.toMatchObject({code:'validation'});await expect(parseContentImport(new TextEncoder().encode(JSON.stringify({...manifest,statuses:[]})),limits)).rejects.toMatchObject({code:'validation'});await expect(parseContentImport(bytes,{...limits,maxBytes:bytes.byteLength-1})).rejects.toMatchObject({code:'limit'})})

  it('keeps resumed batches invisible, publishes exactly, replays, and reverses to the original corpus',async()=>{
    const {pg,manifest,journal,plan,session:start}=await prepare(),staged=await stageAll(pg,manifest,journal,plan,start)
    expect(staged.state).toBe('staged')
    expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_entries`)))[0]).toMatchObject({count:0})
    expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_type_definitions`)))[0]).toMatchObject({count:0})
    const predicted=await pg.db.transaction((tx)=>predictPublishedContentImportCorpus(tx,staged)),publishCommand=command(plan,staged,'publish:1'),published=await pg.db.transaction((tx)=>publishContentImport(tx,journal,authorization,effects,admin,staged.id,publishCommand)),replay=await pg.db.transaction((tx)=>publishContentImport(tx,journal,authorization,effects,admin,staged.id,publishCommand))
    expect(replay).toEqual(published);expect(published.state).toBe('published');expect(published.publishedCorpusVersion).toBe(predicted);expect(published.publishedCorpusVersion).toMatch(/^[0-9a-f]{64}$/)
    expect((await pg.db.transaction((tx)=>tx.execute<{type:string;status:string;hostSessionId:string|null}>(sql`SELECT type,status,host_session_id AS "hostSessionId" FROM content_entries WHERE id=${entryId}`)))[0]).toMatchObject({type:'book',status:'draft',hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa1'})
    expect((await pg.db.transaction((tx)=>tx.execute<{origin:string;hostSessionId:string|null}>(sql`SELECT origin,host_session_id AS "hostSessionId" FROM content_type_definitions WHERE key='book'`)))[0]).toMatchObject({origin:'import',hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa1'})
    expect((await pg.db.transaction((tx)=>tx.execute<{hostSessionId:string|null}>(sql`SELECT host_session_id AS "hostSessionId" FROM content_revisions WHERE id=${revisionId}`)))[0]).toMatchObject({hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa1'})
    const reversePlan=await pg.db.transaction((tx)=>planPublishedContentImportReverse(tx,journal,authorization,admin,published.id,{scope:plan.scope,requiredAuthority:plan.requiredAuthority,expectedPolicyVersion:plan.authorizationPolicyVersion})),reverseCommand={...command(plan,published,'reverse:1'),expectedReversePlanHash:reversePlan.reversePlanHash,expectedPublishedCorpusVersion:reversePlan.publishedCorpusVersion},reversed=await pg.db.transaction((tx)=>reversePublishedContentImport(tx,journal,authorization,effects,admin,reversePlan,reverseCommand))
    expect(reversed.state).toBe('reversed');expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_entries`)))[0]).toMatchObject({count:0});expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_type_definitions`)))[0]).toMatchObject({count:0});expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_definition_versions`)))[0]).toMatchObject({count:0})
  })

  it('preserves inactive status registry state on publication',async()=>{
    const pg=await fixture(),base=await bundle(),manifest=JSON.parse(JSON.stringify({...base,statuses:base.statuses.map((raw)=>({...raw as Record<string,unknown>,active:false})),entries:[],revisions:[]})) as ContentExportBundle,journal=createDbContentImportJournal(),plan=await pg.db.transaction((tx)=>planContentImport(tx,authorization,admin,manifest,{scope:{tenantKey:'tenant-1',typeKeys:['book']},conflictPolicy:'mappedReplacement'}))
    expect(plan.conflicts).toEqual([])
    let session=await pg.db.transaction((tx)=>startContentImport(tx,journal,authorization,admin,manifest,{plan,staging:{hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa2',visibility:'hiddenUntilPublish'},operationId:'inactive:start'}))
    session=await stageAll(pg,manifest,journal,plan,session)
    session=await pg.db.transaction((tx)=>publishContentImport(tx,journal,authorization,effects,admin,session.id,command(plan,session,'inactive:publish')))
    expect(session.state).toBe('published')
    expect((await pg.db.transaction((tx)=>tx.execute<{active:boolean|number}>(sql`SELECT active FROM content_status_definitions WHERE key='draft'`)))[0]).toMatchObject({active:false})
  })

  it('shadows a definitions-only mapped replacement without overwriting a code-owned type',async()=>{
    const pg=await fixture(),base=await bundle(),codeType=normalizeContentTypeDefinition({...bookType(),statusKeys:['draft'],description:'Code-owned definition'}),codeHash=await canonicalContentMigrationHash(codeType),sourceType=normalizeContentTypeDefinition({...bookType(),statusKeys:['draft'],description:'Imported shadow definition'}),sourceHash=await canonicalContentMigrationHash(sourceType),manifest=JSON.parse(JSON.stringify({...base,types:[{key:'book',origin:'db',version:1,revision:1,active:true,canonicalHash:sourceHash,definition:sourceType,versions:[{revision:1,canonicalHash:sourceHash,definition:sourceType,origin:'db',createdAt:fixed}]}],entries:[],revisions:[]})) as ContentExportBundle,now=fixed
    await pg.db.execute(sql`INSERT INTO content_type_definitions (key,origin,version,active,current_revision,canonical_hash,definition,created_at,updated_at) VALUES ('book','code',1,true,1,${codeHash},${JSON.stringify(codeType)},${now},${now})`)
    await pg.db.execute(sql`INSERT INTO content_definition_versions (definition_kind,definition_key,revision,canonical_hash,definition,origin,created_at) VALUES ('type','book',1,${codeHash},${JSON.stringify(codeType)},'code',${now})`)
    const journal=createDbContentImportJournal(),plan=await pg.db.transaction((tx)=>planContentImport(tx,authorization,admin,manifest,{scope:{tenantKey:'tenant-1',typeKeys:['book']},conflictPolicy:'mappedReplacement'}))
    expect(plan.conflicts).toEqual([])
    let session=await pg.db.transaction((tx)=>startContentImport(tx,journal,authorization,admin,manifest,{plan,staging:{hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa3',visibility:'hiddenUntilPublish'},operationId:'shadow:start'}))
    session=await stageAll(pg,manifest,journal,plan,session)
    session=await pg.db.transaction((tx)=>publishContentImport(tx,journal,authorization,effects,admin,session.id,command(plan,session,'shadow:publish')))
    expect(session.state).toBe('published')
    expect((await pg.db.transaction((tx)=>tx.execute<{origin:string;canonicalHash:string;shadowedDbVersion:number|null}>(sql`SELECT origin,canonical_hash AS "canonicalHash",shadowed_db_version AS "shadowedDbVersion" FROM content_type_definitions WHERE key='book'`)))[0]).toMatchObject({origin:'code',canonicalHash:codeHash,shadowedDbVersion:2})
    expect((await pg.db.transaction((tx)=>tx.execute<{revision:number;canonicalHash:string;origin:string;hostSessionId:string|null}>(sql`SELECT revision,canonical_hash AS "canonicalHash",origin,host_session_id AS "hostSessionId" FROM content_definition_versions WHERE definition_kind='type' AND definition_key='book' AND revision=2`)))[0]).toMatchObject({revision:2,canonicalHash:sourceHash,origin:'import',hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa3'})
  })

  it('preserve leaves an existing definition and its immutable history unchanged',async()=>{
    const pg=await fixture(),base=await bundle(),destType=normalizeContentTypeDefinition({...bookType(),statusKeys:['draft'],taxonomies:[]}),destHash=await canonicalContentMigrationHash(destType),sourceType=destType,sourceHash=destHash,statusRaw=base.statuses[0] as Record<string,unknown>,statusDefinition=statusRaw.definition as ContentStatusDefinition,statusHash=String(statusRaw.canonicalHash)
    const manifest=JSON.parse(JSON.stringify({...base,types:[{key:'book',origin:'db',version:2,revision:2,active:true,canonicalHash:sourceHash,definition:sourceType,versions:[{revision:1,canonicalHash:destHash,definition:destType,origin:'db',createdAt:fixed},{revision:2,canonicalHash:sourceHash,definition:sourceType,origin:'db',createdAt:fixed}]}],entries:[],revisions:[]})) as ContentExportBundle
    await pg.db.execute(sql`INSERT INTO content_type_definitions (key,origin,version,active,current_revision,canonical_hash,definition,created_at,updated_at) VALUES ('book','db',1,true,1,${destHash},${JSON.stringify(destType)},${fixed},${fixed})`)
    await pg.db.execute(sql`INSERT INTO content_definition_versions (definition_kind,definition_key,revision,canonical_hash,definition,origin,created_at) VALUES ('type','book',1,${destHash},${JSON.stringify(destType)},'db',${fixed})`)
    await pg.db.execute(sql`INSERT INTO content_status_definitions (key,origin,version,active,current_revision,canonical_hash,definition,created_at,updated_at) VALUES ('draft','db',1,true,1,${statusHash},${JSON.stringify(statusDefinition)},${fixed},${fixed})`)
    await pg.db.execute(sql`INSERT INTO content_definition_versions (definition_kind,definition_key,revision,canonical_hash,definition,origin,created_at) VALUES ('status','draft',1,${statusHash},${JSON.stringify(statusDefinition)},'db',${fixed})`)
    const journal=createDbContentImportJournal(),plan=await pg.db.transaction((tx)=>planContentImport(tx,authorization,admin,manifest,{scope:{tenantKey:'tenant-1',typeKeys:['book']},conflictPolicy:'preserve'}));expect(plan.conflicts).toEqual([])
    let session=await pg.db.transaction((tx)=>startContentImport(tx,journal,authorization,admin,manifest,{plan,staging:{hostSessionId:'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaa4',visibility:'hiddenUntilPublish'},operationId:'preserve:start'}));session=await stageAll(pg,manifest,journal,plan,session);session=await pg.db.transaction((tx)=>publishContentImport(tx,journal,authorization,effects,admin,session.id,command(plan,session,'preserve:publish')));expect(session.state).toBe('published')
    expect((await pg.db.transaction((tx)=>tx.execute<{version:number;currentRevision:number;canonicalHash:string;hostSessionId:string|null}>(sql`SELECT version,current_revision AS "currentRevision",canonical_hash AS "canonicalHash",host_session_id AS "hostSessionId" FROM content_type_definitions WHERE key='book'`)))[0]).toMatchObject({version:1,currentRevision:1,canonicalHash:destHash,hostSessionId:null})
    expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_definition_versions WHERE definition_kind='type' AND definition_key='book'`)))[0]).toMatchObject({count:1})
    expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_definition_versions WHERE definition_kind='type' AND definition_key='book' AND revision=2`)))[0]).toMatchObject({count:0})
  })

  it('rejects reuse of a shared host session under a different start operation with a typed conflict',async()=>{const {pg,manifest,journal,plan,session}=await prepare();await expect(pg.db.transaction((tx)=>startContentImport(tx,journal,authorization,admin,manifest,{plan,staging:session.staging,operationId:'start:other-operation'}))).rejects.toMatchObject({code:'conflict'});expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_import_journal WHERE import_id=${session.id}`)))[0]).toMatchObject({count:1})})

  it('rolls back journal-only staging without exposing canonical rows',async()=>{const {pg,manifest,journal,plan,session}=await prepare(),resumed=await pg.db.transaction((tx)=>resumeContentImport(tx,journal,authorization,admin,manifest,session.id,command(plan,session,'rollback:resume'))),rolled=await pg.db.transaction((tx)=>rollbackContentImport(tx,journal,authorization,effects,admin,resumed.id,command(plan,resumed,'rollback:apply')));expect(rolled.state).toBe('rolledBack');expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_entries`)))[0]).toMatchObject({count:0});expect((await pg.db.transaction((tx)=>tx.execute<{count:number}>(sql`SELECT COUNT(*)::int AS count FROM content_type_definitions`)))[0]).toMatchObject({count:0})})

  it('allows exactly one resume claimant at one journal coordinate',async()=>{const {pg,manifest,journal,plan,session}=await prepare(),results=await Promise.allSettled([pg.db.transaction((tx)=>resumeContentImport(tx,journal,authorization,admin,manifest,session.id,command(plan,session,'race:resume:a'))),pg.dbB.transaction((tx)=>resumeContentImport(tx,journal,authorization,admin,manifest,session.id,command(plan,session,'race:resume:b')))]);oneWinner(results)})

  it('allows resume rather than an invalid publish to commit at a non-staged coordinate',async()=>{const {pg,manifest,journal,plan,session}=await prepare(),results=await Promise.allSettled([pg.db.transaction((tx)=>resumeContentImport(tx,journal,authorization,admin,manifest,session.id,command(plan,session,'race:resume:publish:resume'))),pg.dbB.transaction((tx)=>publishContentImport(tx,journal,authorization,effects,admin,session.id,command(plan,session,'race:resume:publish:publish')))]);oneWinner(results);expect(results.some((result)=>result.status==='fulfilled'&&result.value.state==='applying')).toBe(true)})

  it('allows exactly one rollback/resume claimant at one journal coordinate',async()=>{const {pg,manifest,journal,plan,session}=await prepare(),results=await Promise.allSettled([pg.db.transaction((tx)=>resumeContentImport(tx,journal,authorization,admin,manifest,session.id,command(plan,session,'race:rollback:resume'))),pg.dbB.transaction((tx)=>rollbackContentImport(tx,journal,authorization,effects,admin,session.id,command(plan,session,'race:rollback:rollback')))]);oneWinner(results)})
})