import { createHash } from 'node:crypto'
import { XrpcError } from '@radial/atproto'
import {
  activeGoals,
  COLLECTIONS,
  compareCodePoints,
  type ArtifactRecord,
  type IndexedRecord,
  type MaterializedIndex,
  type MergeRecord,
} from '@radial/core'
import { actorAccepts, type ActorRegistry, type LoadedActor } from './actors.js'
import { base32Encode } from './turn-socket.js'

/**
 * The forge state source the poller reads (design §10). A minimal LOCAL structural type so this
 * module compiles and tests can mock it before Issue 1's `forge.ts` lands; the orchestrator-pinned
 * interface is exactly `getPullRequestState(prUrl) -> {state, mergedAt?}`, so at integration this
 * type is deleted and the real `ForgeAdapter` (which structurally satisfies it) is imported instead.
 */
export interface ForgeStateSource {
  getPullRequestState(
    prUrl: string,
    context?: { gitUrl?: string },
  ): Promise<{ state: 'open' | 'merged' | 'closed'; mergedAt?: string; headRef?: string; headRepoFullName?: string }>
  projectIdentity?(gitUrl: string): Promise<string>
}

export interface MergeCandidate {
  /** The goal-scoped implementation artifact whose PR we should poll. */
  artifact: IndexedRecord<ArtifactRecord>
  prUrl: string
  /** The remote of the project this artifact's goal hangs off, when the index still holds that
   * project. A HINT for the adapter, never part of routing (`PullStateContext` in `forge.ts`): a
   * tangled pull URL names the repository's own DID, and the owner whose repo a merge is recorded
   * in appears nowhere in it — so without this an appview outage leaves the merge unobservable. */
  gitUrl?: string
  /** The actor whose repo a merge annotation would be written into (see `selectWritingActor`). */
  actor: LoadedActor
}

/** Stable dedup key for "one merge per artifact (version)": the exact strongref a merge pins. */
const artifactKey = (record: { uri: string; cid: string }): string => `${record.uri}#${record.cid}`

/**
 * Deterministic rkey for a merge annotation of an exact artifact version: `merge-` plus 24 base32
 * chars of sha256(artifactUri#artifactCid). Per-actor repos make this unique per daemon-actor, so a
 * restart that re-derives the same rkey collides on the PDS (RecordAlreadyExists) instead of writing
 * a second record — closing the restart-during-ingestion-lag double-write that the in-memory
 * `alreadyWritten` set alone cannot (it is lost on restart). Mirrors `planArtifactRkey` in
 * turn-socket.ts and its RecordAlreadyExists adoption path.
 */
export function mergeRkey(artifactUri: string, artifactCid: string): string {
  const digest = createHash('sha256').update(`${artifactUri}#${artifactCid}`).digest()
  return `merge-${base32Encode(digest).slice(0, 24)}`
}

/**
 * The writing actor is deterministic (design decision, locked): among the ELIGIBLE actors, the
 * `implementation`-producing one if present, else the lowest DID. Ties within the
 * implementation-producing set also break by lowest DID. `eligible` must already be restricted to
 * actors that hold an active membership grant in the target space (see `selectMergeCandidates`), so
 * a merge is never written under a DID whose records the space would discard. Returns undefined when
 * no eligible actor exists.
 */
export function selectWritingActor(
  eligible: LoadedActor[],
  spaceUri: string,
): LoadedActor | undefined {
  const byDid = (a: LoadedActor, b: LoadedActor): number => compareCodePoints(a.did, b.did)
  // "The implementer" is a per-space fact like every other capability: a profile scoped to plan-only
  // here is not this space's implementer, whatever it does in the space next door.
  const implementers = eligible
    .filter((actor) => actorAccepts(actor, 'implementation', spaceUri))
    .sort(byDid)
  if (implementers[0]) return implementers[0]
  return [...eligible].sort(byDid)[0]
}

/**
 * Pure selection gate: which merged-observation candidates are worth polling right now. A candidate
 * is a goal-scoped `implementation` artifact that (a) links a PR, (b) is not in `alreadyWritten`
 * (the in-process guard against the sync-lag double-write race), and (c) has no trusted merge in the
 * index authored by one of OUR actor DIDs — the "a given daemon writes at most one merge per
 * artifact" rule, deduped against already-ingested writes. Equivalent merges from OTHER operators'
 * DIDs are intentionally ignored here: each daemon still posts its own single annotation.
 *
 * The writing actor is restricted to actors holding an active membership grant in THIS index's space
 * (materialize is per-space, so `index.members` is that space's trust view). If no loaded actor is
 * an active member, every candidate is skipped and `onSkip` is invoked once — otherwise the daemon
 * would emit merge records that `trustRecords` discards, producing no merged signal and endless
 * ignored writes. Ordered by artifact uri so polling/scheduling is stable across ticks.
 */
export function selectMergeCandidates(
  index: MaterializedIndex,
  actors: ActorRegistry,
  alreadyWritten: Set<string>,
  onSkip?: (reason: string) => void,
): MergeCandidate[] {
  const activeMemberDids = new Set(
    index.members.filter((member) => member.active).map((member) => member.did),
  )
  const eligible = actors.all.filter((actor) => activeMemberDids.has(actor.did))
  const actor = selectWritingActor(eligible, index.space.uri)
  if (!actor) {
    onSkip?.('no loaded actor holds an active membership grant in this space to author a merge')
    return []
  }
  const ourDids = new Set(actors.all.map((candidate) => candidate.did))

  const candidates: MergeCandidate[] = []
  const gitUrlByProject = new Map(index.projects.map((view) => [view.target.uri, view.gitUrl]))
  // A shelved goal is set aside: agents stop polling/merging its PRs. `activeGoals` covers a goal any
  // member has ended, and a goal under a project its author archived.
  for (const goalView of activeGoals(index)) {
    const gitUrl = gitUrlByProject.get(goalView.target.value.project.uri)
    for (const artifact of goalView.artifacts) {
      if (artifact.value.type !== 'implementation') continue
      const prUrl = artifact.value.links?.pr
      if (!prUrl) continue
      const key = artifactKey(artifact)
      if (alreadyWritten.has(key)) continue
      const alreadyMergedByUs = goalView.merges.some(
        (merge) => artifactKey(merge.value.artifact) === key && ourDids.has(merge.did),
      )
      if (alreadyMergedByUs) continue
      candidates.push({ artifact, prUrl, ...(gitUrl ? { gitUrl } : {}), actor })
    }
  }
  return candidates.sort((a, b) => compareCodePoints(a.artifact.uri, b.artifact.uri))
}

export interface MergePollerDeps {
  adapter: ForgeStateSource
  /** Monotonic wall clock in epoch milliseconds; injectable so tests drive backoff without timers. */
  now: () => number
  /** Base per-PR poll interval; also the first backoff step (capped at `backoffMaxMs`). */
  pollIntervalMs: number
  /** Exponential backoff cap. */
  backoffMaxMs: number
  log?: (message: string) => void
}

interface PrSchedule {
  nextPollAt: number
  backoffMs: number
}

/**
 * Background merge-observation poller, modeled on `TurnDispatcher`: `pump` fires off forge lookups
 * for eligible PRs and returns immediately — it never awaits a lookup, so it never blocks the
 * ingestion loop that calls it. Per-PR exponential backoff (base `pollIntervalMs`, capped at
 * `backoffMaxMs` — and the first step is likewise capped, so a misconfigured base > cap can't
 * produce an uncapped first delay) is applied on open/closed/error; a merged PR gets exactly one
 * merge annotation written (via the selected actor) and is then dropped from scheduling forever.
 *
 * Double-write defense is two-layered. Within a process, the in-memory `#alreadyWritten` set closes
 * the sync-lag race — the same one `dispatch.ts` documents — between writing a merge and the next
 * ingestion cycle surfacing it in `index.merges`. Across a restart (which loses that set), the
 * DETERMINISTIC rkey (`mergeRkey`) makes a re-derived write collide on the PDS; the collision is
 * ADOPTED as success (fetch the existing record, no duplicate), exactly like turn-socket.ts's
 * artifact submission.
 */
export class MergePoller {
  readonly #deps: MergePollerDeps
  readonly #schedules = new Map<string, PrSchedule>()
  readonly #inFlight = new Map<string, Promise<void>>()
  readonly #alreadyWritten = new Set<string>()
  /** Skip reasons already logged, so a stable no-eligible-writer condition is reported once. */
  readonly #loggedSkips = new Set<string>()

  constructor(deps: MergePollerDeps) {
    this.#deps = deps
  }

  get inFlight(): number {
    return this.#inFlight.size
  }

  /** The first backoff step, never above the cap (defends a base > cap misconfiguration). */
  #initialBackoff(): number {
    return Math.min(this.#deps.pollIntervalMs, this.#deps.backoffMaxMs)
  }

  /** Launches due forge lookups for eligible PRs; returns immediately. */
  pump(index: MaterializedIndex, actors: ActorRegistry): void {
    const now = this.#deps.now()
    const candidates = selectMergeCandidates(index, actors, this.#alreadyWritten, (reason) => {
      if (this.#loggedSkips.has(reason)) return
      this.#loggedSkips.add(reason)
      this.#deps.log?.(`not observing merges: ${reason}`)
    })
    for (const candidate of candidates) {
      const { prUrl } = candidate
      if (this.#inFlight.has(prUrl)) continue
      const schedule = this.#schedules.get(prUrl)
      if (schedule && schedule.nextPollAt > now) continue
      if (!schedule) this.#schedules.set(prUrl, { nextPollAt: 0, backoffMs: this.#initialBackoff() })

      const promise = this.#poll(candidate).finally(() => {
        this.#inFlight.delete(prUrl)
      })
      this.#inFlight.set(prUrl, promise)
    }
  }

  async #poll(candidate: MergeCandidate): Promise<void> {
    const { artifact, prUrl, actor } = candidate
    let result: {
      state: 'open' | 'merged' | 'closed'
      mergedAt?: string
      headRef?: string
      headRepoFullName?: string
    }
    try {
      result = await this.#deps.adapter.getPullRequestState(
        prUrl,
        candidate.gitUrl ? { gitUrl: candidate.gitUrl } : {},
      )
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error)
      this.#deps.log?.(`merge poll error for ${prUrl}: ${message}; backing off`)
      this.#backoff(prUrl)
      return
    }

    if (result.state !== 'merged') {
      // open or closed-without-merge: nothing is ever recorded (design decision, locked); keep
      // polling under exponential backoff — a closed PR may still be reopened and merged.
      this.#backoff(prUrl)
      return
    }

    const branch = artifact.value.links?.branch
    if (branch && result.headRef !== branch) {
      this.#deps.log?.(
        `merge identity mismatch for ${prUrl}: expected head ${branch}, got ${result.headRef ?? 'unknown'}; backing off`,
      )
      this.#backoff(prUrl)
      return
    }
    if (branch && candidate.gitUrl && this.#deps.adapter.projectIdentity) {
      try {
        const project = await this.#deps.adapter.projectIdentity(candidate.gitUrl)
        if (result.headRepoFullName !== project) {
          this.#deps.log?.(
            `merge identity mismatch for ${prUrl}: expected repository ${project}, got ${result.headRepoFullName ?? 'unknown'}; backing off`,
          )
          this.#backoff(prUrl)
          return
        }
      } catch (error) {
        this.#deps.log?.(
          `merge identity check failed for ${prUrl}: ${error instanceof Error ? error.message : String(error)}; backing off`,
        )
        this.#backoff(prUrl)
        return
      }
    }

    const key = artifactKey(artifact)
    const record: MergeRecord = {
      $type: COLLECTIONS.merge,
      artifact: { uri: artifact.uri, cid: artifact.cid },
      pr: prUrl,
      // createdAt is our observation time; mergedAt comes from the forge when it reports one.
      ...(result.mergedAt !== undefined ? { mergedAt: result.mergedAt } : {}),
      createdAt: new Date(this.#deps.now()).toISOString(),
    }
    const rkey = mergeRkey(artifact.uri, artifact.cid)
    try {
      let ref: { uri: string; cid: string }
      try {
        ref = await actor.client.create(COLLECTIONS.merge, record, { rkey })
      } catch (error) {
        // Restart-safe adoption: a deterministic-rkey collision means we (this actor) already wrote
        // this merge in a prior process; adopt the existing record as success rather than duplicate.
        if (
          error instanceof XrpcError &&
          error.status === 400 &&
          error.error === 'RecordAlreadyExists'
        ) {
          const existing = await actor.client.getOwnRecord(COLLECTIONS.merge, rkey)
          if (!existing) throw error
          ref = { uri: existing.uri, cid: existing.cid }
        } else {
          throw error
        }
      }
      this.#alreadyWritten.add(key)
      this.#schedules.delete(prUrl) // merged is terminal: stop scheduling this PR forever.
      this.#deps.log?.(`observed merge for ${prUrl} (${artifact.uri}) as ${actor.did}: ${ref.uri}`)
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error)
      this.#deps.log?.(`failed to write merge for ${prUrl}: ${message}; backing off`)
      this.#backoff(prUrl)
    }
  }

  #backoff(prUrl: string): void {
    const now = this.#deps.now()
    const schedule = this.#schedules.get(prUrl) ?? {
      nextPollAt: 0,
      backoffMs: this.#initialBackoff(),
    }
    schedule.nextPollAt = now + schedule.backoffMs
    schedule.backoffMs = Math.min(schedule.backoffMs * 2, this.#deps.backoffMaxMs)
    this.#schedules.set(prUrl, schedule)
  }

  /** Awaits every in-flight lookup (best-effort graceful shutdown; each promise never rejects). */
  async drain(): Promise<void> {
    await Promise.all([...this.#inFlight.values()])
  }
}