import 'server-only'
import { Agent } from '@atproto/api'
import { getOAuthClient, restoreAgent } from './oauth'
import { getBlockEvent, updateBlockEvent, createBlock } from './lexicon'
import { sleep } from './utils'
import type { QueueJob, JobStatus, ProgressMessage } from '@/types'

// ─── In-Memory Job Store ──────────────────────────────────────────────────────
// Anchored to globalThis so state survives Next.js HMR module re-evaluation in
// dev. In production (single Docker process) this is equivalent to module scope.
// For multi-instance deployments, swap for Redis.

declare global {
  // eslint-disable-next-line no-var
  var __activeJobs: Map<string, QueueJob> | undefined
  // eslint-disable-next-line no-var
  var __activeEventRkeys: Set<string> | undefined
}

const activeJobs: Map<string, QueueJob> = (globalThis.__activeJobs ??= new Map())
export const MAX_ACTIVE_JOBS_PER_USER = 3

// Track which event rkeys currently have a running job. Prevents concurrent
// jobs on the same event, which causes persistProgress to stomp each other's state.
const activeEventRkeys: Set<string> = (globalThis.__activeEventRkeys ??= new Set())

export function isEventJobRunning(eventRkey: string): boolean {
  return activeEventRkeys.has(eventRkey)
}

export function countRunningJobsForDid(did: string): number {
  let count = 0
  for (const job of activeJobs.values()) {
    if (job.did === did && job.status === 'running') count++
  }
  return count
}

export function canStartJobForDid(did: string): boolean {
  return countRunningJobsForDid(did) < MAX_ACTIVE_JOBS_PER_USER
}

// ─── Public API ───────────────────────────────────────────────────────────────

export function getJob(jobId: string): QueueJob | undefined {
  return activeJobs.get(jobId)
}

export function addSseClient(
  jobId: string,
  controller: ReadableStreamDefaultController
): void {
  activeJobs.get(jobId)?.sseClients.add(controller)
}

export function removeSseClient(
  jobId: string,
  controller: ReadableStreamDefaultController
): void {
  activeJobs.get(jobId)?.sseClients.delete(controller)
}

/**
 * Start a block job. Creates the in-memory job, kicks off background processing,
 * and returns immediately. Progress is streamed via SSE.
 */
export async function startBlockJob(
  jobId: string,
  did: string,
  eventRkey: string,
  dids: string[]
): Promise<QueueJob> {
  const job: QueueJob = {
    id: jobId,
    did,
    eventRkey,
    status: 'running' as JobStatus,
    total: dids.length,
    completed: 0,
    failed: 0,
    pendingDids: [...dids],
    blockedDids: [],
    failedDids: [],
    blockedRkeys: [],
    sseClients: new Set(),
    startedAt: new Date(),
  }

  activeJobs.set(jobId, job)
  activeEventRkeys.add(eventRkey)

  // Fire and forget — runs in background while caller returns
  processJob(job)
    .catch((err) => {
      console.error(`[queue] Job ${jobId} crashed unexpectedly:`, err)
      job.status = 'failed'
      broadcastProgress(job)
    })
    .finally(() => {
      activeEventRkeys.delete(eventRkey)
    })

  return job
}

/**
 * Append new DIDs to an existing block event. Pre-seeds the job with the
 * event's current state so persistProgress writes a merged record (not a reset).
 * If listUri is provided, newly blocked accounts are also added to that list.
 */
export async function startAppendJob(
  jobId: string,
  did: string,
  eventRkey: string,
  newDids: string[],
  existingState: {
    blockedDids: string[]
    blockedRkeys: string[]
    failedDids: string[]
  },
  listUri?: string
): Promise<QueueJob> {
  const job: QueueJob = {
    id: jobId,
    did,
    eventRkey,
    status: 'running' as JobStatus,
    // Total covers everything: existing + new
    total: existingState.blockedDids.length + existingState.failedDids.length + newDids.length,
    // Start counters from existing completed/failed counts
    completed: existingState.blockedDids.length,
    failed: existingState.failedDids.length,
    pendingDids: [...newDids],
    blockedDids: [...existingState.blockedDids],
    failedDids: [...existingState.failedDids],
    blockedRkeys: [...existingState.blockedRkeys],
    sseClients: new Set(),
    startedAt: new Date(),
    listUri,
  }

  activeJobs.set(jobId, job)
  activeEventRkeys.add(eventRkey)

  processJob(job)
    .catch((err) => {
      console.error(`[queue] Append job ${jobId} crashed:`, err)
      job.status = 'failed'
      broadcastProgress(job)
    })
    .finally(() => {
      activeEventRkeys.delete(eventRkey)
    })

  return job
}

// ─── Job Processing ───────────────────────────────────────────────────────────

async function processJob(job: QueueJob): Promise<void> {
  // Restore once — the OAuthSession inside the Agent handles token refresh
  // transparently on every request via its fetchHandler.
  const agent = await restoreAgent(job.did)
  const retryAttempts = new Map<string, number>()

  while (job.pendingDids.length > 0) {
    const targetDid = job.pendingDids[0]
    let succeeded = false

    try {
      const rkey = await createBlock(agent, job.did, targetDid)

      job.blockedDids.push(targetDid)
      job.blockedRkeys.push(rkey)
      job.completed++
      succeeded = true

      // If this job is appending to an existing block list, also create a listitem
      if (job.listUri) {
        try {
          await agent.com.atproto.repo.createRecord({
            repo: job.did,
            collection: 'app.bsky.graph.listitem',
            record: {
              $type: 'app.bsky.graph.listitem',
              list: job.listUri,
              subject: targetDid,
              createdAt: new Date().toISOString(),
            },
          })
        } catch {
          // Non-fatal — block record is created; list item failure is logged only
          console.warn(`[queue] Failed to add ${targetDid} to list ${job.listUri}`)
        }
      }
    } catch (err: unknown) {
      const status = (err as { status?: number })?.status
      const headers = (err as { headers?: Record<string, string> })?.headers

      const attempts = retryAttempts.get(targetDid) ?? 0
      if (status === 429 && attempts < 1) {
        // Back off and retry this DID once
        const retryAfterSec = parseInt(headers?.['retry-after'] ?? '30')
        const backoff = Math.min(retryAfterSec * 1000, 60_000)
        console.warn(`[queue] Rate limited. Backing off ${backoff}ms before retry.`)
        await sleep(backoff)
        retryAttempts.set(targetDid, attempts + 1)
        continue // retry same DID
      }

      // Any other error: mark as failed, move on
      job.failedDids.push(targetDid)
      job.failed++
      retryAttempts.delete(targetDid)
      console.error(`[queue] Failed to block ${targetDid}:`, (err as Error)?.message ?? err)
    }

    // Only advance the queue if we didn't loop back for a retry
    if (succeeded || job.failedDids[job.failedDids.length - 1] === targetDid) {
      retryAttempts.delete(targetDid)
      job.pendingDids.shift()
    }

    broadcastProgress(job)

    // Persist to PDS every 10 blocks, and always on completion
    const processed = job.completed + job.failed
    if (processed % 10 === 0 || job.pendingDids.length === 0) {
      await persistProgress(job, agent).catch((err) =>
        console.error('[queue] PDS persist failed:', err)
      )
    }

    // Conservative rate limit: 1 block per second
    if (job.pendingDids.length > 0) {
      await sleep(1000)
    }
  }

  job.status = 'complete'
  await persistProgress(job, agent).catch((err) =>
    console.error('[queue] Final PDS persist failed:', err)
  )
  broadcastProgress(job)

  // Keep job in memory for 5 minutes so late SSE clients can get final status
  setTimeout(() => activeJobs.delete(job.id), 5 * 60_000)
}

// ─── PDS Persistence ──────────────────────────────────────────────────────────

async function persistProgress(job: QueueJob, agent: Agent): Promise<void> {
  for (let attempt = 0; attempt < 3; attempt++) {
    const { record, cid } = await getBlockEvent(agent, job.did, job.eventRkey)

    try {
      await updateBlockEvent(agent, job.did, job.eventRkey, {
        ...record,
        pendingDids: [...job.pendingDids],
        blockedDids: [...job.blockedDids],
        failedDids: [...job.failedDids],
        blockedRkeys: [...job.blockedRkeys],
        stats: {
          total: job.total,
          completed: job.completed,
          failed: job.failed,
        },
      }, { swapRecord: cid })
      return
    } catch (err: unknown) {
      const status = (err as { status?: number })?.status
      const errorName = (err as { error?: string })?.error
      if ((status === 409 || errorName === 'InvalidSwap') && attempt < 2) {
        await sleep(250)
        continue
      }
      throw err
    }
  }
}

// ─── SSE Broadcasting ─────────────────────────────────────────────────────────

function broadcastProgress(job: QueueJob): void {
  const message: ProgressMessage = {
    jobId: job.id,
    status: job.status,
    total: job.total,
    completed: job.completed,
    failed: job.failed,
  }

  const encoded = new TextEncoder().encode(`data: ${JSON.stringify(message)}\n\n`)
  const isDone = job.status !== 'running'

  for (const controller of job.sseClients) {
    try {
      controller.enqueue(encoded)
      if (isDone) controller.close()
    } catch {
      // Client disconnected — clean up
      job.sseClients.delete(controller)
    }
  }
}