import type { ChatSessionQueueItem } from '@cradle/db'
import { chatSessionQueueItems } from '@cradle/db'
import { and, eq, isNull } from 'drizzle-orm'

import { AppError } from '../../../errors/app-error'
import { currentUnixSeconds } from '../../../helpers/time'
import { db } from '../../../infra'
import type { ChatMessageStatus } from '../run/stream-chunks'
import { reduceChatSessionEvents } from './aggregate'
import type { ChatSessionDomainError } from './decide'
import { decideChatSessionEvents } from './decide'
import type { ChatRuntimeTx } from './event-store'
import { appendSessionEvent, readSessionEvents } from './event-store'
import { publishSessionTailEvents } from './event-tail'
import type {
  ChatSessionEvent,
  MessageRecordedFact,
  QueueProjectionStatus,
  StoredChatSessionEvent,
  TerminalRunEventType,
} from './events'
import { projectSessionEvent } from './projectors'
import { runSessionActorTask } from './session-actor'

export function commitSessionEvents(sessionId: string, events: ChatSessionEvent[]): Promise<void> {
  if (events.length === 0) {
    return Promise.resolve()
  }
  return runSessionActorTask(sessionId, () =>
    commitSessionEventsInTransaction(sessionId, events)).then((storedEvents) => {
    publishSessionTailEvents(storedEvents)
  })
}

export function commitSessionEventsWithProjection(
  sessionId: string,
  events: ChatSessionEvent[],
  projectAdditionalChanges: (tx: ChatRuntimeTx) => void,
): Promise<void> {
  return runSessionActorTask(sessionId, () => {
    const storedEvents: StoredChatSessionEvent[] = []
    db().transaction((tx) => {
      storedEvents.push(...appendDecidedSessionEvents(tx, sessionId, events))
      projectAdditionalChanges(tx)
    })
    return storedEvents
  }).then((storedEvents) => {
    publishSessionTailEvents(storedEvents)
  })
}

export function commitPreparedSessionEventsWithProjection<Result>(
  sessionId: string,
  prepare: (tx: ChatRuntimeTx) => {
    events: ChatSessionEvent[]
    result: Result
  },
  projectAdditionalChanges: (tx: ChatRuntimeTx) => void,
): Promise<Result> {
  return runSessionActorTask(sessionId, () => {
    const storedEvents: StoredChatSessionEvent[] = []
    const result = db().transaction((tx) => {
      const preparation = prepare(tx)
      storedEvents.push(...appendDecidedSessionEvents(tx, sessionId, preparation.events))
      projectAdditionalChanges(tx)
      return preparation.result
    })
    return {
      storedEvents,
      result,
    }
  }).then(({ storedEvents, result }) => {
    publishSessionTailEvents(storedEvents)
    return result
  })
}

export function commitSessionEventsInTransaction(
  sessionId: string,
  events: ChatSessionEvent[],
): StoredChatSessionEvent[] {
  const storedEvents: StoredChatSessionEvent[] = []
  db().transaction((tx) => {
    storedEvents.push(...appendDecidedSessionEvents(tx, sessionId, events))
  })
  return storedEvents
}

export function appendDecidedSessionEvents(
  tx: ChatRuntimeTx,
  sessionId: string,
  events: ChatSessionEvent[],
  options: {
    projectEvent?: (event: StoredChatSessionEvent) => boolean
  } = {},
): StoredChatSessionEvent[] {
  if (events.length === 0) {
    return []
  }

  const state = reduceChatSessionEvents(readSessionEvents(sessionId, tx))
  state.aggregateId = sessionId
  const decision = decideChatSessionEvents(state, events)
  if (!decision.ok) {
    throwDomainError(decision.error)
  }

  const storedEvents: StoredChatSessionEvent[] = []
  let expectedVersion = state.version
  for (const event of decision.events) {
    const stored = appendSessionEvent(tx, {
      aggregateId: sessionId,
      event,
      expectedVersion,
    })
    if (options.projectEvent?.(stored) ?? true) {
      projectSessionEvent(tx, stored)
    }
    storedEvents.push(stored)
    expectedVersion = stored.version
  }
  return storedEvents
}

/**
 * Recovery-only append path for a projection row that has already been validated as terminal.
 * Normal commands must use {@link appendDecidedSessionEvents}; this path exists so startup can
 * close legacy multi-run storms whose historical event ordering predates the all-origin
 * single-active-run invariant.
 */
export function appendValidatedRecoverySessionEvents(
  tx: ChatRuntimeTx,
  sessionId: string,
  input: {
    runId: string
    messageId: string | null
    events: ChatSessionEvent[]
    projectEvent?: (event: StoredChatSessionEvent) => boolean
  },
): StoredChatSessionEvent[] {
  if (!input.events.some(event =>
    (event.type === 'RunCompleted' || event.type === 'RunFailed' || event.type === 'RunAborted')
    && event.payload.runId === input.runId)) {
    throw new Error(`Recovery append requires a terminal fact for run ${input.runId}`)
  }

  const state = reduceChatSessionEvents(readSessionEvents(sessionId, tx))
  let expectedVersion = state.version
  const storedEvents: StoredChatSessionEvent[] = []
  for (const event of input.events) {
    assertRecoveryEventBelongsToRun(event, input)
    const stored = appendSessionEvent(tx, {
      aggregateId: sessionId,
      event,
      expectedVersion,
    })
    if (input.projectEvent?.(stored) ?? true) {
      projectSessionEvent(tx, stored)
    }
    storedEvents.push(stored)
    expectedVersion = stored.version
  }
  return storedEvents
}

function assertRecoveryEventBelongsToRun(
  event: ChatSessionEvent,
  input: { runId: string, messageId: string | null },
): void {
  switch (event.type) {
    case 'RunStarted':
      if (event.payload.run.id === input.runId) {
        return
      }
      break
    case 'AssistantMessageCompleted':
      if (input.messageId !== null && event.payload.message.id === input.messageId) {
        return
      }
      break
    case 'RunCompleted':
    case 'RunFailed':
    case 'RunAborted':
      if (event.payload.runId === input.runId) {
        return
      }
      break
    case 'QueueItemEnqueued':
      return
    default:
      break
  }
  throw new Error(`Unsupported recovery event ${event.type} for run ${input.runId}`)
}

export function readRunTerminalEventType(
  status: Exclude<ChatMessageStatus, 'streaming'>,
): TerminalRunEventType {
  return status === 'complete' ? 'RunCompleted' : status === 'aborted' ? 'RunAborted' : 'RunFailed'
}

export function readRunStopReason(status: Exclude<ChatMessageStatus, 'streaming'>): string {
  return status === 'complete'
    ? 'response.completed'
    : status === 'aborted'
      ? 'response.cancelled'
      : 'response.failed'
}

export function readQueueTerminalStatus(
  status: Exclude<ChatMessageStatus, 'streaming'>,
): QueueProjectionStatus {
  return status === 'complete' ? 'completed' : status === 'aborted' ? 'cancelled' : 'failed'
}

export async function claimSessionQueueItem(
  sessionId: string,
  queueItemId: string,
): Promise<ChatSessionQueueItem | undefined> {
  const result = await runSessionActorTask(sessionId, () => {
    return db().transaction((tx) => {
      const row = tx
        .select()
        .from(chatSessionQueueItems)
        .where(
          and(
            eq(chatSessionQueueItems.id, queueItemId),
            eq(chatSessionQueueItems.sessionId, sessionId),
                        eq(chatSessionQueueItems.status, 'pending'),
          ),
        )
        .get()
      if (!row) {
        return { item: undefined, storedEvents: [] }
      }
      const updatedAt = currentUnixSeconds()
      const storedEvents = appendDecidedSessionEvents(tx, sessionId, [
        {
          type: 'QueueItemClaimed',
          payload: {
            queueItemId,
            sessionId,
            updatedAt,
          },
        },
      ])
      return {
        item: {
          ...row,
          status: 'running' as const,
          errorText: null,
          updatedAt,
        },
        storedEvents,
      }
    })
  })
  publishSessionTailEvents(result.storedEvents)
  return result.item
}

export async function releaseSessionQueueItem(
  sessionId: string,
  queueItemId: string,
): Promise<void> {
  const updatedAt = currentUnixSeconds()
  await commitSessionEvents(sessionId, [
    {
      type: 'QueueItemReleased',
      payload: { queueItemId, sessionId, updatedAt },
    },
  ])
}

export async function commitLastTurnRolledBack(input: {
  sessionId: string
  messageIds: string[]
  providerRuntimeKind: string
  providerSessionId: string | null
  providerRolledBackTurns: number
  fileChangesReverted: false
  updatedAt?: number
}): Promise<void> {
  await commitSessionEvents(input.sessionId, [
    {
      type: 'LastTurnRolledBack',
      payload: {
        sessionId: input.sessionId,
        messageIds: input.messageIds,
        providerRuntimeKind: input.providerRuntimeKind,
        providerSessionId: input.providerSessionId,
        providerRolledBackTurns: input.providerRolledBackTurns,
        fileChangesReverted: input.fileChangesReverted,
        updatedAt: input.updatedAt ?? currentUnixSeconds(),
      },
    },
  ])
}

export async function recordImportedSessionMessages(input: {
  sessionId: string
  messages: Array<MessageRecordedFact & { status: 'complete' }>
}): Promise<void> {
  await commitSessionEvents(
    input.sessionId,
    input.messages.map((message): ChatSessionEvent => ({
      type: 'MessageImported',
      payload: { message },
    })),
  )
}

export function recordImportedSessionMessagesInTransaction(
  transaction: ChatRuntimeTx,
  input: {
    sessionId: string
    messages: Array<MessageRecordedFact & { status: 'complete' }>
  },
): StoredChatSessionEvent[] {
  return appendDecidedSessionEvents(
    transaction,
    input.sessionId,
    input.messages.map((message): ChatSessionEvent => ({
      type: 'MessageImported',
      payload: { message },
    })),
  )
}

export function clearProviderTargetFromSessionQueuesInTransaction(
  tx: ChatRuntimeTx,
  input: {
    providerTargetId: string
    updatedAt: number
  },
): StoredChatSessionEvent[] {
  const rows = tx
    .select({
      id: chatSessionQueueItems.id,
      sessionId: chatSessionQueueItems.sessionId,
    })
    .from(chatSessionQueueItems)
    .where(eq(chatSessionQueueItems.providerTargetId, input.providerTargetId))
    .all()
  if (rows.length === 0) {
    return []
  }

  const rowsBySessionId = new Map<string, typeof rows>()
  for (const row of rows) {
    const sessionRows = rowsBySessionId.get(row.sessionId)
    if (sessionRows) {
      sessionRows.push(row)
    }
 else {
      rowsBySessionId.set(row.sessionId, [row])
    }
  }

  const storedEvents: StoredChatSessionEvent[] = []
  for (const [sessionId, sessionRows] of rowsBySessionId) {
    storedEvents.push(
      ...appendDecidedSessionEvents(
        tx,
        sessionId,
        sessionRows.map((row): ChatSessionEvent => ({
          type: 'QueueItemProviderTargetCleared',
          payload: {
            queueItemId: row.id,
            sessionId,
            providerTargetId: input.providerTargetId,
            updatedAt: input.updatedAt,
          },
        })),
      ),
    )
  }
  return storedEvents
}

export async function failSessionQueueItem(
  sessionId: string,
  queueItemId: string,
  errorText: string,
): Promise<void> {
  const updatedAt = currentUnixSeconds()
  await commitSessionEvents(sessionId, [
    {
      type: 'QueueItemFailed',
      payload: { queueItemId, sessionId, errorText, updatedAt },
    },
  ])
}

/** Mark a durable submitted-input row complete without opening a Cradle run. */
export async function completeSessionQueueItem(
  sessionId: string,
  queueItemId: string,
): Promise<void> {
  const updatedAt = currentUnixSeconds()
  await commitSessionEvents(sessionId, [
    {
      type: 'QueueItemCompleted',
      payload: {
        queueItemId,
        sessionId,
        updatedAt,
      },
    },
  ])
}

export async function recoverOrphanedQueueItemClaims(sessionId: string): Promise<number> {
  const result = await runSessionActorTask(sessionId, () => {
    return db().transaction((tx) => {
      const rows = tx
        .select()
        .from(chatSessionQueueItems)
        .where(
          and(
            eq(chatSessionQueueItems.sessionId, sessionId),
                        eq(chatSessionQueueItems.status, 'running'),
            isNull(chatSessionQueueItems.startedRunId),
          ),
        )
        .all()
      if (rows.length === 0) {
        return { count: 0, storedEvents: [] }
      }

      const updatedAt = currentUnixSeconds()
      const storedEvents = appendDecidedSessionEvents(
        tx,
        sessionId,
        rows.map((row): ChatSessionEvent => ({
          type: 'QueueItemReleased',
          payload: {
            queueItemId: row.id,
            sessionId,
            updatedAt,
          },
        })),
      )
      return { count: rows.length, storedEvents }
    })
  })
  publishSessionTailEvents(result.storedEvents)
  return result.count
}

export async function normalizeSessionQueuePositions(sessionId: string): Promise<number> {
  const result = await runSessionActorTask(sessionId, () => {
    return db().transaction((tx) => {
      const pendingRows = tx
        .select()
        .from(chatSessionQueueItems)
        .where(
          and(
            eq(chatSessionQueueItems.sessionId, sessionId),
                        eq(chatSessionQueueItems.status, 'pending'),
          ),
        )
        .orderBy(chatSessionQueueItems.position, chatSessionQueueItems.createdAt)
        .all()

      const updatedAt = currentUnixSeconds()
      let changed = 0
      const events: ChatSessionEvent[] = []
      pendingRows.forEach((row, index) => {
        const position = index + 1
        if (row.position === position) {
          return
        }
        events.push({
          type: 'QueueItemReordered',
          payload: {
            queueItemId: row.id,
            sessionId,
            position,
            updatedAt,
          },
        })
        changed += 1
      })
      const storedEvents = appendDecidedSessionEvents(tx, sessionId, events)
      return { count: changed, storedEvents }
    })
  })
  publishSessionTailEvents(result.storedEvents)
  return result.count
}

export async function cancelQueuedSessionItem(
  sessionId: string,
  queueItemId: string,
): Promise<ChatSessionQueueItem | undefined> {
  const result = await runSessionActorTask(sessionId, () => {
    return db().transaction((tx) => {
      const row = tx
        .select()
        .from(chatSessionQueueItems)
        .where(
          and(
            eq(chatSessionQueueItems.id, queueItemId),
            eq(chatSessionQueueItems.sessionId, sessionId),
                      ),
        )
        .get()
      const cancellable
        = row !== undefined
          && (row.status === 'pending' || (row.status === 'running' && row.startedRunId === null))
      if (!row || !cancellable) {
        return { item: row, storedEvents: [] }
      }
      const updatedAt = currentUnixSeconds()
      const storedEvents = appendDecidedSessionEvents(tx, sessionId, [
        {
          type: 'QueueItemCancelled',
          payload: { queueItemId, sessionId, updatedAt },
        },
      ])
      return {
        item: {
          ...row,
          status: 'cancelled' as const,
          errorText: null,
          updatedAt,
        },
        storedEvents,
      }
    })
  })
  publishSessionTailEvents(result.storedEvents)
  return result.item
}

export async function recordQueuePositions(
  sessionId: string,
  rows: Array<Pick<ChatSessionQueueItem, 'id' | 'sessionId' | 'status' | 'position'>>,
): Promise<void> {
  const updatedAt = currentUnixSeconds()
  const events = rows.map(
    (row, index): ChatSessionEvent => ({
      type: 'QueueItemReordered',
      payload: {
        queueItemId: row.id,
        sessionId,
        position: index + 1,
        updatedAt,
      },
    }),
  )
  await commitSessionEvents(sessionId, events)
}

function throwDomainError(error: ChatSessionDomainError): never {
  throw new AppError({
    code: error.code === 'run_already_active' ? 'chat_run_in_progress' : `chat_session_${error.code}`,
    status: 409,
    message: error.message,
    details: error.details,
  })
}
