//! Periodic desired-state reconciliation. See DESIGN.md §8.

use std::error::Error;
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use arr_db::Db;
use tokio::sync::watch;
use tokio::time::{Instant, MissedTickBehavior};

pub const RECONCILE_INTERVAL: Duration = Duration::from_secs(30);
pub const RSS_INTERVAL: Duration = Duration::from_mins(10);
pub const METADATA_INTERVAL: Duration = Duration::from_hours(24);
pub const REAPER_INTERVAL: Duration = Duration::from_mins(5);

const RECONCILE_START_DELAY: Duration = Duration::ZERO;
const METADATA_START_DELAY: Duration = Duration::from_secs(5);
const REAPER_START_DELAY: Duration = Duration::from_secs(10);
const RSS_START_DELAY: Duration = Duration::from_secs(20);

const RECONCILE_ACTION_TIMEOUT: Duration = Duration::from_secs(25);
const REAPER_ACTION_TIMEOUT: Duration = Duration::from_mins(1);
const RSS_ACTION_TIMEOUT: Duration = Duration::from_mins(5);
const METADATA_ACTION_TIMEOUT: Duration = Duration::from_mins(30);

pub type ActionError = Box<dyn Error + Send + Sync>;
pub type ActionFuture<'a> =
    Pin<Box<dyn Future<Output = Result<Vec<Outcome>, ActionError>> + Send + 'a>>;

/// One independently scheduled class of reconciliation work.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Tick {
    Reconcile,
    Rss,
    Metadata,
    Reaper,
}

impl fmt::Display for Tick {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::Reconcile => "reconcile",
            Self::Rss => "rss",
            Self::Metadata => "metadata",
            Self::Reaper => "reaper",
        })
    }
}

/// A desired/actual gap closed by an action during a tick.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Outcome {
    gap: String,
    action: String,
}

impl Outcome {
    #[must_use]
    pub fn new(gap: impl Into<String>, action: impl Into<String>) -> Self {
        Self {
            gap: gap.into(),
            action: action.into(),
        }
    }
}

/// A concrete reconciler registered for one tick class.
///
/// Implementations query persistent domain rows on every call. They may keep
/// in-flight state in memory, but the first call must rebuild it from the
/// external system so restarting the process converges without a job table.
pub trait Action: fmt::Debug + Send + Sync {
    fn name(&self) -> &'static str;

    fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a>;
}

#[derive(Debug)]
struct RegisteredAction {
    tick: Tick,
    action: Arc<dyn Action>,
}

#[derive(Debug, Clone, Copy)]
struct Schedule {
    reconcile: Duration,
    rss: Duration,
    metadata: Duration,
    reaper: Duration,
}

impl Default for Schedule {
    fn default() -> Self {
        Self {
            reconcile: RECONCILE_INTERVAL,
            rss: RSS_INTERVAL,
            metadata: METADATA_INTERVAL,
            reaper: REAPER_INTERVAL,
        }
    }
}

impl Schedule {
    const fn interval(self, tick: Tick) -> Duration {
        match tick {
            Tick::Reconcile => self.reconcile,
            Tick::Rss => self.rss,
            Tick::Metadata => self.metadata,
            Tick::Reaper => self.reaper,
        }
    }

    const fn start_delay(tick: Tick) -> Duration {
        match tick {
            Tick::Reconcile => RECONCILE_START_DELAY,
            Tick::Rss => RSS_START_DELAY,
            Tick::Metadata => METADATA_START_DELAY,
            Tick::Reaper => REAPER_START_DELAY,
        }
    }

    const fn action_timeout(tick: Tick) -> Duration {
        match tick {
            Tick::Reconcile => RECONCILE_ACTION_TIMEOUT,
            Tick::Rss => RSS_ACTION_TIMEOUT,
            Tick::Metadata => METADATA_ACTION_TIMEOUT,
            Tick::Reaper => REAPER_ACTION_TIMEOUT,
        }
    }
}

/// The daemon's four-lane scheduler. Actions register into a lane; no queued
/// work is persisted because each action rediscovers gaps from domain rows.
#[derive(Debug)]
pub struct ReconcileLoop {
    database: Db,
    schedule: Schedule,
    action_timeout_override: Option<Duration>,
    actions: Vec<RegisteredAction>,
}

impl ReconcileLoop {
    #[must_use]
    pub fn new(database: Db) -> Self {
        Self {
            database,
            schedule: Schedule::default(),
            action_timeout_override: None,
            actions: Vec::new(),
        }
    }

    /// Register an action in the cadence that owns it.
    #[must_use]
    pub fn register<A>(mut self, tick: Tick, action: A) -> Self
    where
        A: Action + 'static,
    {
        self.actions.push(RegisteredAction {
            tick,
            action: Arc::new(action),
        });
        self
    }

    /// Run until shutdown. Reconciliation runs immediately on startup so
    /// actions can reconstruct transient state before serving later ticks.
    pub async fn run(self, shutdown: watch::Receiver<bool>) {
        let this = Arc::new(self);
        let reconcile = Arc::clone(&this).run_lane(Tick::Reconcile, shutdown.clone());
        let rss = Arc::clone(&this).run_lane(Tick::Rss, shutdown.clone());
        let metadata = Arc::clone(&this).run_lane(Tick::Metadata, shutdown.clone());
        let reaper = Arc::clone(&this).run_lane(Tick::Reaper, shutdown);

        tokio::join!(reconcile, rss, metadata, reaper);
    }

    async fn run_lane(self: Arc<Self>, tick: Tick, mut shutdown: watch::Receiver<bool>) {
        if *shutdown.borrow() {
            return;
        }

        let interval = self.schedule.interval(tick);
        let first_tick = Instant::now() + Schedule::start_delay(tick);
        let mut timer = tokio::time::interval_at(first_tick, interval);
        timer.set_missed_tick_behavior(MissedTickBehavior::Skip);

        loop {
            tokio::select! {
                biased;
                changed = shutdown.changed() => {
                    if changed.is_err() || *shutdown.borrow() {
                        return;
                    }
                }
                _ = timer.tick() => {
                    if *shutdown.borrow() {
                        return;
                    }
                    tokio::select! {
                        biased;
                        changed = shutdown.changed() => {
                            if changed.is_err() || *shutdown.borrow() {
                                return;
                            }
                        }
                        _ = self.run_tick(tick) => {}
                    }
                }
            }
        }
    }

    async fn run_tick(&self, tick: Tick) -> TickReport {
        let mut report = TickReport::default();
        let action_timeout = self
            .action_timeout_override
            .unwrap_or_else(|| Schedule::action_timeout(tick));

        for registered in self.actions.iter().filter(|action| action.tick == tick) {
            match tokio::time::timeout(action_timeout, registered.action.run(&self.database)).await
            {
                Ok(Ok(outcomes)) => {
                    for outcome in outcomes {
                        tracing::info!(
                            tick = %tick,
                            worker = registered.action.name(),
                            gap = %outcome.gap,
                            action = %outcome.action,
                            "reconcile action"
                        );
                        report.actions_taken += 1;
                    }
                }
                Ok(Err(error)) => {
                    report.failures += 1;
                    tracing::error!(
                        tick = %tick,
                        worker = registered.action.name(),
                        error = %error,
                        "reconcile action failed"
                    );
                }
                Err(_) => {
                    report.failures += 1;
                    tracing::error!(
                        tick = %tick,
                        worker = registered.action.name(),
                        timeout_seconds = action_timeout.as_secs(),
                        "reconcile action timed out"
                    );
                }
            }
        }

        if report.actions_taken == 0 && report.failures == 0 {
            tracing::debug!(tick = %tick, "reconcile tick idle");
        } else {
            tracing::info!(
                tick = %tick,
                actions_taken = report.actions_taken,
                failures = report.failures,
                "reconcile tick"
            );
        }
        report
    }
}

#[derive(Debug, Default, PartialEq, Eq)]
struct TickReport {
    actions_taken: usize,
    failures: usize,
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use std::collections::HashMap;
    use std::io;

    use tokio::sync::{Mutex, Notify};

    use super::*;

    #[derive(Clone, Debug, Default)]
    struct FakeTransmission {
        torrents: Arc<Mutex<HashMap<i64, String>>>,
    }

    impl FakeTransmission {
        async fn find_or_add(&self, movie_id: i64) -> (String, bool) {
            let mut torrents = self.torrents.lock().await;
            if let Some(hash) = torrents.get(&movie_id) {
                return (hash.clone(), false);
            }

            let hash = format!("fake-{movie_id}");
            torrents.insert(movie_id, hash.clone());
            (hash, true)
        }

        async fn len(&self) -> usize {
            self.torrents.lock().await.len()
        }
    }

    #[derive(Debug)]
    struct FakeGrabAction {
        transmission: FakeTransmission,
        release_id: i64,
        fail_after_add: bool,
    }

    #[derive(Clone, Debug, Default)]
    struct StalledAction {
        started: Arc<Notify>,
    }

    impl Action for StalledAction {
        fn name(&self) -> &'static str {
            "stalled"
        }

        fn run<'a>(&'a self, _database: &'a Db) -> ActionFuture<'a> {
            Box::pin(async move {
                self.started.notify_one();
                std::future::pending().await
            })
        }
    }

    impl Action for FakeGrabAction {
        fn name(&self) -> &'static str {
            "fake-grab"
        }

        fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> {
            Box::pin(async move {
                let movie_id = sqlx::query_scalar::<_, i64>(
                    "SELECT id FROM movies m
                     WHERE wanted = 1 AND blocked = 0 AND state = 'missing'
                       AND NOT EXISTS (
                           SELECT 1 FROM grabs g
                           WHERE g.target_kind = 'movie' AND g.target_id = m.id
                             AND g.state IN ('sent', 'downloaded', 'imported')
                       )
                     ORDER BY id LIMIT 1",
                )
                .fetch_optional(database.pool())
                .await?;
                let Some(movie_id) = movie_id else {
                    return Ok(Vec::new());
                };

                let (infohash, added) = self.transmission.find_or_add(movie_id).await;
                if added && self.fail_after_add {
                    return Err(Box::new(io::Error::new(
                        io::ErrorKind::Interrupted,
                        "simulated process death after Transmission accepted the torrent",
                    )) as ActionError);
                }

                sqlx::query(
                    "INSERT INTO grabs (release_id, target_kind, target_id, infohash)
                     VALUES (?, 'movie', ?, ?)",
                )
                .bind(self.release_id)
                .bind(movie_id)
                .bind(infohash)
                .execute(database.pool())
                .await?;

                Ok(vec![Outcome::new(
                    format!("wanted movie {movie_id} has no active grab"),
                    if added {
                        "added torrent and recorded grab"
                    } else {
                        "recovered existing torrent and recorded grab"
                    },
                )])
            })
        }
    }

    async fn seeded_database() -> (tempfile::TempDir, Db, i64) {
        let directory = tempfile::tempdir().unwrap();
        let database = Db::connect(directory.path().join("arr.db")).await.unwrap();
        database.migrate().await.unwrap();

        let root_id: i64 =
            sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'movie' AND audience = 'main'")
                .fetch_one(database.pool())
                .await
                .unwrap();
        sqlx::query("INSERT INTO movies (tmdb_id, title, root_id) VALUES (1, 'Movie', ?)")
            .bind(root_id)
            .execute(database.pool())
            .await
            .unwrap();
        let release_id = sqlx::query(
            "INSERT INTO releases
             (indexer_id, guid, name, size, download_url, parsed, verdict)
             VALUES (1, 'guid', 'Movie.1080p', 1, 'magnet:test', '{}', 'eligible')",
        )
        .execute(database.pool())
        .await
        .unwrap()
        .last_insert_rowid();

        (directory, database, release_id)
    }

    #[test]
    fn default_schedule_matches_the_design() {
        let schedule = Schedule::default();
        assert_eq!(schedule.interval(Tick::Reconcile), Duration::from_secs(30));
        assert_eq!(schedule.interval(Tick::Rss), Duration::from_mins(10));
        assert_eq!(schedule.interval(Tick::Metadata), Duration::from_hours(24));
        assert_eq!(schedule.interval(Tick::Reaper), Duration::from_mins(5));

        assert_eq!(Schedule::start_delay(Tick::Reconcile), Duration::ZERO);
        assert_eq!(
            Schedule::start_delay(Tick::Metadata),
            Duration::from_secs(5)
        );
        assert_eq!(Schedule::start_delay(Tick::Reaper), Duration::from_secs(10));
        assert_eq!(Schedule::start_delay(Tick::Rss), Duration::from_secs(20));

        assert_eq!(
            Schedule::action_timeout(Tick::Reconcile),
            Duration::from_secs(25)
        );
        assert_eq!(
            Schedule::action_timeout(Tick::Reaper),
            Duration::from_mins(1)
        );
        assert_eq!(Schedule::action_timeout(Tick::Rss), Duration::from_mins(5));
        assert_eq!(
            Schedule::action_timeout(Tick::Metadata),
            Duration::from_mins(30)
        );
    }

    #[tokio::test]
    async fn restart_converges_without_duplicate_external_work() {
        let (_directory, database, release_id) = seeded_database().await;
        let transmission = FakeTransmission::default();

        let interrupted = ReconcileLoop::new(database.clone()).register(
            Tick::Reconcile,
            FakeGrabAction {
                transmission: transmission.clone(),
                release_id,
                fail_after_add: true,
            },
        );
        let first = interrupted.run_tick(Tick::Reconcile).await;
        assert_eq!(first.failures, 1);
        assert_eq!(transmission.len().await, 1);
        drop(interrupted);

        let restarted = ReconcileLoop::new(database.clone()).register(
            Tick::Reconcile,
            FakeGrabAction {
                transmission: transmission.clone(),
                release_id,
                fail_after_add: false,
            },
        );
        let recovered = restarted.run_tick(Tick::Reconcile).await;
        let stable = restarted.run_tick(Tick::Reconcile).await;

        assert_eq!(recovered.actions_taken, 1);
        assert_eq!(stable.actions_taken, 0);
        assert_eq!(transmission.len().await, 1);
        let grabs: i64 = sqlx::query_scalar("SELECT count(*) FROM grabs")
            .fetch_one(database.pool())
            .await
            .unwrap();
        assert_eq!(grabs, 1);
    }

    #[tokio::test]
    async fn a_stalled_action_times_out() {
        let (_directory, database, _release_id) = seeded_database().await;
        let mut reconcile =
            ReconcileLoop::new(database).register(Tick::Reconcile, StalledAction::default());
        reconcile.action_timeout_override = Some(Duration::from_millis(1));

        let report = reconcile.run_tick(Tick::Reconcile).await;

        assert_eq!(report.failures, 1);
        assert_eq!(report.actions_taken, 0);
    }

    #[tokio::test]
    async fn shutdown_cancels_an_in_flight_action() {
        let (_directory, database, _release_id) = seeded_database().await;
        let action = StalledAction::default();
        let started = Arc::clone(&action.started);
        let reconcile = ReconcileLoop::new(database).register(Tick::Reconcile, action);
        let (shutdown_tx, shutdown_rx) = watch::channel(false);
        let task = tokio::spawn(reconcile.run(shutdown_rx));
        started.notified().await;

        shutdown_tx.send(true).unwrap();

        tokio::time::timeout(Duration::from_millis(100), task)
            .await
            .expect("reconcile loop stopped")
            .expect("reconcile task did not panic");
    }
}