/**
 * SSE Event Stream — DOM Event Bridge
 *
 * Server-Sent Events endpoint that bridges server-side events to the browser.
 * Only events with `clientBroadcast: true` in their EventDefinition are sent.
 * Events are scoped to the authenticated user's tenant.
 *
 * Client consumer: `packages/ui/src/backend/injection/eventBridge.ts`
 */

import { resolveRequestContext } from '@open-mercato/shared/lib/api/context'
import type { OrganizationScopeService } from '@open-mercato/shared/lib/auth/principal-service'
import { createLogger } from '@open-mercato/shared/lib/logger'
import { parseNumberWithDefault } from '@open-mercato/shared/lib/number'
import { isBroadcastEvent } from '@open-mercato/shared/modules/events'
import {
  CROSS_PROCESS_EVENT_INSTANCE_ID,
  registerCrossProcessEventListener,
  registerGlobalEventTap,
} from '../../../../bus'
import type { EmitOptions } from '../../../../types'

export const metadata = {
  GET: { requireAuth: true },
}

const HEARTBEAT_INTERVAL_MS = 30_000
const DEFAULT_AUTH_REVALIDATION_INTERVAL_MS = 30_000
const DEFAULT_CONNECTION_MAX_AGE_MS = 5 * 60_000
const MIN_AUTH_REVALIDATION_INTERVAL_MS = 1_000
const MIN_CONNECTION_MAX_AGE_MS = 1_000
const CONNECTION_MAX_AGE_JITTER_RATIO = 0.15
const MAX_TIMER_DELAY_MS = 2_147_483_647
const MAX_PAYLOAD_BYTES = 4096

const logger = createLogger('events').child({ component: 'stream' })

type SseConnectionIdentity = {
  tenantId: string
  organizationId: string | null
  userId: string
  roleIds: string[]
}

type SseConnection = {
  identity: SseConnectionIdentity
  send: (data: string) => void
  close: () => void
}

function resolveTimingConfig(): {
  authRevalidationIntervalMs: number
  connectionMaxAgeMs: number
} {
  return {
    authRevalidationIntervalMs: Math.min(
      parseNumberWithDefault(
        process.env.OM_EVENTS_SSE_AUTH_REVALIDATION_INTERVAL_MS,
        DEFAULT_AUTH_REVALIDATION_INTERVAL_MS,
        { integer: true, min: MIN_AUTH_REVALIDATION_INTERVAL_MS },
      ),
      MAX_TIMER_DELAY_MS,
    ),
    connectionMaxAgeMs: applyConnectionMaxAgeJitter(
      parseNumberWithDefault(
        process.env.OM_EVENTS_SSE_CONNECTION_MAX_AGE_MS,
        DEFAULT_CONNECTION_MAX_AGE_MS,
        { integer: true, min: MIN_CONNECTION_MAX_AGE_MS },
      ),
    ),
  }
}

function applyConnectionMaxAgeJitter(configuredMs: number): number {
  const jitterFactor = 1 + (Math.random() * 2 - 1) * CONNECTION_MAX_AGE_JITTER_RATIO
  const jitteredMs = Math.round(configuredMs * jitterFactor)
  return Math.min(Math.max(jitteredMs, MIN_CONNECTION_MAX_AGE_MS), MAX_TIMER_DELAY_MS)
}

function normalizeRoleIds(input: unknown): string[] {
  if (!Array.isArray(input)) return []
  return Array.from(new Set(
    input
      .filter((role): role is string => typeof role === 'string')
      .map((role) => role.trim())
      .filter((role) => role.length > 0),
  )).sort((left, right) => left.localeCompare(right))
}

function resolveOrganizationScopeService(
  ctx: Awaited<ReturnType<typeof resolveRequestContext>>['ctx'],
): OrganizationScopeService | null {
  try {
    const service = ctx.container.resolve<OrganizationScopeService>('organizationScopeService')
    return service && typeof service.resolveForRequest === 'function' ? service : null
  } catch {
    return null
  }
}

function normalizeId(input: unknown): string | null {
  return typeof input === 'string' && input.trim().length > 0 ? input.trim() : null
}

async function resolveConnectionIdentity(
  ctx: Awaited<ReturnType<typeof resolveRequestContext>>['ctx'],
  request: Request,
): Promise<SseConnectionIdentity | null> {
  const canonicalTenantId = normalizeId(ctx.auth?.tenantId)
  if (!canonicalTenantId || !ctx.auth?.sub) return null
  const organizationScopeService = resolveOrganizationScopeService(ctx)
  if (!organizationScopeService) return null
  const scope = await organizationScopeService.resolveForRequest({ auth: ctx.auth, request })
  const tenantId = normalizeId(scope.tenantId)
  // `selectedId` is null whenever the caller has "All organizations" selected —
  // the default an unrestricted admin's organization switcher persists — so using
  // it alone leaves the connection with no organization and `matchesAudience`
  // then rejects EVERY organization-scoped event. Keep the home-organization
  // fallback the bridge has always used; the allowed-set check below still
  // refuses an organization the principal may not see.
  const organizationId = normalizeId(scope.selectedId) ?? normalizeId(ctx.auth.orgId)
  if (!tenantId || tenantId !== canonicalTenantId || scope.selectionRejected) return null
  if (organizationId && Array.isArray(scope.allowedIds) && !scope.allowedIds.includes(organizationId)) {
    return null
  }
  return {
    tenantId,
    organizationId,
    userId: ctx.auth.sub,
    roleIds: normalizeRoleIds(ctx.auth.roles),
  }
}

function identitiesMatch(left: SseConnectionIdentity, right: SseConnectionIdentity): boolean {
  return left.tenantId === right.tenantId
    && left.organizationId === right.organizationId
    && left.userId === right.userId
    && left.roleIds.length === right.roleIds.length
    && left.roleIds.every((roleId, index) => roleId === right.roleIds[index])
}

function hasApiKeyCredentials(req: Request): boolean {
  if ((req.headers.get('x-api-key') ?? '').trim().length > 0) return true
  const authorization = (req.headers.get('authorization') ?? '').trim()
  return /^apikey(?:\s|$)/i.test(authorization)
}

function createAuthRevalidationRequest(req: Request): Request {
  return new Request(req.url, {
    method: req.method,
    headers: req.headers,
  })
}

function collectStringValues(input: unknown): string[] {
  if (!Array.isArray(input)) return []
  const values: string[] = []
  for (const value of input) {
    if (typeof value !== 'string') continue
    const trimmed = value.trim()
    if (!trimmed) continue
    values.push(trimmed)
  }
  return values
}

function normalizeAudience(data: Record<string, unknown>, options?: EmitOptions): {
  tenantId: string | null
  organizationScopes: string[]
  recipientUserScopes: string[]
  recipientRoleScopes: string[]
} {
  const hasTrustedScope = options != null
    && Object.prototype.hasOwnProperty.call(options, 'tenantId')
  const trustedTenantId = typeof options?.tenantId === 'string' && options.tenantId.trim().length > 0
    ? options.tenantId.trim()
    : null
  const tenantId = hasTrustedScope
    ? trustedTenantId
    : (typeof data.tenantId === 'string' ? data.tenantId : null)
  const organizationScopes = new Set<string>()
  const trustedOrganizationId = typeof options?.organizationId === 'string' && options.organizationId.trim().length > 0
    ? options.organizationId.trim()
    : null
  if (hasTrustedScope) {
    if (trustedOrganizationId) organizationScopes.add(trustedOrganizationId)
    for (const organizationId of collectStringValues(options?.organizationIds)) {
      organizationScopes.add(organizationId)
    }
  } else {
    if (typeof data.organizationId === 'string' && data.organizationId.trim().length > 0) {
      organizationScopes.add(data.organizationId.trim())
    }
    for (const organizationId of collectStringValues(data.organizationIds)) {
      organizationScopes.add(organizationId)
    }
  }

  const recipientUserScopes = new Set<string>()
  if (typeof data.recipientUserId === 'string' && data.recipientUserId.trim().length > 0) {
    recipientUserScopes.add(data.recipientUserId.trim())
  }
  for (const userId of collectStringValues(data.recipientUserIds)) {
    recipientUserScopes.add(userId)
  }

  const recipientRoleScopes = new Set<string>()
  if (typeof data.recipientRoleId === 'string' && data.recipientRoleId.trim().length > 0) {
    recipientRoleScopes.add(data.recipientRoleId.trim())
  }
  for (const roleId of collectStringValues(data.recipientRoleIds)) {
    recipientRoleScopes.add(roleId)
  }

  return {
    tenantId,
    organizationScopes: Array.from(organizationScopes),
    recipientUserScopes: Array.from(recipientUserScopes),
    recipientRoleScopes: Array.from(recipientRoleScopes),
  }
}

function matchesAudience(conn: SseConnection, audience: ReturnType<typeof normalizeAudience>): boolean {
  if (!audience.tenantId) return false
  if (conn.identity.tenantId !== audience.tenantId) return false

  if (audience.organizationScopes.length > 0) {
    if (!conn.identity.organizationId) return false
    if (!audience.organizationScopes.includes(conn.identity.organizationId)) return false
  }

  if (audience.recipientUserScopes.length > 0 && !audience.recipientUserScopes.includes(conn.identity.userId)) {
    return false
  }

  if (audience.recipientRoleScopes.length > 0) {
    const roleMatched = conn.identity.roleIds.some((roleId) => audience.recipientRoleScopes.includes(roleId))
    if (!roleMatched) return false
  }

  return true
}

/**
 * Global connection registry.
 * All active SSE connections are tracked here.
 * The event bus subscriber iterates this set on each broadcast event.
 */
const connections = new Set<SseConnection>()

let globalTapRegistered = false

async function broadcastEventToConnections(
  eventName: string,
  payload: Record<string, unknown>,
  options?: EmitOptions,
): Promise<void> {
  if (!eventName || connections.size === 0) return

  if (!isBroadcastEvent(eventName)) return

  const data = payload ?? {}
  const audience = normalizeAudience(data, options)

  const organizationId = audience.organizationScopes[0] ?? ''
  let ssePayload = buildSsePayload(eventName, data, organizationId)

  if (new TextEncoder().encode(ssePayload).length > MAX_PAYLOAD_BYTES) {
    ssePayload = buildTruncatedPayload(eventName, data, organizationId)
    if (new TextEncoder().encode(ssePayload).length > MAX_PAYLOAD_BYTES) {
      logger.warn('Event payload exceeds size limit, skipping', { event: eventName, maxBytes: MAX_PAYLOAD_BYTES })
      return
    }
  }

  for (const conn of connections) {
    if (!matchesAudience(conn, audience)) continue

    try {
      conn.send(ssePayload)
    } catch {
      // Connection may have been closed; cleanup happens via abort handler
    }
  }
}

function buildSsePayload(eventName: string, data: Record<string, unknown>, organizationId: string): string {
  return JSON.stringify({
    id: eventName,
    payload: data,
    timestamp: Date.now(),
    organizationId,
  })
}

function buildTruncatedPayload(eventName: string, data: Record<string, unknown>, organizationId: string): string {
  const entityRef: Record<string, unknown> = {
    truncated: true,
  }
  if (typeof data.id === 'string' && data.id.trim().length > 0) {
    entityRef.id = data.id.trim()
  }
  if (typeof data.entityId === 'string' && data.entityId.trim().length > 0) {
    entityRef.entityId = data.entityId.trim()
  }
  if (typeof data.entityType === 'string' && data.entityType.trim().length > 0) {
    entityRef.entityType = data.entityType.trim()
  }
  return JSON.stringify({
    id: eventName,
    payload: entityRef,
    timestamp: Date.now(),
    organizationId,
  })
}

/**
 * Ensure a process-wide event tap is registered (once).
 * This captures emits from all request-scoped EventBus instances.
 */
function ensureGlobalTapSubscription(): void {
  if (globalTapRegistered) return
  globalTapRegistered = true

  registerGlobalEventTap(async (eventName, payload, options) => {
    await broadcastEventToConnections(eventName, (payload ?? {}) as Record<string, unknown>, options)
  })

  registerCrossProcessEventListener(async (envelope) => {
    // Every envelope this process publishes carries originInstanceId, so an
    // envelope without one can only come from an older process during a
    // rolling deploy and is always foreign. Falling back to originPid would
    // discard it whenever both containers run under the same pid — commonly
    // pid 1 — silently dropping browser events across the upgrade window.
    if (envelope.originInstanceId === CROSS_PROCESS_EVENT_INSTANCE_ID) return
    await broadcastEventToConnections(
      envelope.event,
      (envelope.payload ?? {}) as Record<string, unknown>,
      envelope.options,
    )
  })
}

export async function GET(req: Request): Promise<Response> {
  if (hasApiKeyCredentials(req)) {
    return new Response('Unauthorized', { status: 401 })
  }
  const { ctx } = await resolveRequestContext(req)
  if (ctx.auth?.isApiKey === true) {
    return new Response('Unauthorized', { status: 401 })
  }
  const initialIdentity = await resolveConnectionIdentity(ctx, req)
  if (!initialIdentity) {
    return new Response('Unauthorized', { status: 401 })
  }
  const authorizedIdentity = initialIdentity
  const { authRevalidationIntervalMs, connectionMaxAgeMs } = resolveTimingConfig()

  ensureGlobalTapSubscription()

  const encoder = new TextEncoder()
  let heartbeatTimer: ReturnType<typeof setInterval> | null = null
  let authRevalidationTimer: ReturnType<typeof setInterval> | null = null
  let maxAgeTimer: ReturnType<typeof setTimeout> | null = null
  let streamController: ReadableStreamDefaultController<Uint8Array> | null = null
  let connection: SseConnection | null = null
  let authRevalidationInFlight = false
  let cleanedUp = false
  const onAbort = () => closeConnection()

  const stream = new ReadableStream({
    start(controller) {
      streamController = controller
      const send = (data: string) => {
        if (cleanedUp) return
        try {
          streamController?.enqueue(encoder.encode(`data: ${data}\n\n`))
        } catch {
          closeConnection()
        }
      }

      connection = {
        identity: authorizedIdentity,
        send,
        close: closeConnection,
      }
      connections.add(connection)

      // Flush an initial comment so the runtime sends the response headers and
      // first body byte immediately. Without it, the streamed Response stays
      // unflushed until the first heartbeat (30s) or matching event, which
      // delays the browser EventSource `open` event — clients that gate work on
      // a "connected" signal would otherwise stall for up to 30s after mount.
      // Comment lines (`:` prefix) are ignored by EventSource message parsing.
      streamController.enqueue(encoder.encode(': connected\n\n'))

      // Start heartbeat to keep connection alive
      heartbeatTimer = setInterval(() => {
        try {
          streamController?.enqueue(encoder.encode(':heartbeat\ndata: :heartbeat\n\n'))
        } catch {
          closeConnection()
        }
      }, HEARTBEAT_INTERVAL_MS)

      authRevalidationTimer = setInterval(() => {
        void revalidateAuthorization()
      }, authRevalidationIntervalMs)

      maxAgeTimer = setTimeout(() => {
        closeConnection()
      }, connectionMaxAgeMs)
    },
    cancel() {
      cleanup()
    },
  })

  async function revalidateAuthorization(): Promise<void> {
    if (cleanedUp || authRevalidationInFlight) return
    authRevalidationInFlight = true
    try {
      const validationRequest = createAuthRevalidationRequest(req)
      const { ctx: freshCtx } = await resolveRequestContext(validationRequest)
      if (cleanedUp) return
      const freshIdentity = await resolveConnectionIdentity(freshCtx, validationRequest)
      if (!freshIdentity || !identitiesMatch(authorizedIdentity, freshIdentity)) {
        closeConnection()
        return
      }
      if (connection) connection.identity = freshIdentity
    } catch {
      closeConnection()
    } finally {
      authRevalidationInFlight = false
    }
  }

  function closeConnection(): void {
    if (cleanedUp) return
    const controller = streamController
    cleanup()
    try {
      controller?.close()
    } catch {}
  }

  function cleanup() {
    if (cleanedUp) return
    cleanedUp = true
    if (heartbeatTimer) {
      clearInterval(heartbeatTimer)
      heartbeatTimer = null
    }
    if (authRevalidationTimer) {
      clearInterval(authRevalidationTimer)
      authRevalidationTimer = null
    }
    if (maxAgeTimer) {
      clearTimeout(maxAgeTimer)
      maxAgeTimer = null
    }
    if (connection) {
      connections.delete(connection)
      connection = null
    }
    streamController = null
    // Detach from the request signal so reconnect churn does not accumulate
    // listeners and closures on long-lived AbortSignals.
    req.signal.removeEventListener('abort', onAbort)
  }

  req.signal.addEventListener('abort', onAbort, { once: true })
  if (req.signal.aborted) onAbort()

  return new Response(stream, {
    status: 200,
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache, no-transform',
      'Connection': 'keep-alive',
      'X-Accel-Buffering': 'no',
    },
  })
}

export const openApi = {
  GET: {
    summary: 'Subscribe to server events via SSE (DOM Event Bridge)',
    description: 'Server-bounded SSE connection for staff cookie or Bearer authentication that receives server-side events marked with clientBroadcast: true. API-key credentials are not accepted. Events are server-filtered by tenant, organization, recipient user, and recipient role, with periodic canonical authorization revalidation.',
    tags: ['Events'],
    responses: {
      200: {
        description: 'Event stream (text/event-stream)',
        content: {
          'text/event-stream': {
            schema: {
              type: 'object',
              properties: {
                id: { type: 'string', description: 'Event identifier (e.g., example.todo.created)' },
                payload: { type: 'object', description: 'Event-specific data' },
                timestamp: { type: 'number', description: 'Server timestamp (ms since epoch)' },
                organizationId: { type: 'string', description: 'Organization scope' },
              },
            },
          },
        },
      },
      401: { description: 'Not authenticated' },
    },
  },
}