import { DatabaseSync } from "node:sqlite";
import { mkdirSync } from "node:fs";
import { dirname } from "node:path";
import { join } from "node:path";
import { MEMORY_DB_DIR, PIPELINE_JOBS_FILE } from "../paths.ts";
import { logEvent } from "../../util/notice.ts";

export type PipelineJobType = "l1" | "l2" | "l3";
/**
 * `drain` = L1 收尾后因「L0 仍有积压」而连续排的下一批（2026-09-19）。
 * `trigger` = L3 的台账式触发（`agent_settled` 五级门控命中）。
 * `signal`  = L3 的 **P1 显式信号**（`l2_commit` 的 `personaSignal`）。
 * `manual`  = 手动命令（`/my-memory:l*-trigger`），执行器据此 force 掉门控与熔断。
 *
 * 为什么 `signal` 必须与 `trigger` 分开（2026-09-21 修）：L3 的 executeJob 会在 claim 时
 * 重估门控，而重估时没有信号上下文 → P1 丢了，只剩 P2–P4，于是"L2 刚提交、明确要求刷新画像"
 * 这个意图被静默降级成"门控不成立 → skipped"（实测 job 51：入队后 5ms 就 skipped）。
 * 来源即意图，必须带到 claim 时。
 */
export type PipelineJobSource = "interval" | "manual" | "l1_commit" | "drain" | "trigger" | "signal";
export type PipelineJobStatus = "queued" | "running" | "done" | "skipped" | "failed";

/** recoverStuckRunning 只回收 started_at 早于该阈值的 running（防子进程误杀刚启动的 job）。 */
export const PIPELINE_RECOVER_MIN_AGE_MS = 10 * 60 * 1000;

export interface PipelineJob {
	id: number;
	type: PipelineJobType;
	source: PipelineJobSource;
	status: PipelineJobStatus;
	createdAt: string;
	startedAt: string | null;
	finishedAt: string | null;
	error: string | null;
}

export interface EnqueueResult {
	id: number;
	duplicate: boolean;
}

const SCHEMA_SQL = `
CREATE TABLE IF NOT EXISTS jobs (
  id          INTEGER PRIMARY KEY AUTOINCREMENT,
  type        TEXT NOT NULL,
  source      TEXT NOT NULL,
  status      TEXT NOT NULL,
  created_at  TEXT NOT NULL,
  started_at  TEXT,
  finished_at TEXT,
  error       TEXT
);

CREATE INDEX IF NOT EXISTS idx_jobs_status_id ON jobs(status, id);
`;

interface Row {
	id: number;
	type: string;
	source: string;
	status: string;
	created_at: string;
	started_at: string | null;
	finished_at: string | null;
	error: string | null;
}

function mapRow(r: Row): PipelineJob {
	return {
		id: r.id,
		type: r.type as PipelineJobType,
		source: r.source as PipelineJobSource,
		status: r.status as PipelineJobStatus,
		createdAt: r.created_at,
		startedAt: r.started_at,
		finishedAt: r.finished_at,
		error: r.error,
	};
}

/**
 * L1/L2/L3 流水线队列（node:sqlite）。
 * 主进程 worker 与子代理进程（l1_commit / l2_commit / l3_commit）都会打开同一文件；WAL + busy_timeout 防并发写。
 * L3 自 2026-09-19 起也走这里（此前只有 L1/L2，L3 用 `l3/run.lock` 自己单飞）。
 */
export class PipelineJobsDB {
	private db: DatabaseSync;
	private readonly profilePath: string;
	readonly dbPath: string;

	constructor(profilePath: string) {
		this.profilePath = profilePath;
		const memDir = join(profilePath, MEMORY_DB_DIR);
		mkdirSync(memDir, { recursive: true });
		this.dbPath = join(memDir, PIPELINE_JOBS_FILE);
		// 建到**文件所在目录**（PIPELINE_JOBS_FILE 是多段相对路径 `pipeline/jobs.sqlite`）：
		// 只建 memoryDB/ 的话，全新 profile 会 "unable to open database file"
		//（生产 profile 里 pipeline/ 恰好已存在，所以这个缺口一直没暴露）
		mkdirSync(dirname(this.dbPath), { recursive: true });
		this.db = new DatabaseSync(this.dbPath);
		this.db.exec("PRAGMA journal_mode = WAL;");
		this.db.exec("PRAGMA busy_timeout = 5000;");
		this.db.exec(SCHEMA_SQL);
	}

	/** 已有同 type 的 queued 则不插入；running 也视为 duplicate（不堆第二条）。 */
	enqueue(type: PipelineJobType, source: PipelineJobSource): EnqueueResult {
		return this.transaction(() => {
			const existing = this.db
				.prepare("SELECT * FROM jobs WHERE type = ? AND status IN ('queued', 'running') ORDER BY id ASC LIMIT 1")
				.get(type) as Row | undefined;
			if (existing) return { id: existing.id, duplicate: true };

			const now = new Date().toISOString();
			const info = this.db.prepare("INSERT INTO jobs (type, source, status, created_at) VALUES (?, ?, 'queued', ?)").run(type, source, now);
			return { id: Number(info.lastInsertRowid), duplicate: false };
		});
	}

	/** 写事务保护读后写；失败时保留原始错误，并释放 SQLite 写锁。 */
	private transaction<T>(run: () => T): T {
		this.db.exec("BEGIN IMMEDIATE");
		try {
			const result = run();
			this.db.exec("COMMIT");
			return result;
		} catch (err) {
			try { this.db.exec("ROLLBACK"); } catch { /* 不覆盖原始错误 */ }
			throw err;
		}
	}

	/**
	 * 当前 queued 的全部 job，**按 claim 优先级排序**（L3 永远最后，其余按入队序）。
	 *
	 * 这是"谁下一个跑"的唯一出口：`worker.drain` 拿它跳过熔断层之后挑一个。
	 * 早先还有个只取第一条的 `peekNext()`（2026-09-21 删）—— 它让"队首恰好是熔断层"变成
	 * 整条队列停摆，见 `worker.ts` 的 `pickNextClaimable`。
	 */
	listQueued(): PipelineJob[] {
		const rows = this.db
			.prepare(
				"SELECT * FROM jobs WHERE status = 'queued' ORDER BY CASE type WHEN 'l3' THEN 1 ELSE 0 END ASC, id ASC",
			)
			.all() as Row[];
		return rows.map(mapRow);
	}

	/** queued → running 与全局空闲检查是同一条写语句，跨进程也只能认领一个。 */
	claim(id: number): PipelineJob | undefined {
		const now = new Date().toISOString();
		const row = this.db.prepare(
			`UPDATE jobs SET status = 'running', started_at = ?
			 WHERE id = ? AND status = 'queued'
			 AND NOT EXISTS (SELECT 1 FROM jobs WHERE status = 'running') RETURNING *`,
		).get(now, id) as Row | undefined;
		return row ? mapRow(row) : undefined;
	}

	getJob(id: number): PipelineJob | undefined {
		const row = this.db.prepare("SELECT * FROM jobs WHERE id = ?").get(id) as Row | undefined;
		return row ? mapRow(row) : undefined;
	}

	hasActive(type: PipelineJobType): boolean {
		const row = this.db.prepare("SELECT 1 AS n FROM jobs WHERE type = ? AND status IN ('queued', 'running') LIMIT 1").get(type) as
			| { n: number }
			| undefined;
		return Boolean(row);
	}

	complete(id: number, status: "done" | "skipped" | "failed", error?: string): void {
		const now = new Date().toISOString();
		this.db
			.prepare("UPDATE jobs SET status = ?, finished_at = ?, error = ? WHERE id = ? AND status IN ('queued', 'running')")
			.run(status, now, error ?? null, id);
	}

	/** l1_commit：把当前 running 的 l1 标 done（没有也不抛错）。 */
	completeRunningL1(): number {
		return this.completeRunning("l1");
	}

	/**
	 * l2_commit：把当前 running 的 l2 标 done。
	 *
	 * 为什么 L2 也需要它：`runPiSubagent` 是 async:true（派发后毫秒级返回），所以
	 * `l2/index.ts` 里 `await runPiSubagent(...)` 之后紧接着读游标**必然还没推进** ——
	 * 旧代码据此就把 job 标 done，于是任务在 0 秒"完成"、队列以为空闲、立刻又 claim 下一条
	 * → 2026-09-19 实测两个 L2 子代理并发改同一批场景块（与 L3 那次并发同源）。
	 * 真完成信号只能由子代理自己写：它提交成功时调这个方法。
	 */
	completeRunningL2(): number {
		return this.completeRunning("l2");
	}

	/**
	 * l3_commit：把当前 running 的 l3 标 done。
	 *
	 * 与 L2 同源：`runPiSubagent` 是 async:true（派发后毫秒级返回），所以"派发完就读游标"
	 * 必然还是旧值。L3 并入队列后，唯一可信的完成信号就是子代理自己提交时调的这一个
	 * （取代了此前的 `l3/run.lock` + 主进程观察游标那一套，见 §11.11）。
	 */
	completeRunningL3(): number {
		return this.completeRunning("l3");
	}

	/**
	 * 把当前 running 的指定层标 done（子代理提交成功时调用；没有匹配行返回 0）。
	 */
	completeRunning(type: PipelineJobType): number {
		return this.transaction(() => {
			const rows = this.db.prepare("SELECT id FROM jobs WHERE type = ? AND status = 'running' LIMIT 2").all(type) as { id: number }[];
			if (rows.length > 1) {
				const message = `multiple running ${type} jobs: refuse ambiguous completion (${rows.map((r) => r.id).join(", ")})`;
				logEvent("pipeline", message, this.profilePath);
				throw new Error(message);
			}
			if (!rows.length) return 0;
			const info = this.db.prepare("UPDATE jobs SET status = 'done', finished_at = ? WHERE id = ? AND status = 'running'")
				.run(new Date().toISOString(), rows[0].id);
			return Number(info.changes);
		});
	}

	/**
	 * 是否有**任何** job 处于 running（跨层、跨进程）。
	 *
	 * 用于调度前快速退出；查询结果可能立即过时。
	 * 跨进程互斥由 claim() 内部的原子条件 UPDATE 保证。
	 */
	hasAnyRunning(): boolean {
		const row = this.db.prepare("SELECT 1 AS n FROM jobs WHERE status = 'running' LIMIT 1").get() as
			| { n: number }
			| undefined;
		return Boolean(row);
	}

	/**
	 * 认领「超时误判」的 job —— 把最近被标 failed 的超时 job 纠正为 done（各层的薄封装见下）。
	 *
	 * 背景（2026-09-15 L1 实测，2026-09-21 L2 复现）：子代理实际耗时可能远超超时值。
	 * executeJob 等满超时后把 job 标 failed 并 abort，但**abort 不保证杀死子代理** ——
	 * 实测 L1 那次它继续跑到 `l1_commit`（job 70：00:03:27 标 failed，00:10:47 才提交），
	 * L2 那次同样在第 6.5 分钟还在改场景块。此时：
	 *   - `completeRunning*()` 的 `status='running'` 匹配不到（job 已 failed）→ 空操作；
	 *   - 主进程的计数重置分支（`after.status==='done'`）永远进不去
	 *     → consecutiveFailures 只增不减，累计 3 次即熔断，整层自动提取停摆。
	 *
	 * 只认领**最近 N 分钟内**、error 恰为该层超时文案的 failed job —— 避免把真正的失败
	 * （recovered / 其它错误）洗成成功。
	 *
	 * @param withinMs 只认领 finished_at 在此时长内的 failed job（默认 15 分钟）
	 * @returns 被纠正的 job 数
	 */
	private reclaimTimedOut(type: PipelineJobType, errorText: string, withinMs: number): number {
		const now = new Date();
		const cutoff = new Date(now.getTime() - withinMs).toISOString();
		const info = this.db
			.prepare(
				`UPDATE jobs SET status = 'done', finished_at = ?, error = NULL
				 WHERE type = ? AND status = 'failed'
				   AND error = ?
				   AND finished_at IS NOT NULL AND finished_at >= ?`,
			)
			.run(now.toISOString(), type, errorText, cutoff);
		return Number(info.changes);
	}

	/** l1_commit：认领超时误判的 l1 job（错误文案必须精确等于执行器写的那句）。 */
	reclaimTimedOutL1(withinMs = 15 * 60_000): number {
		return this.reclaimTimedOut("l1", "timeout without l1_commit", withinMs);
	}

	/**
	 * l2_commit：认领超时误判的 l2 job（2026-09-21 补，与 L1 同款）。
	 *
	 * 为什么 L2 也需要：L2 的超时曾只有 180s，连续四次（jobs 55/61/64/68）都在 3.5 分钟被判
	 * failed，而子代理还在跑、最后照样提交 —— 于是"活干了、账记失败"，三次就熔断。
	 * 超时已提到 600s（见 l2/config.ts），这条认领兜住"更慢但仍成功"的那批。
	 */
	reclaimTimedOutL2(withinMs = 15 * 60_000): number {
		return this.reclaimTimedOut("l2", "timeout without l2_commit", withinMs);
	}

	/** 进程重启：只回收「足够旧」的 running，避免 L1 子进程 session_start 误杀刚 spawn 的 job。 */
	recoverStuckRunning(minAgeMs = PIPELINE_RECOVER_MIN_AGE_MS): number {
		const cutoff = Date.now() - minAgeMs;
		const rows = this.db.prepare("SELECT * FROM jobs WHERE status = 'running'").all() as Row[];
		const finishedAt = new Date().toISOString();
		let n = 0;
		const upd = this.db.prepare(
			"UPDATE jobs SET status = 'failed', finished_at = ?, error = 'recovered: leftover running' WHERE id = ? AND status = 'running'",
		);
		for (const r of rows) {
			const started = r.started_at ? Date.parse(r.started_at) : NaN;
			if (Number.isFinite(started) && started > cutoff) continue;
			if (upd.run(finishedAt, r.id).changes === 1) n++;
		}
		return n;
	}

	close(): void {
		try {
			this.db.close();
		} catch {
			// already closed
		}
	}
}

interface DatabaseEntry { db: PipelineJobsDB; users: number; closeRequested: boolean }
const instances = new Map<string, DatabaseEntry>();

export function getPipelineJobsDB(profilePath?: string): PipelineJobsDB {
	const p = profilePath ?? process.env.PI_PROFILE_PATH ?? "";
	if (!p) throw new Error("PipelineJobsDB: profilePath 未提供且 PI_PROFILE_PATH 未设置");
	let entry = instances.get(p);
	if (!entry) {
		entry = { db: new PipelineJobsDB(p), users: 0, closeRequested: false };
		instances.set(p, entry);
	}
	return entry.db;
}

/** 会话与在途执行各持一个租约；reset 不得关掉仍有执行器/轮询使用的连接。 */
export function retainPipelineJobsDB(profilePath: string): { db: PipelineJobsDB; release: () => void } {
	const db = getPipelineJobsDB(profilePath);
	const entry = instances.get(profilePath)!;
	entry.users++;
	entry.closeRequested = false;
	let released = false;
	return { db, release: () => {
		if (released) return;
		released = true;
		entry.users--;
		if (entry.users === 0 && entry.closeRequested && instances.get(profilePath) === entry) {
			db.close();
			instances.delete(profilePath);
		}
	} };
}

export function closePipelineJobsDB(profilePath?: string, expected?: PipelineJobsDB): void {
	const p = profilePath ?? process.env.PI_PROFILE_PATH ?? "";
	const entry = instances.get(p);
	if (entry && (!expected || entry.db === expected)) {
		entry.closeRequested = true;
		if (entry.users > 0) return;
		entry.db.close();
		instances.delete(p);
	}
}