use std::process::Stdio;

use chrono::Local;
use croner::Cron;
use tokio::io::AsyncReadExt;
use tokio::process::Command;

use crate::config::{self, Config};
use crate::state::{JobRuntime, JobState, SchedulerState};

pub async fn load_jobs(config: &Config, state: &SchedulerState) {
	let mut jobs = state.write().await;
	jobs.clear();
	for job_config in &config.jobs {
		let mut runtime = JobRuntime::new(job_config.clone());
		if let Some(next) = next_run(&job_config.cron) {
			runtime.next_run = Some(next);
		}
		jobs.insert(job_config.name.clone(), runtime);
	}
}

pub async fn reload_jobs(config: &Config, state: &SchedulerState) {
	let mut jobs = state.write().await;

	let existing_names: Vec<String> = jobs.keys().cloned().collect();
	let new_names: Vec<String> = config.jobs.iter().map(|j| j.name.clone()).collect();

	for name in &existing_names {
		if !new_names.contains(name) {
			jobs.remove(name);
		}
	}

	for job_config in &config.jobs {
		if let Some(existing) = jobs.get_mut(&job_config.name) {
			existing.config = job_config.clone();
		} else {
			let mut runtime = JobRuntime::new(job_config.clone());
			if let Some(next) = next_run(&job_config.cron) {
				runtime.next_run = Some(next);
			}
			jobs.insert(job_config.name.clone(), runtime);
		}
	}
}

pub async fn run_scheduler(state: SchedulerState) {
	loop {
		tokio::time::sleep(std::time::Duration::from_secs(1)).await;

		let now = Local::now();
		let mut to_run = Vec::new();

		{
			let jobs = state.read().await;
			for (name, runtime) in jobs.iter() {
				if runtime.state == JobState::Paused || runtime.state == JobState::Stopped {
					continue;
				}

				if runtime.state == JobState::Running && !runtime.config.overlap {
					continue;
				}

				if let Some(backoff_until) = runtime.backoff_until {
					if now < backoff_until {
						continue;
					}
				}

				if let Some(next) = runtime.next_run {
					if now >= next {
						to_run.push(name.clone());
					}
				}
			}
		}

		for name in to_run {
			let state = state.clone();
			tokio::spawn(async move {
				execute_job(&name, &state).await;
			});
		}
	}
}

async fn execute_job(name: &str, state: &SchedulerState) {
	let (command, dir, timeout) = {
		let mut jobs = state.write().await;
		let Some(runtime) = jobs.get_mut(name) else {
			return;
		};
		runtime.state = JobState::Running;
		let cmd = runtime.config.command.clone();
		let dir = runtime.config.dir.clone();
		let timeout = runtime.config.timeout.as_ref().and_then(|t| config::parse_duration(t));
		(cmd, dir, timeout)
	};

	tracing::info!(job = %name, command = %command, "starting");

	let mut cmd = Command::new("sh");
	cmd.args(["-c", &command])
		.stdout(Stdio::piped())
		.stderr(Stdio::piped());

	if let Some(ref dir) = dir {
		cmd.current_dir(dir);
	}

	let result = match cmd.spawn() {
		Ok(mut child) => {
			if let Some(timeout_dur) = timeout {
				let stdout_handle = child.stdout.take();
				let stderr_handle = child.stderr.take();
				match tokio::time::timeout(timeout_dur, child.wait()).await {
					Ok(Ok(status)) => {
						let mut out = Vec::new();
						let mut err = Vec::new();
						if let Some(mut s) = stdout_handle {
							let _ = s.read_to_end(&mut out).await;
						}
						if let Some(mut s) = stderr_handle {
							let _ = s.read_to_end(&mut err).await;
						}
						Some(std::process::Output { status, stdout: out, stderr: err })
					}
					Ok(Err(e)) => {
						tracing::error!(job = %name, error = %e, "process error");
						None
					}
					Err(_) => {
						tracing::warn!(job = %name, "timed out, killing");
						let _ = child.kill().await;
						None
					}
				}
			} else {
				match child.wait_with_output().await {
					Ok(output) => Some(output),
					Err(e) => {
						tracing::error!(job = %name, error = %e, "process error");
						None
					}
				}
			}
		}
		Err(e) => {
			tracing::error!(job = %name, error = %e, "failed to spawn");
			None
		}
	};

	let mut jobs = state.write().await;
	let Some(runtime) = jobs.get_mut(name) else {
		return;
	};

	match result {
		Some(output) => {
			let code = output.status.code().unwrap_or(-1);

			if !output.stdout.is_empty() {
				let stdout = String::from_utf8_lossy(&output.stdout);
				for line in stdout.lines() {
					tracing::info!(job = %name, channel = "stdout", "{}", line);
				}
			}
			if !output.stderr.is_empty() {
				let stderr = String::from_utf8_lossy(&output.stderr);
				for line in stderr.lines() {
					tracing::warn!(job = %name, channel = "stderr", "{}", line);
				}
			}

			if code == 0 {
				tracing::info!(job = %name, "succeeded");
				runtime.record_success();
			} else {
				tracing::warn!(job = %name, exit_code = code, "failed");
				runtime.record_failure(code);
			}
		}
		None => {
			runtime.record_failure(-1);
		}
	}

	if runtime.state != JobState::Stopped && runtime.state != JobState::Paused {
		if let Some(next) = next_run(&runtime.config.cron) {
			runtime.next_run = Some(next);
		}
	}
}

pub fn next_run_for(cron_expr: &str) -> Option<chrono::DateTime<Local>> {
	let cron = Cron::new(cron_expr).parse().ok()?;
	let now = Local::now();
	cron.find_next_occurrence(&now, false).ok()
}

fn next_run(cron_expr: &str) -> Option<chrono::DateTime<Local>> {
	next_run_for(cron_expr)
}

pub async fn execute_job_by_name(name: &str, state: &SchedulerState) {
	execute_job(name, state).await;
}

#[cfg(test)]
mod tests {
	use super::*;

	#[test]
	fn next_run_for_valid_cron() {
		let result = next_run_for("* * * * *");
		assert!(result.is_some());
		let next = result.unwrap();
		assert!(next > Local::now());
	}

	#[test]
	fn next_run_for_specific_schedule() {
		let result = next_run_for("0 0 * * *"); // midnight daily
		assert!(result.is_some());
		let next = result.unwrap();
		assert_eq!(next.format("%H:%M").to_string(), "00:00");
	}

	#[test]
	fn next_run_for_invalid_cron() {
		assert!(next_run_for("not valid").is_none());
		assert!(next_run_for("").is_none());
		assert!(next_run_for("* * *").is_none());
	}

	#[test]
	fn next_run_for_is_in_future() {
		let now = Local::now();
		let next = next_run_for("*/5 * * * *").unwrap();
		assert!(next > now);
	}

	#[tokio::test]
	async fn load_jobs_populates_state() {
		let config = Config {
			jobs: vec![
				crate::config::JobConfig {
					name: "a".to_string(),
					cron: "* * * * *".to_string(),
					command: "echo a".to_string(),
					dir: None,
					on_error: crate::config::ErrorPolicy::Continue,
					overlap: false,
					timeout: None,
				},
				crate::config::JobConfig {
					name: "b".to_string(),
					cron: "0 * * * *".to_string(),
					command: "echo b".to_string(),
					dir: None,
					on_error: crate::config::ErrorPolicy::Continue,
					overlap: false,
					timeout: None,
				},
			],
		};

		let state = crate::state::new_state();
		load_jobs(&config, &state).await;

		let jobs = state.read().await;
		assert_eq!(jobs.len(), 2);
		assert!(jobs.contains_key("a"));
		assert!(jobs.contains_key("b"));
		assert!(jobs["a"].next_run.is_some());
		assert!(jobs["b"].next_run.is_some());
	}

	#[tokio::test]
	async fn reload_adds_and_removes_jobs() {
		let config1 = Config {
			jobs: vec![
				crate::config::JobConfig {
					name: "keep".to_string(),
					cron: "* * * * *".to_string(),
					command: "echo keep".to_string(),
					dir: None,
					on_error: crate::config::ErrorPolicy::Continue,
					overlap: false,
					timeout: None,
				},
				crate::config::JobConfig {
					name: "remove".to_string(),
					cron: "* * * * *".to_string(),
					command: "echo remove".to_string(),
					dir: None,
					on_error: crate::config::ErrorPolicy::Continue,
					overlap: false,
					timeout: None,
				},
			],
		};

		let state = crate::state::new_state();
		load_jobs(&config1, &state).await;

		let config2 = Config {
			jobs: vec![
				crate::config::JobConfig {
					name: "keep".to_string(),
					cron: "* * * * *".to_string(),
					command: "echo keep-updated".to_string(),
					dir: None,
					on_error: crate::config::ErrorPolicy::Continue,
					overlap: false,
					timeout: None,
				},
				crate::config::JobConfig {
					name: "new".to_string(),
					cron: "0 * * * *".to_string(),
					command: "echo new".to_string(),
					dir: None,
					on_error: crate::config::ErrorPolicy::Continue,
					overlap: false,
					timeout: None,
				},
			],
		};

		reload_jobs(&config2, &state).await;

		let jobs = state.read().await;
		assert_eq!(jobs.len(), 2);
		assert!(jobs.contains_key("keep"));
		assert!(jobs.contains_key("new"));
		assert!(!jobs.contains_key("remove"));
		assert_eq!(jobs["keep"].config.command, "echo keep-updated");
	}
}
