import { randomUUID } from 'node:crypto'
import { Client, Pool } from 'pg'
import { getSslConfig } from '@open-mercato/shared/lib/db/ssl'
import { createLogger } from '@open-mercato/shared/lib/logger'
import type { EmitOptions, EventPayload } from './types'

const BRIDGE_CHANNEL = 'om_event_bridge'
const MAX_MESSAGE_BYTES = 7_000
const RECONNECT_DELAY_MS = 1_000
const CROSS_PROCESS_EVENT_INSTANCE_ID_KEY = '__openMercatoCrossProcessEventInstanceId__'

const logger = createLogger('events')

/**
 * PIDs can collide across containers, so self-echo suppression must not rely
 * on originPid alone. Every publisher stamps envelopes with this random
 * per-process instance id; consumers prefer it and fall back to originPid
 * only for envelopes published by older processes during a rolling deploy.
 */
function getCrossProcessEventInstanceId(): string {
  const globalScope = globalThis as Record<string, unknown>
  const existing = globalScope[CROSS_PROCESS_EVENT_INSTANCE_ID_KEY]
  if (typeof existing === 'string' && existing.length > 0) return existing
  const created = randomUUID()
  globalScope[CROSS_PROCESS_EVENT_INSTANCE_ID_KEY] = created
  return created
}

export const CROSS_PROCESS_EVENT_INSTANCE_ID = getCrossProcessEventInstanceId()

type BridgeEnvelope = {
  event: string
  payload: EventPayload
  options?: EmitOptions
  originPid: number
  originInstanceId?: string
}

type CrossProcessEventListener = (envelope: BridgeEnvelope) => void | Promise<void>
type PgNotificationMessage = {
  channel: string
  payload?: string
}

let publisherPool: InstanceType<typeof Pool> | null | undefined
let listenerClient: InstanceType<typeof Client> | null = null
let listenerConnectPromise: Promise<void> | null = null
let listenerReconnectTimer: ReturnType<typeof setTimeout> | null = null
const listeners = new Set<CrossProcessEventListener>()

function getDatabaseUrl(): string | null {
  const value = process.env.DATABASE_URL?.trim()
  return value && value.length > 0 ? value : null
}

function getPgBridgeConnectionString(): string | null {
  const connectionString = getDatabaseUrl()
  if (!connectionString) return null

  try {
    const url = new URL(connectionString)
    for (const key of ['sslmode', 'sslcert', 'sslkey', 'sslrootcert', 'sslcrl']) {
      url.searchParams.delete(key)
    }
    return url.toString()
  } catch {
    return connectionString
  }
}

function getPublisherPool(): InstanceType<typeof Pool> | null {
  if (publisherPool !== undefined) return publisherPool
  const connectionString = getPgBridgeConnectionString()
  if (!connectionString) {
    publisherPool = null
    return null
  }
  const ssl = getSslConfig()
  publisherPool = new Pool({
    connectionString,
    max: 2,
    ...(ssl ? { ssl } : {}),
  })
  return publisherPool
}

async function dispatchEnvelope(envelope: BridgeEnvelope): Promise<void> {
  for (const listener of listeners) {
    try {
      await Promise.resolve(listener(envelope))
    } catch (error) {
      logger.error('Cross-process listener error', { event: envelope.event, err: error })
    }
  }
}

function clearReconnectTimer(): void {
  if (!listenerReconnectTimer) return
  clearTimeout(listenerReconnectTimer)
  listenerReconnectTimer = null
}

async function closeListenerClient(): Promise<void> {
  if (!listenerClient) return
  const client = listenerClient
  listenerClient = null
  try {
    client.removeAllListeners('notification')
    client.removeAllListeners('error')
    client.removeAllListeners('end')
    await client.end()
  } catch {
    // Ignore shutdown errors.
  }
}

function scheduleReconnect(): void {
  if (listenerReconnectTimer || listeners.size === 0) return
  listenerReconnectTimer = setTimeout(() => {
    listenerReconnectTimer = null
    void ensureCrossProcessListener()
  }, RECONNECT_DELAY_MS)
}

async function ensureCrossProcessListener(): Promise<void> {
  if (listenerClient || listenerConnectPromise || listeners.size === 0) return
  const connectionString = getPgBridgeConnectionString()
  if (!connectionString) return

  listenerConnectPromise = (async () => {
    const ssl = getSslConfig()
    const client = new Client({
      connectionString,
      ...(ssl ? { ssl } : {}),
    })

    client.on('notification', (message: PgNotificationMessage) => {
      if (message.channel !== BRIDGE_CHANNEL || !message.payload) return
      try {
        const parsed = JSON.parse(message.payload) as BridgeEnvelope
        if (!parsed || typeof parsed.event !== 'string') return
        void dispatchEnvelope(parsed)
      } catch (error) {
        logger.warn('Failed to parse cross-process bridge payload', { err: error })
      }
    })

    const handleDisconnect = () => {
      if (listenerClient === client) {
        listenerClient = null
      }
      scheduleReconnect()
    }

    client.on('error', handleDisconnect)
    client.on('end', handleDisconnect)

    await client.connect()
    await client.query(`LISTEN ${BRIDGE_CHANNEL}`)
    listenerClient = client
  })()
    .catch((error) => {
      logger.warn('Cross-process event bridge listener failed', { err: error })
      scheduleReconnect()
    })
    .finally(() => {
      listenerConnectPromise = null
    })

  await listenerConnectPromise
}

export async function publishCrossProcessEvent(
  event: string,
  payload: EventPayload,
  options?: EmitOptions,
): Promise<void> {
  const pool = getPublisherPool()
  if (!pool) return

  const bridgeOptions = options
    && Object.prototype.hasOwnProperty.call(options, 'tenantId')
    && options.tenantId === undefined
    ? { ...options, tenantId: null }
    : options

  const envelope: BridgeEnvelope = {
    event,
    payload,
    options: bridgeOptions,
    originPid: process.pid,
    originInstanceId: CROSS_PROCESS_EVENT_INSTANCE_ID,
  }

  const serialized = JSON.stringify(envelope)
  if (Buffer.byteLength(serialized, 'utf8') > MAX_MESSAGE_BYTES) {
    logger.warn('Cross-process event dropped: payload exceeds size limit', { event, maxBytes: MAX_MESSAGE_BYTES })
    return
  }

  await pool.query('SELECT pg_notify($1, $2)', [BRIDGE_CHANNEL, serialized])
}

export function registerCrossProcessEventListener(listener: CrossProcessEventListener): () => void {
  listeners.add(listener)
  void ensureCrossProcessListener()

  return () => {
    listeners.delete(listener)
    if (listeners.size === 0) {
      clearReconnectTimer()
      void closeListenerClient()
    }
  }
}