// OPERATOR/CLI ONLY. Runtime host must never import this private-key module.
import pg from 'pg'
import {createPrivateKey,createPublicKey,sign} from 'node:crypto'
import {open,lstat,realpath,unlink} from 'node:fs/promises'
import {writeFile,withJournalLease} from './operator-files.mjs'
import path from 'node:path'
import {captureDatabaseSourceObservation,assertDatabaseSourceObservationCurrent} from '../../src/db/database-source-observation.js'
import {MIGRATION_MANIFEST,migrationManifestDigest} from '../../src/db/migration-manifest.js'


import {ownedContext,sameOwned,sourceCoordinates,protectedStructure,readCurrent,HISTORY,CURRENT} from './context.mjs'
import {bytes,canonical,sha,nextEpoch,POLICY_DOMAIN,HEAD_DOMAIN,KEY,liveValidity,policySchema,headSchema,publicKey} from '../../src/db/application-writer-profile-format.js'
export function createConnectedPublisher(installation,collectFixedCatalogue,sourceVerifier,approval){
 const PUBLISHER=installation.roleBindings.publisher.name
class ProfilePublicationError extends Error{
 automaticReplay=false
 constructor(outcome,phase,proposal,cleanupUnconfirmed,cause){super('Protected policy publication requires exact inspection',{cause});Object.assign(this,{outcome,phase,proposal,cleanupUnconfirmed})}
}
async function privateKey(filename){
 const resolved=await realpath(filename),meta=await lstat(filename)
 if(resolved!==path.resolve(filename)||!meta.isFile()||meta.isSymbolicLink()||meta.size<1||meta.size>8192)throw Error('Explicit bounded operator-only key file required')
 const handle=await open(resolved,'r');let data
 try{const stat=await handle.stat();if(stat.dev!==meta.dev||stat.ino!==meta.ino||stat.size!==meta.size)throw Error('Operator key changed');data=await handle.readFile();if(data.length!==stat.size)throw Error('Operator key size changed')}finally{await handle.close()}
 let key;try{key=createPrivateKey(data)}finally{data.fill(0)}
 if(key.asymmetricKeyType!=='ed25519'||sha(createPublicKey(key).export({format:'der',type:'spki'}))!==installation.publicKeySpkiSha256)throw Error('Private key does not match fixed installation')
 publicKey(installation);return key
}
// One retained actual Pool client; no injected SQL facade, SET ROLE, new pool,
// detached application callbacks or cleanup/COMMIT races.
async function session(pool,deadlineAt,work){
 if(!(pool instanceof pg.Pool)||!Number.isSafeInteger(deadlineAt)||deadlineAt<=Date.now()||deadlineAt-Date.now()>120000)throw Error('Bounded genuine publisher pool/lifetime required')
 let client,retired=false,released=false,locked=false,begun=false,commitSent=false,committed=false,confirmedValue,proposal,phase='checkout',stopReject,cleanupUnconfirmed=false
 const stopped=new Promise((_,reject)=>{stopReject=reject});void stopped.catch(()=>{})
 const release=bad=>{if(!client||released)return;released=true;client.off('error',fault);client.off('end',fault);try{client.release(bad)}catch{cleanupUnconfirmed=true}}
 const fault=()=>{if(retired)return;retired=true;cleanupUnconfirmed=true;release(true);stopReject(Error('Publisher connection/lifetime ended'))}
 const timer=setTimeout(fault,Math.max(1,deadlineAt-Date.now())),bounded=p=>Promise.race([p,stopped])
 const query=async(text,values)=>{if(retired||Date.now()>=deadlineAt)throw Error('Publisher lifetime closed');return bounded(client.query(text,values))}
 const state={query,phase:value=>{phase=value},proposal:value=>{proposal=Object.freeze(value)},begun:value=>{begun=value},commit:async value=>{
  confirmedValue=Object.freeze({...value})
  commitSent=true
  const response=await bounded(client.query('COMMIT').then(result=>{if(result.command==='COMMIT'){committed=true;begun=false}return result}))
  if(response.command!=='COMMIT')throw Error('Publisher COMMIT acknowledgement absent')
 }}
 let value,primary,failed=false
 try{
  const pending=pool.connect().then(value=>{client=value;if(retired){release(true);return value}client.on('error',fault);client.on('end',fault);return value});void pending.catch(()=>{})
  await bounded(pending);if(retired)throw Error('Late publisher checkout')
  await query("SET statement_timeout='10s'");await query("SET lock_timeout='5s'")
  phase='exclusive-admission';await query('SELECT pg_catalog.pg_advisory_lock($1::bigint)',[KEY]);locked=true
  value=await work(state)
 }catch(error){failed=true;primary=error}
 finally{
  if(!retired&&client){
   if(begun&&!commitSent)try{const r=await query('ROLLBACK');if(r.command!=='ROLLBACK')throw Error('Publisher rollback unconfirmed');begun=false}catch{fault()}
   if(commitSent&&!committed){cleanupUnconfirmed=true;retired=true;release(true)}
   if(!retired&&locked)try{const r=await query('SELECT pg_catalog.pg_advisory_unlock($1::bigint) AS released',[KEY]);if(r.rows.length!==1||r.rows[0].released!==true)throw Error('Publisher unlock unconfirmed');locked=false}catch{fault()}
   if(!retired)try{await query('RESET lock_timeout');await query('RESET statement_timeout')}catch{fault()}
  }
  clearTimeout(timer);release(retired||cleanupUnconfirmed||begun||locked||committed&&failed)
 }
 if(committed)return Object.freeze({...confirmedValue,...value,outcome:'committed',proposal,cleanupUnconfirmed:cleanupUnconfirmed||retired||failed,automaticReplay:false})
 if(failed||cleanupUnconfirmed||retired)throw new ProfilePublicationError(commitSent||begun?'unknown':'not-committed',phase,proposal,cleanupUnconfirmed||retired,primary)
 return value
}
async function publishCurrentProfile(pool,{privateKeyFile,expectedHeadSha256,journalDirectory,validityMs=86400000,deadlineAt=Date.now()+120000}){
 if(!Number.isSafeInteger(validityMs)||validityMs<1||validityMs>86400000||expectedHeadSha256!==null&&!(/^[a-f0-9]{64}$/).test(expectedHeadSha256))throw Error('Explicit previous head and bounded validity required')
 const key=await privateKey(privateKeyFile),directory=await realpath(journalDirectory),pendingFile=path.join(directory,'pending-publication.json')
 try{await lstat(pendingFile);throw Error('Pending publication must be reconciled before signing')}catch(error){if(error.code!=='ENOENT')throw error}
 const result=await session(pool,deadlineAt,async s=>{
  s.phase('publisher-read');await s.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');s.begun(true)
  const first=await ownedContext(s,installation,PUBLISHER,'publisher-read')
  const facade=Object.freeze({query:(sql,values)=>s.query(sql,values)}),options={deadlineAt}
  const observation=await captureDatabaseSourceObservation(facade,options)
  await sourceVerifier.verify(facade,observation,options)
  if(canonical(sourceCoordinates(first))!==canonical(installation.expectedSource))throw Error('Pinned source mismatch')
  const security=await protectedStructure(s,installation),catalogue=await collectFixedCatalogue(s)
  const markers=(await s.query('SELECT id FROM public.schema_migrations ORDER BY id LIMIT 45')).rows.map(row=>row.id)
  if(canonical(markers)!==canonical(MIGRATION_MANIFEST.map(pin=>pin.file.slice(0,-4))))throw Error('Exact current migration marker metadata required')
  const old=await readCurrent(s,installation,{allowEmpty:true,validNow:false})
  if((old?.headSha256??null)!==expectedHeadSha256)throw Error('Expected protected current head mismatch')
  const catalogSha256=sha(canonical({base:catalogue.catalogSha256,security:security.sha256}))
  if(approval.expiresAt<=Date.now()||approval.manifestSha256!==migrationManifestDigest()||approval.definitionSha256!==catalogue.definitionSha256||approval.catalogSha256!==catalogSha256)throw Error('Independently approved policy differs from actual source44 catalogue')
  if(old&&(canonical(old.policy.source)!==canonical(sourceCoordinates(first))||canonical(old.policy.roles)!==canonical(security.roles)||old.policy.manifestSha256!==migrationManifestDigest()||old.policy.definitionSha256!==catalogue.definitionSha256||old.policy.catalogSha256!==catalogSha256))throw Error('Renewal cannot adopt a changed source/subject/policy')
  await sourceVerifier.verify(facade,observation,options);await assertDatabaseSourceObservationCurrent(facade,observation,options);sameOwned(first,await ownedContext(s,installation,PUBLISHER,'publisher-read'))
  await s.query('COMMIT');s.begun(false)
  const number=old?nextEpoch(old.epoch):'1',now=Date.now()
  const body=policySchema.parse({format:'application-writer-policy-v1',installationId:installation.installationId,epoch:number,issuedAt:now,notBefore:now,notAfter:now+validityMs,source:sourceCoordinates(first),sourceObservationSha256:sha(canonical(observation)),roles:security.roles,manifestSha256:migrationManifestDigest(),definitionSha256:catalogue.definitionSha256,catalogSha256,futureLeaves:'fixed12-native-policy-v1'})
  const envelope=(domain,body)=>bytes({body,signature:sign(null,Buffer.from(domain+canonical(body)),key).toString('base64url')})
  const policyBytes=envelope(POLICY_DOMAIN,body),policySha256=sha(policyBytes)
  const head=headSchema.parse({format:'application-writer-head-v1',installationId:installation.installationId,epoch:number,policySha256,previousHeadSha256:old?.headSha256??null})
  const headBytes=envelope(HEAD_DOMAIN,head),headSha256=sha(headBytes)
  const proposal={installationId:installation.installationId,epoch:number,headSha256,previousHeadSha256:head.previousHeadSha256,policySha256,publisherPid:first.pid}
  s.proposal(proposal)
  // Public, bounded, exclusively created journal; neither key nor SQL data.
  await writeFile(path.join(directory,'proposal-'+headSha256+'.json'),bytes(proposal),{flag:'wx',mode:0o600})
  await writeFile(pendingFile,bytes(proposal),{flag:'wx',mode:0o600})
  s.phase('publisher-write');await s.query('BEGIN ISOLATION LEVEL READ COMMITTED READ WRITE');s.begun(true)
  await s.query('SELECT pg_catalog.pg_current_xact_id()');await s.query('SELECT pg_catalog.pg_advisory_xact_lock_shared($1::bigint)',[KEY])
  const owner=await ownedContext(s,installation,PUBLISHER,'publisher-write')
  if(owner.pid!==first.pid||canonical(sourceCoordinates(owner))!==canonical(sourceCoordinates(first)))throw Error('Retained publisher source/backend changed')
  await s.query(`SELECT id FROM mmapp_control.${CURRENT} WHERE id=1 FOR UPDATE`)
  const current=await readCurrent(s,installation,{allowEmpty:true,validNow:false})
  if((current?.headSha256??null)!==expectedHeadSha256)throw Error('CAS predecessor changed')
  const againSecurity=await protectedStructure(s,installation),againCatalogue=await collectFixedCatalogue(s)
  if(sha(canonical({base:againCatalogue.catalogSha256,security:againSecurity.sha256}))!==catalogSha256)throw Error('Policy drift during retained exclusive publication')
  if(Date.now()>=body.notAfter||Date.now()>=approval.expiresAt)throw Error('Policy or approval expired before publication dispatch')
  await s.query(`INSERT INTO mmapp_control.${HISTORY}(installation_id,epoch,head_sha256,previous_installation_id,previous_epoch,previous_head_sha256,policy_sha256,policy_bytes,head_bytes) VALUES($1::uuid,$2::bigint,$3,$4::uuid,$5::bigint,$6,$7,$8::bytea,$9::bytea)`,[installation.installationId,number,headSha256,old?installation.installationId:null,old?.epoch??null,old?.headSha256??null,policySha256,policyBytes,headBytes])
  const changed=old?await s.query(`UPDATE mmapp_control.${CURRENT} SET epoch=$1::bigint,head_sha256=$2 WHERE id=1 AND installation_id=$3::uuid AND epoch=$4::bigint AND head_sha256=$5 RETURNING id`,[number,headSha256,installation.installationId,old.epoch,old.headSha256]):await s.query(`INSERT INTO mmapp_control.${CURRENT}(id,installation_id,epoch,head_sha256) VALUES(1,$1::uuid,$2::bigint,$3) RETURNING id`,[installation.installationId,number,headSha256])
  if(changed.rows.length!==1)throw Error('Exactly one protected current CAS required')
  sameOwned(owner,await ownedContext(s,installation,PUBLISHER,'publisher-write'));if(Date.now()>=body.notAfter||Date.now()>=approval.expiresAt)throw Error('Policy or approval expired before publication terminal');s.phase('commit');const result={epoch:number,headSha256,notAfter:body.notAfter};await s.commit(result)
  const terminal=await ownedContext(s,installation,PUBLISHER,'publisher-session')
  if(terminal.pid!==first.pid||canonical(sourceCoordinates(terminal))!==canonical(sourceCoordinates(first)))throw Error('Publisher session continuity unconfirmed after actual ACK')
  return result
 })
 // A journal failure cannot turn a confirmed publication into retryable failure.
 if(result.outcome==='committed'&&!result.cleanupUnconfirmed){try{const cursor=await boundedPending(pendingFile);if(cursor.headSha256!==result.headSha256)throw Error('Pending cursor differs');await unlink(pendingFile)}catch{return Object.freeze({...result,journalUnconfirmed:true})}}
 return result
}
async function inspectPublication(pool,proposal,{deadlineAt=Date.now()+120000}={}){
 if(!proposal||proposal.installationId!==installation.installationId||!(/^[a-f0-9]{64}$/).test(proposal.headSha256)||proposal.previousHeadSha256!==null&&!(/^[a-f0-9]{64}$/).test(proposal.previousHeadSha256))throw Error('Bounded public proposal required')
 return session(pool,deadlineAt,async s=>{
  s.phase('publication-inspection');await s.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');s.begun(true)
  const first=await ownedContext(s,installation,PUBLISHER,'publisher-read');await protectedStructure(s,installation)
  if(canonical(sourceCoordinates(first))!==canonical(installation.expectedSource))throw Error('Pinned inspection source mismatch')
  const facade=Object.freeze({query:(sql,values)=>s.query(sql,values)}),options={deadlineAt}
  const observation=await captureDatabaseSourceObservation(facade,options);await sourceVerifier.verify(facade,observation,options)
  const current=await readCurrent(s,installation,{allowEmpty:true,validNow:false})
  const proposed=(await s.query(`SELECT epoch::text,head_sha256 FROM mmapp_control.${HISTORY} WHERE installation_id=$1::uuid AND epoch=$2::bigint`,[installation.installationId,proposal.epoch])).rows
  await sourceVerifier.verify(facade,observation,options);sameOwned(first,await ownedContext(s,installation,PUBLISHER,'publisher-read'))
  const response=await s.query('ROLLBACK');if(response.command!=='ROLLBACK')throw Error('Inspection terminal unconfirmed');s.begun(false)
  if(current?.headSha256===proposal.headSha256&&proposed.length===1&&proposed[0].head_sha256===proposal.headSha256)return {state:'confirmed',headSha256:current.headSha256}
  if((current?.headSha256??null)===proposal.previousHeadSha256&&proposed.length===0)return {state:'not-committed',headSha256:current?.headSha256??null}
  throw Error('Publication state requires operator inspection; no retry')
 })
}
// Read-only diagnostics on the actual protected publisher session. This does
// not admit application/operator writes, bootstrap, restore, or READY.
async function preflightCurrentProfile(pool,{deadlineAt=Date.now()+120000}={}){
 return session(pool,deadlineAt,async s=>{
  s.phase('profile-preflight');await s.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');s.begun(true)
  const first=await ownedContext(s,installation,PUBLISHER,'publisher-read')
  if(canonical(sourceCoordinates(first))!==canonical(installation.expectedSource))throw Error('Pinned preflight source mismatch')
  const facade=Object.freeze({query:(sql,values)=>s.query(sql,values)}),options={deadlineAt}
  const observation=await captureDatabaseSourceObservation(facade,options),initialSource=await sourceVerifier.verify(facade,observation,options)
  const security=await protectedStructure(s,installation),catalogue=await collectFixedCatalogue(s)
  const current=await readCurrent(s,installation)
  if(!current)throw Error('Current signed policy required for preflight')
  const catalogSha256=sha(canonical({base:catalogue.catalogSha256,security:security.sha256})),manifestSha256=migrationManifestDigest()
  if(approval.expiresAt<=Date.now()||approval.manifestSha256!==manifestSha256||approval.definitionSha256!==catalogue.definitionSha256||approval.catalogSha256!==catalogSha256)throw Error('Independent preflight approval differs from actual source44 catalogue')
  if(canonical(current.policy.source)!==canonical(sourceCoordinates(first))||canonical(current.policy.roles)!==canonical(security.roles)||current.policy.manifestSha256!==manifestSha256||current.policy.definitionSha256!==catalogue.definitionSha256||current.policy.catalogSha256!==catalogSha256)throw Error('Current signed policy differs from actual source/roles/catalogue')
  const terminalSource=await sourceVerifier.verify(facade,observation,options);await assertDatabaseSourceObservationCurrent(facade,observation,options)
  sameOwned(first,await ownedContext(s,installation,PUBLISHER,'publisher-read'))
  // Recheck the live interval after the potentially lengthy catalogue walk.
  const clock=(await s.query('SELECT (EXTRACT(EPOCH FROM pg_catalog.clock_timestamp())*1000)::bigint::text AS now')).rows
  if(clock.length!==1)throw Error('Preflight terminal server clock unavailable')
  liveValidity(current.policy,Number(clock[0].now))
  if(approval.expiresAt<=Date.now())throw Error('Preflight approval expired before terminal')
  const terminal=await s.query('ROLLBACK');if(terminal.command!=='ROLLBACK')throw Error('Preflight terminal unconfirmed');s.begun(false)
  return Object.freeze({state:'matched',epoch:current.epoch,headSha256:current.headSha256,notAfter:current.policy.notAfter,
   sourceMatched:true,rolesMatched:true,catalogueMatched:true,platformProtectionVerified:initialSource?.platformProtectionVerified===true&&terminalSource?.platformProtectionVerified===true,writerAuthorized:false,generationAuthenticated:false,readyAuthorized:false})
 })
}
async function boundedPending(filename){
 const h=await open(filename,'r');try{const stat=await h.stat();if(!stat.isFile()||stat.size<1||stat.size>2048)throw Error('Bounded operator pending cursor required');const raw=await h.readFile();if(raw.length!==stat.size)throw Error('Pending cursor changed');const p=JSON.parse(raw.toString('utf8'));if(!bytes(p).equals(raw)||Object.keys(p).sort().join(',')!=='epoch,headSha256,installationId,policySha256,previousHeadSha256,publisherPid'||!(/^[1-9][0-9]*$/).test(p.epoch)||!Number.isSafeInteger(p.publisherPid))throw Error('Exact pending cursor refused');return p}finally{await h.close()}
}
async function reconcilePendingPublication(pool,{journalDirectory,deadlineAt=Date.now()+120000}){
 const directory=await realpath(journalDirectory),filename=path.join(directory,'pending-publication.json')
 let proposal;try{proposal=await boundedPending(filename)}catch(error){if(error.code==='ENOENT')return {state:'idle'};throw error}
 const result=await inspectPublication(pool,proposal,{deadlineAt})
 const after=await boundedPending(filename);if(canonical(after)!==canonical(proposal))throw Error('Pending cursor changed during protected inspection')
 await unlink(filename)
 return {...result,proposal}
}
return Object.freeze({
 preflightCurrentProfile,
 publishCurrentProfile:(pool,options)=>withJournalLease(options.journalDirectory,()=>publishCurrentProfile(pool,options)),
 inspectPublication,
 reconcilePendingPublication:(pool,options)=>withJournalLease(options.journalDirectory,()=>reconcilePendingPublication(pool,options)),
})
}