import { EventType } from "@umt/contracts";
import { OutboxRelay, type WorkerHandle } from "@umt/tasks";
import type { Logger } from "pino";
import { collectOutboxRoutes, tasksLogSink } from "../context.ts";
import { type EventBusPort, sseConnectionCount } from "../http/sse.ts";
import { type HeartbeatWriter, startHeartbeatWriter } from "../modules/system/heartbeat.ts";
import type { ServerContext, ServerModule } from "../modules/types.ts";
import {
  groupTasksByLane,
  IO_LANES,
  type IoLane,
  LaneController,
  type LaneOverrides,
  type LaneTask,
  setLaneController,
} from "./lanes.ts";

/**
 * outbox → SSE 桥:这些前缀的 outbox 事件在中继投递成功后镜像到实时总线(payload 原样,带业务 id),
 * 首页、收件箱、任务、来源与下载页据此失效刷新。feed.due 是调度器内部节拍,不推给浏览器。
 * 镜像列表同时决定中继认领哪些没有路由的事件:生命周期广播事件都在这里,否则没有订阅方时会一直挂在 pending。
 */
const SSE_MIRROR_PREFIXES = [
  "feed.",
  "decision.",
  "download.",
  "organization.",
  "subscription.",
  // 生命周期广播事件(ADR-006):没有订阅方时也要被中继认领(标记已投递),顺带推给浏览器
  "scrape.",
  "subtitle.",
  "media.",
  "site.",
] as const;
const SSE_MIRROR_EXCLUDED = new Set<string>([EventType.feedDue]);
export const SSE_MIRRORED_OUTBOX_TYPES: readonly string[] = Object.values(EventType).filter(
  (t) => SSE_MIRROR_PREFIXES.some((p) => t.startsWith(p)) && !SSE_MIRROR_EXCLUDED.has(t),
);

/** agent 组任务:整段会话占一槽数分钟,与文件复制等 io 任务分池(以任务声明的 lane = "agent" 为准,这是兜底名单) */
export { AGENT_TASKS } from "./lanes.ts";

/**
 * worker 分组:
 *   all   — 单个引擎 worker 执行全部任务(测试 / 省略分组时的兼容行为)
 *   io    — 除 agent 道外的全部任务,按 lane 分文件 / 网络 / 维护三个引擎 worker + OutboxRelay + 模块后台循环
 *   agent — 只执行 agent 道任务;不跑 relay 与模块循环
 *   split — 同一进程里 io 三条道 + agent 共四个引擎 worker(--role all)
 */
export type WorkerGroup = "all" | "io" | "agent" | "split";

export type PoolLabel = "all" | IoLane | "agent";

export interface WorkerDeps {
  logger: Logger;
  heartbeatMs: number;
  /** role=all 时与 API 共享同一总线,心跳可在 SSE 中看到 */
  bus?: EventBusPort;
  /** 业务装配;省略时只跑心跳(测试用) */
  runtime?: {
    ctx: ServerContext;
    modules: readonly ServerModule[];
    /** 设置 outbox 中继实例到 ctx.outbox.relay */
    setRelay?: (relay: OutboxRelay) => void;
    outboxIntervalMs?: number;
    /** all 组(单池)并发 */
    concurrency?: number;
    /** agent 组并发 */
    agentConcurrency?: number;
    /** 缺省 all */
    group?: WorkerGroup;
    /** 注册表里的全部任务(名称 + lane);group 为 io/agent/split 时必填 */
    tasks?: readonly LaneTask[];
    /** 分道:环境变量覆盖与测试用的节奏参数 */
    lanes?: {
      overrides?: LaneOverrides;
      refreshMs?: number;
      busyCacheMs?: number;
      maxYieldMs?: number;
    };
  };
  /** 事件循环延迟探针:采样间隔与告警阈值(测试可调小) */
  lagProbe?: { intervalMs?: number; warnMs?: number };
}

export interface WorkerStats {
  /** 事件循环偏差超过阈值的次数 */
  loopLagWarnings: number;
  /** 观测到的最大偏差(ms) */
  maxLoopLagMs: number;
}

export interface RunningWorker {
  stop(): Promise<void>;
  stats(): WorkerStats;
  /** io / split 组的分道控制器 */
  lanes: LaneController | null;
}

export interface PoolPlan {
  label: PoolLabel;
  /** 任务名(匹配该任务全部版本);all 池省略 */
  versions?: string[];
  concurrency?: number;
}

/** 按分组拆出引擎 worker 的任务选择;io 组按 lane 拆三条道(空道不起) */
export function workerPlan(
  group: WorkerGroup,
  tasks: readonly LaneTask[],
  concurrency: { all?: number; agent?: number } & LaneOverrides = {},
): PoolPlan[] {
  const { byLane } = groupTasksByLane(tasks);
  const pool = (label: IoLane | "agent"): PoolPlan[] =>
    byLane[label].length
      ? [{ label, versions: byLane[label], ...(concurrency[label] ? { concurrency: concurrency[label] } : {}) }]
      : [];
  const io = IO_LANES.flatMap((l) => pool(l));
  switch (group) {
    case "all":
      return [{ label: "all", ...(concurrency.all ? { concurrency: concurrency.all } : {}) }];
    case "io":
      return io;
    case "agent":
      return pool("agent");
    case "split":
      return [...io, ...pool("agent")];
  }
}

/**
 * 事件循环延迟探针:setInterval 实际触发时刻与预期的偏差超过 warnMs 记一次 warn + 计数。
 * 偏差大说明有同步重活占住主线程(租约续期 15s、租约 30s,偏差逼近这个量级就有丢租约风险)。
 */
export function startLoopLagProbe(log: Logger, opts: { intervalMs?: number; warnMs?: number } = {}) {
  const intervalMs = opts.intervalMs ?? 1000;
  const warnMs = opts.warnMs ?? 5000;
  const stats: WorkerStats = { loopLagWarnings: 0, maxLoopLagMs: 0 };
  let expected = performance.now() + intervalMs;
  const timer = setInterval(() => {
    const now = performance.now();
    const lag = Math.max(0, now - expected);
    expected = now + intervalMs;
    if (lag > stats.maxLoopLagMs) stats.maxLoopLagMs = Math.round(lag);
    if (lag > warnMs) {
      stats.loopLagWarnings += 1;
      log.warn({ lagMs: Math.round(lag), warnings: stats.loopLagWarnings }, "worker 事件循环延迟超过阈值");
    }
  }, intervalMs);
  return {
    stats: () => ({ ...stats }),
    stop: () => clearInterval(timer),
  };
}

/**
 * worker 角色:心跳 + 任务引擎 worker(执行全部已注册任务)+ OutboxRelay(按模块的 outboxRoutes 路由)
 * + 各模块的后台循环(start)。stop 逆序停止;引擎连接本身由进程退出时关闭。
 */
export async function startWorker({ logger, heartbeatMs, bus, runtime, lagProbe }: WorkerDeps): Promise<RunningWorker> {
  const group = runtime?.group ?? "all";
  const log = logger.child({ component: "worker", group });
  const probe = startLoopLagProbe(log, lagProbe);
  let n = 0;
  const beat = () => {
    n += 1;
    const s = probe.stats();
    log.debug({ n, ...s }, "worker heartbeat");
    bus?.publish(EventType.heartbeat, { role: group === "agent" ? "worker:agent" : "worker", n, ...s });
  };

  const started: ServerModule[] = [];
  let relay: OutboxRelay | null = null;
  let lanes: LaneController | null = null;
  const plan: PoolPlan[] = [];
  const handles = new Map<PoolLabel, WorkerHandle>();
  if (runtime) {
    const { ctx } = runtime;
    if (group !== "all" && !runtime.tasks) throw new Error(`worker 分组 ${group} 需要 tasks`);
    const tasks = runtime.tasks ?? [];
    plan.push(...workerPlan(group, tasks, { all: runtime.concurrency, agent: runtime.agentConcurrency }));
    const laned = plan.filter((p): p is PoolPlan & { label: IoLane } =>
      (IO_LANES as readonly string[]).includes(p.label),
    );
    if (laned.length) {
      const { byLane, missing } = groupTasksByLane(tasks);
      if (missing.length)
        log.warn({ tasks: missing }, "这些任务没有声明 lane，按缺省归入网络道（请在 defineTask 里补 lane）");
      lanes = new LaneController({
        engine: ctx.engine,
        logger,
        lanes: { file: byLane.file, network: byLane.network, maintenance: byLane.maintenance },
        readSettings: async () => (await ctx.settings.get("automation")).workerLanes,
        ...(runtime.lanes?.overrides ? { overrides: runtime.lanes.overrides } : {}),
        ...(runtime.lanes?.refreshMs ? { refreshMs: runtime.lanes.refreshMs } : {}),
        ...(runtime.lanes?.busyCacheMs !== undefined ? { busyCacheMs: runtime.lanes.busyCacheMs } : {}),
        ...(runtime.lanes?.maxYieldMs ? { maxYieldMs: runtime.lanes.maxYieldMs } : {}),
      });
      await lanes.start();
      setLaneController(ctx, lanes);
    }
    for (const p of plan) {
      const lane = (IO_LANES as readonly string[]).includes(p.label) ? (p.label as IoLane) : null;
      const handle =
        lane && lanes
          ? await ctx.engine.startWorker({
              versions: p.versions ?? [],
              concurrency: lanes.slots(lane),
              maxActive: () => lanes?.maxActive(lane) ?? 0,
              ...(lane === "maintenance" ? { claimGate: () => lanes?.maintenanceMayStart() ?? true } : {}),
            })
          : await ctx.engine.startWorker({
              ...(p.versions ? { versions: p.versions } : {}),
              ...(p.concurrency ? { concurrency: p.concurrency } : {}),
            });
      if (handle) handles.set(p.label, handle);
      if (lane) lanes?.attach(lane, handle || undefined);
    }
  }
  // outbox 中继与模块后台循环只在 io 组(agent 组进程只执行会话任务)
  if (runtime && group !== "agent") {
    const { ctx, modules } = runtime;
    relay = new OutboxRelay({
      db: ctx.db,
      engine: ctx.engine,
      routes: collectOutboxRoutes(modules, ctx),
      logSink: tasksLogSink(logger),
      mirror: {
        types: SSE_MIRRORED_OUTBOX_TYPES,
        publish: (ev) => ctx.events.publish(ev.type, ev.payload),
      },
      ...(runtime.outboxIntervalMs ? { intervalMs: runtime.outboxIntervalMs } : {}),
    });
    runtime.setRelay?.(relay);
    relay.start();
    for (const m of modules) {
      await m.start?.(ctx);
      started.push(m);
    }
  }

  // 进程与健康页:每 15s 把心跳/延迟/并发/运行中 run 数写进 process_heartbeat(写失败不影响 worker);
  // io 组另写各道的并发、本进程运行中、全库排队(extra.lanes),事件循环延迟照常记录(将来拆进程的判据)
  let hbWriter: HeartbeatWriter | null = null;
  if (runtime) {
    const { ctx } = runtime;
    const groupTasks = new Set(plan.flatMap((p) => p.versions ?? []));
    const countAll = plan.some((p) => !p.versions);
    const agentPool = plan.find((p) => p.label === "agent");
    hbWriter = startHeartbeatWriter({
      db: ctx.db,
      logger,
      kind: group === "split" ? "all" : group === "agent" ? "worker:agent" : "worker",
      sample: async () => {
        const running = await ctx.engine.list({ status: "running", limit: 500 }).catch(() => null);
        const s = probe.stats();
        const laneStats = lanes ? await lanes.snapshot() : null;
        const concurrency: Record<string, number> = {};
        if (laneStats) {
          for (const l of laneStats) concurrency[l.lane] = l.concurrency;
          concurrency.io = laneStats.reduce((a, l) => a + l.concurrency, 0);
        } else if (plan.some((p) => p.label === "all")) {
          concurrency.io = runtime.concurrency ?? 4;
        }
        if (agentPool) concurrency.agent = handles.get("agent")?.concurrency ?? runtime.agentConcurrency ?? 4;
        return {
          concurrency,
          runningRuns: running
            ? running.filter((r) => r.status === "running" && (countAll || groupTasks.has(r.taskName))).length
            : null,
          sseConnections: group === "split" ? sseConnectionCount() : null,
          extra: {
            group,
            lagWarnings: s.loopLagWarnings,
            probeMaxLagMs: s.maxLoopLagMs,
            ...(laneStats ? { lanes: laneStats } : {}),
          },
        };
      },
    });
  }

  log.info(
    {
      heartbeatMs,
      modules: started.map((m) => m.name),
      pools: plan.map((p) => ({
        group: p.label,
        concurrency:
          lanes && (IO_LANES as readonly string[]).includes(p.label)
            ? lanes.limit(p.label as IoLane)
            : (p.concurrency ?? 4),
        tasks: p.versions?.length ?? "all",
      })),
    },
    "worker started",
  );
  beat();
  const timer = setInterval(beat, heartbeatMs);
  return {
    stats: () => probe.stats(),
    lanes,
    async stop() {
      clearInterval(timer);
      probe.stop();
      lanes?.stop();
      if (runtime) setLaneController(runtime.ctx, null);
      await hbWriter?.stop();
      for (const m of started.reverse()) {
        try {
          await m.stop?.();
        } catch (err) {
          log.error({ err, module: m.name }, "模块停止失败");
        }
      }
      await relay?.stop();
      log.info("worker stopped");
    },
  };
}