import {mapLimit} from '@/lib/concurrency'
import {
  classifyFailure,
  deadLetter,
  nextAttempt,
  park,
  readyHeads,
  remove,
  retryLater,
  type Outbox,
  type OutboxEntry,
  type RetryPolicy,
} from '@/lib/outbox'

// the drain engine behind the outbox store, dependency-injected so the store
// can wire the real effects (zustand state, the wall clock, timers, cleanError)
// while a test wires fakes. it owns the flush serialization, the per-entry
// ack/retry/dead-letter transition and the re-drain loop — the orchestration
// the store used to hold inline, which native imports kept beyond a unit test's
// reach. the pure queue transitions it composes live in @/lib/outbox

// the runner only replays a mutation; rollback (used when discarding a dead
// entry) is the store's concern, so it needs just run()
type RunHandler = {run: (payload: unknown) => Promise<void>}

export type OutboxRunnerDeps = {
  getQueue: () => Outbox
  setQueue: (map: (queue: Outbox) => Outbox) => void
  getHandler: (kind: string) => RunHandler | undefined
  now: () => number
  policy: RetryPolicy
  concurrency: number
  describeError: (err: unknown) => string
  // run once when a drain settles — the store coalesces refreshes + reschedules
  // the backoff wake here
  onSettle?: () => void
}

export const createOutboxRunner = (deps: OutboxRunnerDeps) => {
  // serialize flushes: a second trigger during a drain re-runs it once after,
  // so nothing is dropped and the same head is never sent twice concurrently
  let flushing = false
  let rerun = false

  const runEntry = async (entry: OutboxEntry): Promise<void> => {
    const handler = deps.getHandler(entry.kind)
    // an unknown kind (a queue written by a newer build, or before its store
    // registered): park it briefly rather than spin or crash — a later flush,
    // once the handler exists, drains it. parking burns no attempt: waiting
    // on registration is never the mutation's fault
    if (!handler) {
      deps.setQueue((queue) =>
        park(
          queue,
          entry.id,
          deps.now() + deps.policy.cap,
          'no handler',
        ),
      )
      return
    }
    try {
      await handler.run(entry.payload)
      deps.setQueue((queue) => remove(queue, entry.id))
    } catch (err) {
      const decision = nextAttempt(
        entry,
        classifyFailure(err),
        deps.now(),
        deps.policy,
      )
      const message = deps.describeError(err)
      deps.setQueue((queue) =>
        'dead' in decision
          ? deadLetter(queue, entry.id, message)
          : retryLater(queue, entry.id, decision.retry, message),
      )
    }
  }

  // drain every ready scope head, then re-drain as acks expose the next head in
  // a chain, until only backed-off or dead entries remain. scopes run in
  // parallel (capped); a scope's entries run strictly in order
  const flush = async (): Promise<void> => {
    if (flushing) {
      rerun = true
      return
    }
    flushing = true
    try {
      for (;;) {
        const heads = readyHeads(deps.getQueue(), deps.now())
        if (!heads.length) break
        await mapLimit(heads, deps.concurrency, runEntry)
      }
    } finally {
      flushing = false
      deps.onSettle?.()
      if (rerun) {
        rerun = false
        void flush()
      }
    }
  }

  return {flush}
}
