service.ts

import type { Database } from "bun:sqlite";
import { randomUUID } from "node:crypto";
import type { AgentRuntime } from "../agent/runtime.ts";
import type { SlackTransport } from "../channels/slack-transport.ts";
import { validateCreateInput } from "./create-validation.ts";
import { executeJob } from "./executor.ts";
import { type SchedulerHealthSummary, computeHealthSummary } from "./health.ts";
import { cleanupOldTerminalJobs, staggerMissedJobs } from "./recovery.ts";
import { rowToJob } from "./row-mapper.ts";
import { computeNextRunAt, serializeScheduleValue, validateSchedule } from "./schedule.ts";
import type { JobUpdateInputParsed } from "./tool-schema.ts";
import { type JobCreateInput, type JobRow, type ScheduledJob, isValidSlackTarget } from "./types.ts";

// Upper bound on the setTimeout delay we pass when arming the next wake-up.
// Both Node and Bun use a 32-bit signed integer for the setTimeout delay, so
// any value above 2^31-1 ms (roughly 24.8 days) silently coerces to about one
// millisecond, which would turn armTimer -> onTimer -> armTimer into a hot
// spin loop for any job whose next fire is more than a few weeks out (long
// at-schedules, cron expressions whose next firing is far in the future,
// every-schedules with large intervals). One hour gives us a ~600x safety
// margin under the overflow boundary while keeping the idle re-arm cost to
// one indexed SQL MIN query per hour, which is effectively free.
const MAX_TIMER_MS = 60 * 60 * 1000;

type SchedulerDeps = {
	db: Database;
	runtime: AgentRuntime;
	slackChannel?: SlackTransport;
	ownerUserId?: string | null;
};

export class Scheduler {
	private db: Database;
	private runtime: AgentRuntime;
	private slackChannel: SlackTransport | undefined;
	private ownerUserId: string | null;
	private timer: ReturnType<typeof setTimeout> | null = null;
	private running = false;
	private executing = false;
	private jobCompleteCallbacks: ((jobName: string, status: string) => void)[] = [];

	constructor(deps: SchedulerDeps) {
		this.db = deps.db;
		this.runtime = deps.runtime;
		this.slackChannel = deps.slackChannel;
		this.ownerUserId = deps.ownerUserId ?? null;
	}

	onJobComplete(cb: (jobName: string, status: string) => void): void {
		this.jobCompleteCallbacks.push(cb);
	}

	/**
	 * Inject the Slack channel after construction. ownerUserId may be null
	 * (C3): owner-targeted delivery is skipped until ownerUserId is set, but
	 * channel-id (C...) and user-id (U...) targets work immediately.
	 */
	setSlackChannel(channel: SlackTransport, ownerUserId: string | null): void {
		this.slackChannel = channel;
		this.ownerUserId = ownerUserId ?? null;
	}

	async start(): Promise<void> {
		if (this.running) return;
		this.running = true;

		// Non-blocking recovery (M1): rewrite next_run_at on past-due rows and
		// let the normal onTimer loop pick them up in sequence. start() returns
		// in milliseconds instead of blocking boot for minutes.
		staggerMissedJobs(this.db);
		cleanupOldTerminalJobs(this.db);
		this.armTimer();
		console.log("[scheduler] Started");
	}

	stop(): void {
		this.running = false;
		if (this.timer) {
			clearTimeout(this.timer);
			this.timer = null;
		}
		console.log("[scheduler] Stopped");
	}

	isRunning(): boolean {
		return this.running;
	}

	createJob(input: JobCreateInput): ScheduledJob {
		// All creation validation lives in one place so the failure modes are
		// obvious and the happy path in this method stays small. See C1, C4,
		// N1, N8, OOS#6.
		const delivery = validateCreateInput(this.db, input);

		const id = randomUUID();
		const scheduleValue = serializeScheduleValue(input.schedule);
		const nextRun = computeNextRunAt(input.schedule);
		if (!nextRun) {
			throw new Error("invalid schedule: validator passed but computeNextRunAt returned null");
		}

		this.db.run(
			`INSERT INTO scheduled_jobs (id, name, description, enabled, schedule_kind, schedule_value, task, delivery_channel, delivery_target, next_run_at, delete_after_run, created_by)
			 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			[
				id,
				input.name,
				input.description ?? null,
				input.enabled === false ? 0 : 1,
				input.schedule.kind,
				scheduleValue,
				input.task,
				delivery.channel,
				delivery.target,
				nextRun.toISOString(),
				input.deleteAfterRun ? 1 : 0,
				input.createdBy ?? "agent",
			],
		);

		this.armTimer();

		const created = this.getJob(id);
		if (!created) throw new Error(`failed to create job: ${id}`);
		return created;
	}

	deleteJob(id: string): boolean {
		const result = this.db.run("DELETE FROM scheduled_jobs WHERE id = ?", [id]);
		if (result.changes > 0) {
			this.armTimer();
			return true;
		}
		return false;
	}

	/**
	 * Flip an active job to paused. armTimer already filters by
	 * `status = 'active'` so the paused row stops firing automatically.
	 * Returns the updated job, or null if the id does not exist.
	 */
	pauseJob(id: string): ScheduledJob | null {
		const result = this.db.run(
			"UPDATE scheduled_jobs SET status = 'paused', updated_at = datetime('now') WHERE id = ? AND status = 'active'",
			[id],
		);
		if (result.changes === 0) {
			return this.getJob(id);
		}
		this.armTimer();
		return this.getJob(id);
	}

	/**
	 * Flip a paused job back to active. Recomputes next_run_at from the stored
	 * schedule so a job paused mid-interval resumes on a fresh cadence.
	 * Resets consecutive_errors so a job paused in its backoff fan-out gets a
	 * clean retry budget. Returns the updated job, or null if the id does not
	 * exist.
	 */
	resumeJob(id: string): ScheduledJob | null {
		const job = this.getJob(id);
		if (!job) return null;
		// Only paused jobs may be resumed. Failed and completed are terminal
		// states; force-reviving them would bypass the lifecycle (e.g.,
		// re-running a one-shot that already deleted itself, or restarting a
		// circuit-broken job without addressing the failure).
		if (job.status !== "paused") return job;
		const nextRun = computeNextRunAt(job.schedule);
		const nextRunIso = nextRun ? nextRun.toISOString() : null;
		this.db.run(
			`UPDATE scheduled_jobs
				SET status = 'active',
					next_run_at = ?,
					consecutive_errors = 0,
					updated_at = datetime('now')
				WHERE id = ? AND status = 'paused'`,
			[nextRunIso, id],
		);
		this.armTimer();
		return this.getJob(id);
	}

	/**
	 * Edit a job's user-authored columns in place. History is preserved:
	 * last_run_at, last_run_status, run_count, consecutive_errors, created_at,
	 * and the stable `id` are never touched. The caller chooses which subset
	 * of {task, description, schedule, delivery, enabled} to change; the
	 * refine() on JobUpdateInputSchema enforces that at least one is present,
	 * so an empty partial here is treated as a no-op (defense in depth).
	 *
	 * When `schedule` changes we recompute next_run_at via the same
	 * computeNextRunAt path resumeJob uses, then armTimer() so a schedule
	 * that pulls the next fire earlier wakes the timer in time. Returns the
	 * fresh row, or null if `id` does not exist.
	 */
	updateJob(id: string, partial: JobUpdateInputParsed): ScheduledJob | null {
		const job = this.getJob(id);
		if (!job) return null;

		const sets: string[] = [];
		const params: Array<string | number | null> = [];

		if (partial.task !== undefined) {
			sets.push("task = ?");
			params.push(partial.task);
		}

		if (partial.description !== undefined) {
			sets.push("description = ?");
			params.push(partial.description);
		}

		if (partial.delivery !== undefined) {
			// Mirror create-validation's slack-target check so an update cannot
			// install a malformed target the create path would have rejected.
			if (partial.delivery.channel === "slack" && !isValidSlackTarget(partial.delivery.target)) {
				throw new Error(
					`invalid delivery.target '${partial.delivery.target}': must be "owner", a Slack channel id (C...), or a Slack user id (U...)`,
				);
			}
			sets.push("delivery_channel = ?");
			params.push(partial.delivery.channel);
			sets.push("delivery_target = ?");
			params.push(partial.delivery.target);
		}

		if (partial.enabled !== undefined) {
			sets.push("enabled = ?");
			params.push(partial.enabled ? 1 : 0);
		}

		if (partial.schedule !== undefined) {
			const scheduleError = validateSchedule(partial.schedule);
			if (scheduleError) {
				throw new Error(`invalid schedule: ${scheduleError}`);
			}
			const nextRun = computeNextRunAt(partial.schedule);
			const nextRunIso = nextRun ? nextRun.toISOString() : null;
			sets.push("schedule_kind = ?");
			params.push(partial.schedule.kind);
			sets.push("schedule_value = ?");
			params.push(serializeScheduleValue(partial.schedule));
			sets.push("next_run_at = ?");
			params.push(nextRunIso);
		}

		if (sets.length === 0) {
			return job;
		}

		sets.push("updated_at = datetime('now')");
		params.push(id);

		this.db.run(`UPDATE scheduled_jobs SET ${sets.join(", ")} WHERE id = ?`, params);
		this.armTimer();
		return this.getJob(id);
	}

	/**
	 * Defensive read: one corrupt row (a future kind, a truncated write) must
	 * not brick the whole list. Bad rows are logged and skipped. See M8.
	 */
	listJobs(): ScheduledJob[] {
		const rows = this.db.query("SELECT * FROM scheduled_jobs ORDER BY created_at DESC").all() as JobRow[];
		const jobs: ScheduledJob[] = [];
		for (const row of rows) {
			try {
				jobs.push(rowToJob(row));
			} catch (err: unknown) {
				const msg = err instanceof Error ? err.message : String(err);
				console.error(`[scheduler] Failed to parse row ${row.id} (${row.name ?? "?"}): ${msg}`);
			}
		}
		return jobs;
	}

	getJob(id: string): ScheduledJob | null {
		const row = this.db.query("SELECT * FROM scheduled_jobs WHERE id = ?").get(id) as JobRow | null;
		if (!row) return null;
		try {
			return rowToJob(row);
		} catch (err: unknown) {
			const msg = err instanceof Error ? err.message : String(err);
			console.error(`[scheduler] Failed to parse row ${row.id}: ${msg}`);
			return null;
		}
	}

	findJobIdByName(name: string | undefined): string | undefined {
		if (!name) return undefined;
		const lowerName = name.toLowerCase();
		for (const job of this.listJobs()) {
			if (job.name.toLowerCase() === lowerName) return job.id;
		}
		return undefined;
	}

	/**
	 * Manual trigger. Respects the single-slot onTimer guard (M2) and the
	 * job status gate (M9). An admin override cannot resurrect a failed job.
	 */
	async runJobNow(id: string): Promise<string> {
		if (this.executing) {
			throw new Error("scheduler is currently executing another job, try again shortly");
		}
		const job = this.getJob(id);
		if (!job) throw new Error(`Job not found: ${id}`);
		if (!job.enabled) throw new Error(`Job is disabled: ${id}`);
		if (job.status !== "active") {
			throw new Error(`Job ${id} is in status '${job.status}', only active jobs can be run`);
		}

		this.executing = true;
		try {
			return await this.runExecutor(job);
		} finally {
			this.executing = false;
		}
	}

	/** Minimal health snapshot for the /health endpoint (M5). */
	getHealthSummary(): SchedulerHealthSummary {
		return computeHealthSummary(this.db);
	}

	armTimer(): void {
		if (!this.running) return;

		if (this.timer) {
			clearTimeout(this.timer);
			this.timer = null;
		}

		const row = this.db
			.query(
				"SELECT MIN(next_run_at) as next FROM scheduled_jobs WHERE enabled = 1 AND status = 'active' AND next_run_at IS NOT NULL",
			)
			.get() as { next: string | null } | null;

		if (!row?.next) return;

		// Clamp the setTimeout delay to MAX_TIMER_MS (1 hour) to avoid the
		// 32-bit overflow described on the constant: any value above ~24.8 days
		// would be coerced to roughly 1 ms and hot-loop armTimer. When the next
		// fire is within the clamp we wake at the exact fire time; otherwise
		// we wake at the clamp, re-evaluate the MIN query, and re-arm.
		const delay = Math.max(0, new Date(row.next).getTime() - Date.now());
		const clamped = Math.min(delay, MAX_TIMER_MS);
		this.timer = setTimeout(() => this.onTimer(), clamped);
	}

	private async onTimer(): Promise<void> {
		if (!this.running) return;

		if (this.executing) {
			this.armTimer();
			return;
		}

		this.executing = true;

		try {
			const now = new Date().toISOString();
			const dueRows = this.db
				.query(
					"SELECT * FROM scheduled_jobs WHERE enabled = 1 AND status = 'active' AND next_run_at <= ? ORDER BY next_run_at ASC",
				)
				.all(now) as JobRow[];

			for (const row of dueRows) {
				if (!this.running) break;
				let job: ScheduledJob;
				try {
					job = rowToJob(row);
				} catch (err: unknown) {
					const msg = err instanceof Error ? err.message : String(err);
					console.error(`[scheduler] Skipping unparsable row ${row.id}: ${msg}`);
					continue;
				}
				try {
					const result = await this.runExecutor(job);
					const status = result.startsWith("Error:") ? "error" : "completed";
					this.fireJobCompleteCallbacks(job.name, status);
				} catch (err: unknown) {
					const msg = err instanceof Error ? err.message : String(err);
					console.error(`[scheduler] Job ${job.id} (${job.name}) failed: ${msg}`);
					this.fireJobCompleteCallbacks(job.name, "error");
				}
			}
		} finally {
			this.executing = false;
			this.armTimer();
		}
	}

	private fireJobCompleteCallbacks(name: string, status: string): void {
		for (const cb of this.jobCompleteCallbacks) {
			try {
				cb(name, status);
			} catch {}
		}
	}

	private runExecutor(job: ScheduledJob): Promise<string> {
		return executeJob(job, {
			db: this.db,
			runtime: this.runtime,
			slackChannel: this.slackChannel,
			ownerUserId: this.ownerUserId,
			notifyOwner: (text: string) => this.notifyOwner(text),
		});
	}

	private notifyOwner(text: string): void {
		if (this.slackChannel && this.ownerUserId) {
			this.slackChannel.sendDm(this.ownerUserId, text).catch((err: unknown) => {
				const msg = err instanceof Error ? err.message : String(err);
				console.error(`[scheduler] Failed to notify owner: ${msg}`);
			});
			return;
		}
		console.error(`[scheduler] Terminal failure notify dropped (owner unset): ${text}`);
	}
}