/**
 * Wires the skill module to the upstream event buses.
 *
 * Upstream triggers (all debounced via `queueMicrotask` so they never block
 * the emitter):
 *
 *   - `l2.policy.induced`        → `runSkill({ trigger, policyId })`
 *   - `l2.policy.status_changed` → `runSkill({ trigger, policyId })` when
 *                                  the new status is `active`
 *   - `reward.updated`           → one scoped run per policy linked to the
 *                                  updated episode. Each policy has its own
 *                                  cooldown and pending queue entry. Also
 *                                  drives η adjustment on existing skills.
 *
 * The handle returns `runOnce` for manual runs (used by the CLI / viewer
 * rebuild button) and `applyFeedback` for explicit skill feedback.
 */

import type { L2Event, L2EventBus } from "../memory/l2/types.js";
import type { Logger } from "../logger/types.js";
import type { RewardEvent, RewardEventBus } from "../reward/types.js";
import { rootLogger } from "../logger/index.js";
import {
  applySkillFeedback,
  runSkill,
  type RunSkillDeps,
} from "./skill.js";
import { shouldPromoteCandidate } from "./lifecycle.js";
import type {
  RunSkillInput,
  RunSkillResult,
  SkillEventBus,
  SkillFeedbackKind,
  SkillTrigger,
} from "./types.js";
import type { PolicyId, SkillId } from "../types.js";
import { now as nowMs } from "../time.js";
import { IDLE_ARCHIVE_BATCH_LIMIT } from "../storage/repos/skills.js";

/** Bound one lifecycle pass to ten repository-sized archival batches. */
const IDLE_ARCHIVE_MAX_BATCHES_PER_TICK = 10;

function yieldToEventLoop(): Promise<void> {
  return new Promise((resolve) => setImmediate(resolve));
}

export interface SkillSubscriberDeps
  extends Omit<RunSkillDeps, "log" | "bus"> {
  log?: Logger;
  bus: SkillEventBus;
  l2Bus: L2EventBus;
  rewardBus: RewardEventBus;
}

export interface SkillSubscriberHandle {
  dispose(): void;
  runOnce(input: Omit<RunSkillInput, "trigger"> & { trigger?: SkillTrigger }): Promise<RunSkillResult>;
  applyFeedback(skillId: SkillId, kind: SkillFeedbackKind, magnitude?: number): void;
  lifecycleTick(): Promise<void>;
  /**
   * Await any in-flight scheduled run. Primarily useful in tests where we
   * want to assert on the effects of an event-driven run after the bus has
   * fanned out the event.
   */
  flush(): Promise<void>;
}

export function attachSkillSubscriber(
  deps: SkillSubscriberDeps,
): SkillSubscriberHandle {
  const log = deps.log ?? rootLogger.child({ channel: "core.skill" });
  const runDeps: RunSkillDeps = {
    repos: deps.repos,
    embedder: deps.embedder,
    llm: deps.llm,
    log,
    bus: deps.bus,
    config: deps.config,
  };

  let inflight: Promise<void> | null = null;
  let disposed = false;
  let rewardTimer: ReturnType<typeof setTimeout> | null = null;
  const lastRewardRunAt = new Map<PolicyId, number>();
  const pendingRewardPolicies = new Set<PolicyId>();
  const queued: Array<{
    trigger: SkillTrigger;
    hint?: { policyId?: PolicyId; skillId?: SkillId };
  }> = [];

  async function drain(): Promise<void> {
    while (queued.length > 0) {
      const next = queued.shift()!;
      try {
        await runSkill(
          { trigger: next.trigger, policyId: next.hint?.policyId, skillId: next.hint?.skillId },
          runDeps,
        );
      } catch (err) {
        log.error("skill.run.failed", {
          trigger: next.trigger,
          err: err instanceof Error ? err.message : String(err),
        });
      }
    }
  }

  function triggerRun(
    trigger: SkillTrigger,
    hint?: { policyId?: PolicyId; skillId?: SkillId },
  ): void {
    queued.push({ trigger, hint });
    if (inflight) {
      log.debug("skill.run.queued", { trigger });
      return;
    }
    const promise = drain().finally(() => {
      if (inflight === promise) inflight = null;
    });
    inflight = promise;
  }

  function scheduleRewardRuns(): void {
    if (disposed) return;
    if (rewardTimer) {
      clearTimeout(rewardTimer);
      rewardTimer = null;
    }
    const cooldownMs = Math.max(0, deps.config.cooldownMs);
    const at = nowMs();
    let nextDelay: number | null = null;
    for (const policyId of pendingRewardPolicies) {
      const lastRunAt = lastRewardRunAt.get(policyId);
      const remainingMs = lastRunAt === undefined ? 0 : cooldownMs - (at - lastRunAt);
      if (remainingMs > 0) {
        nextDelay = nextDelay === null ? remainingMs : Math.min(nextDelay, remainingMs);
        log.debug("skill.run.cooldown", {
          trigger: "reward.updated",
          policyId,
          remainingMs,
        });
        continue;
      }
      pendingRewardPolicies.delete(policyId);
      lastRewardRunAt.set(policyId, at);
      triggerRun("reward.updated", { policyId });
    }
    if (nextDelay !== null && pendingRewardPolicies.size > 0) {
      rewardTimer = setTimeout(scheduleRewardRuns, nextDelay);
    }
  }

  function triggerRewardRuns(policyIds: readonly PolicyId[]): void {
    for (const policyId of policyIds) pendingRewardPolicies.add(policyId);
    scheduleRewardRuns();
  }

  const offInduced = deps.l2Bus.on("l2.policy.induced", (evt: L2Event) => {
    if (evt.kind !== "l2.policy.induced") return;
    log.debug("trigger.l2.policy.induced", { policyId: evt.policyId });
    triggerRun("l2.policy.induced", { policyId: evt.policyId });
  });

  const offStatus = deps.l2Bus.on("l2.policy.updated", (evt: L2Event) => {
    if (evt.kind !== "l2.policy.updated") return;
    if (evt.status !== "active") return;
    log.debug("trigger.l2.policy.updated", { policyId: evt.policyId, status: evt.status });
    triggerRun("l2.policy.status_changed", { policyId: evt.policyId });
  });

  const offReward = deps.rewardBus.on("reward.updated", (evt: RewardEvent) => {
    if (evt.kind !== "reward.updated") return;
    log.debug("trigger.reward.updated", {
      episodeId: evt.result.episodeId,
    });
    resolveTrialsForReward(evt);
    const linkedPolicyIds = deps.repos.tracePolicyLinks.getLinkedPolicyIds(evt.result.episodeId);
    const sourceEpisodePolicyIds = deps.repos.policies
      .list({ status: "active", limit: 200 })
      .filter((policy) => policy.sourceEpisodeIds.includes(evt.result.episodeId))
      .map((policy) => policy.id);
    const relatedPolicyIds = Array.from(
      new Set([...linkedPolicyIds, ...sourceEpisodePolicyIds]),
    );
    triggerRewardRuns(relatedPolicyIds);
  });

  function dispose(): void {
    disposed = true;
    if (rewardTimer) clearTimeout(rewardTimer);
    rewardTimer = null;
    pendingRewardPolicies.clear();
    offInduced();
    offStatus();
    offReward();
    log.info("skill.subscriber.disposed");
  }

  async function runOnce(
    input: Omit<RunSkillInput, "trigger"> & { trigger?: SkillTrigger },
  ): Promise<RunSkillResult> {
    const trigger: SkillTrigger = input.trigger ?? "manual";
    return runSkill(
      {
        trigger,
        policyId: input.policyId,
        skillId: input.skillId,
      },
      runDeps,
    );
  }

  function applyFeedback(
    skillId: SkillId,
    kind: SkillFeedbackKind,
    magnitude?: number,
  ): void {
    applySkillFeedback(skillId, kind, runDeps, magnitude);
  }

  function resolveTrialsForReward(evt: Extract<RewardEvent, { kind: "reward.updated" }>): void {
    const rTask = evt.result.rHuman;
    const outcome =
      rTask >= 0.5 ? "pass" :
      rTask <= -0.5 ? "fail" :
      "unknown";
    const trials = deps.repos.skillTrials.listPendingForEpisode(evt.result.episodeId);
    if (trials.length === 0) return;
    for (const trial of trials) {
      const evidence = {
        source: "reward.updated",
        episodeId: evt.result.episodeId,
        rTask,
        threshold: { pass: 0.5, fail: -0.5 },
        reason:
          outcome === "pass"
            ? "rTask >= 0.5"
            : outcome === "fail"
              ? "rTask <= -0.5"
              : "-0.5 < rTask < 0.5",
      };
      const changed = deps.repos.skillTrials.resolve(
        trial.id,
        outcome,
        evt.result.completedAt,
        evidence,
      );
      if (!changed) continue;
      if (outcome === "pass" || outcome === "fail") {
        applySkillFeedback(
          trial.skillId,
          outcome === "pass" ? "trial.pass" : "trial.fail",
          runDeps,
        );
      }
      log.info("skill.trial.resolved", {
        trialId: trial.id,
        skillId: trial.skillId,
        episodeId: evt.result.episodeId,
        outcome,
        rTask,
      });
    }
  }

  async function flush(): Promise<void> {
    // Loop in case additional events arrive while we're draining.
    while (inflight) {
      await inflight;
    }
  }

  /** Promote eligible candidates and archive stale low-η active skills. */
  async function lifecycleTick(): Promise<void> {
    const at = nowMs();
    const candidates = deps.repos.skills.list({ status: "candidate", limit: 500 });
    for (const s of candidates) {
      if (!shouldPromoteCandidate(s, deps.config)) continue;
      deps.repos.skills.setStatus(s.id, "active", at);
      log.info("skill.auto_promoted", { skillId: s.id, name: s.name, eta: s.eta });
      deps.bus.emit({
        kind: "skill.status.changed",
        at,
        skillId: s.id,
        previous: "candidate",
        next: "active",
        transition: "promoted",
      });
    }

    const cutoff = at - deps.config.idleArchiveMs;
    let batchesProcessed = 0;
    let archivedTotal = 0;
    while (batchesProcessed < IDLE_ARCHIVE_MAX_BATCHES_PER_TICK) {
      const archivedSkills = deps.repos.skills.archiveNextIdleBatch({
        minEtaForRetrieval: deps.config.minEtaForRetrieval,
        cutoff,
        updatedAt: at,
        limit: IDLE_ARCHIVE_BATCH_LIMIT,
      });
      batchesProcessed += 1;
      const archivedThisBatch = archivedSkills.length;
      archivedTotal += archivedThisBatch;
      for (const s of archivedSkills) {
        log.debug("skill.idle_archived", {
          skillId: s.id,
          name: s.name,
          eta: s.eta,
          lastUsedAt: s.lastUsedAt ?? null,
          idleArchiveMs: deps.config.idleArchiveMs,
        });
        deps.bus.emit({
          kind: "skill.status.changed",
          at,
          skillId: s.id,
          previous: "active",
          next: "archived",
          transition: "archived",
        });
      }
      if (archivedThisBatch > 0) {
        log.info("skill.idle_archive_batch", {
          batchCount: batchesProcessed,
          archivedCount: archivedThisBatch,
          cutoff,
          minEtaForRetrieval: deps.config.minEtaForRetrieval,
        });
      }
      if (archivedThisBatch < IDLE_ARCHIVE_BATCH_LIMIT) break;
      if (batchesProcessed === IDLE_ARCHIVE_MAX_BATCHES_PER_TICK) {
        log.warn("skill.idle_archive_batch_limit_reached", {
          batchCount: batchesProcessed,
          archivedCount: archivedTotal,
          batchSize: IDLE_ARCHIVE_BATCH_LIMIT,
        });
      } else {
        await yieldToEventLoop();
      }
    }
  }

  return { dispose, runOnce, applyFeedback, flush, lifecycleTick };
}