//! The worker: one thread that sends what the ledger says is waiting, oldest first (M5c spec,
//! section 5). A turn is read again from `loopd`'s log, neutralised, cut into parts Honcho takes,
//! and added to its Honcho session as one batch; the ids Honcho gives go in the ledger. An outage
//! is waited out with a doubling wait and logged when it starts and when it ends; a refused send is
//! marked failed; a refused key stops that workspace until `memoryd` restarts. When nothing waits,
//! the worker decides what `brokerd` could not answer for before, then sleeps until a turn ends.

use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use bxcore::{SessionId, SessionStateReply, Timestamp};

use crate::honcho::client::{Clients, HonchoError, classify};
use crate::honcho::edits;
use crate::honcho::ledger::{self, Action, Item, Ledger, Record, Workspace};
use crate::honcho::naming::{HonchoSession, honcho_session};
use crate::honcho::status::{self, Shared};
use crate::honcho::turns::{TurnSync, log_path};
use crate::serve::Log;

use crate::honcho::messages::ids;
pub use crate::honcho::messages::{MAX_BATCH, MAX_PART, batch, parts, read_turn};
pub use crate::honcho::start::{Opened, open, spawn};
pub use crate::honcho::wake::{Tuning, Wake};

pub const UNREACHABLE: &str = "see docs/runbook.md#honcho-unreachable";
pub const SEND_REFUSED: &str = "see docs/runbook.md#honcho-send-refused";
pub const KEY_REFUSED: &str = "see docs/runbook.md#honcho-key-refused";
pub const LOG_DAMAGED: &str = "see docs/runbook.md#session-log-damaged";

/// What one step of the worker did, and so what the loop does next.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Step {
    /// A turn was sent, or marked failed: go on at once.
    Done,
    /// Honcho is unreachable: wait this long, then try the same turn again.
    Backoff(Duration),
    /// Nothing to send, and the retry decided nothing new: sleep.
    Idle,
    /// The ledger cannot be written: the worker stops.
    Stopped,
}

/// `brokerd`'s answer for a session, for the retry.
pub type AskBox = Box<dyn Fn(&SessionId) -> Result<SessionStateReply, String> + Send>;

/// Why a turn could not be sent.
enum Unsent {
    Honcho(HonchoError),
    /// The turn cannot be read from the log any more, or is too long for one batch.
    Local(&'static str),
    /// The turn's log could not be read now, and why: the log's path and the error.
    Unreadable(String),
}

/// What to send for one waiting item (M5c spec, section 5).
enum Job {
    /// A turn: its session and number, its speaker, and its Honcho session.
    Turn {
        session: SessionId,
        turn: u64,
        speaker: String,
        honcho: Option<HonchoSession>,
    },
    /// A conclusion about `peer`, stamped with `session`'s memory session.
    Conclude {
        peer: String,
        session: SessionId,
        text: String,
    },
    /// Delete the conclusion with this id; `None` when no sent conclusion is live.
    Unconclude { id: Option<String> },
    /// An untrusted note: a fleet message from the agent.
    Fleet {
        session: SessionId,
        edit: String,
        text: String,
    },
}

/// The oldest waiting item: what it is, where it goes, and what to send.
struct Next {
    item: Item,
    workspace: Workspace,
    job: Job,
}

pub struct Worker {
    sync: Arc<Mutex<TurnSync>>,
    /// The ledger's own lock (task 12): taken alone, briefly, never with `sync`'s.
    ledger: Arc<Mutex<Ledger>>,
    sessions: PathBuf,
    clients: Clients,
    wake: Arc<Wake>,
    ask: AskBox,
    log: Log,
    tuning: Tuning,
    agent: String,
    /// In an outage: the next wait.
    outage: Option<Duration>,
    /// While a session log cannot be read: the next wait.
    unreadable: Option<Duration>,
    /// Workspaces whose key was refused: nothing more is sent to them until restart.
    refused: HashSet<Workspace>,
    /// What `bxctl memory sync` shows; its lock is taken alone.
    status: Shared,
    stopped: bool,
}

impl Worker {
    /// `agent` is `[honcho] agent`. `sync` and `wake` are shared with the socket threads.
    pub fn new(
        sync: Arc<Mutex<TurnSync>>,
        clients: Clients,
        wake: Arc<Wake>,
        ask: AskBox,
        log: Log,
        tuning: Tuning,
        agent: &str,
    ) -> Worker {
        let (ledger, sessions, status) = {
            let sync = sync.lock().unwrap_or_else(|p| p.into_inner());
            (
                sync.shared_ledger(),
                sync.sessions().to_path_buf(),
                sync.shared_status(),
            )
        };
        Worker {
            sync,
            ledger,
            sessions,
            clients,
            wake,
            ask,
            log,
            tuning,
            agent: agent.to_string(),
            outage: None,
            unreadable: None,
            refused: HashSet::new(),
            status,
            stopped: false,
        }
    }

    /// Step until the ledger cannot be written; wait as each step says.
    pub fn run(mut self) {
        loop {
            match self.step() {
                Step::Done => {}
                Step::Backoff(wait) => std::thread::sleep(wait),
                Step::Idle => self.wake.wait(self.tuning.idle),
                Step::Stopped => return,
            }
        }
    }

    /// Send the oldest waiting turn, or, with none, retry the undecided ones.
    pub fn step(&mut self) -> Step {
        if self.stopped {
            return Step::Stopped;
        }
        let Some(next) = self.next() else {
            self.retry();
            return match self.next() {
                Some(_) => Step::Done,
                None => Step::Idle,
            };
        };
        match self.send(&next) {
            Ok(sent) => {
                self.recovered();
                self.record(Record::Sent {
                    time: Timestamp::now(),
                    item: next.item,
                    ids: sent,
                })
            }
            Err(Unsent::Honcho(e @ HonchoError::Unreachable { .. })) => self.unreachable(&e),
            Err(Unsent::Honcho(HonchoError::KeyRefused { status })) => {
                self.recovered();
                self.refused.insert(next.workspace);
                let name = self
                    .clients
                    .client(next.workspace)
                    .workspace_id()
                    .to_string();
                (self.log)(&format!(
                    "memoryd: Honcho refused the key of workspace {name} (HTTP {status}); nothing more is sent to it until memoryd restarts; {KEY_REFUSED}"
                ));
                status::update(&self.status, |s| {
                    s.refused.insert(next.workspace, (name, status));
                });
                Step::Done
            }
            Err(Unsent::Honcho(e)) => {
                self.recovered();
                (self.log)(&format!(
                    "memoryd: {} was not sent: {e}; it is marked failed; {SEND_REFUSED}",
                    ledger::label(&next.item)
                ));
                self.record(Record::Failed {
                    time: Timestamp::now(),
                    item: next.item,
                    status: e.status(),
                    kind: e.kind().to_string(),
                })
            }
            Err(Unsent::Unreadable(why)) => self.unreadable(&next.item, why),
            Err(Unsent::Local(kind)) => {
                let pointer = if kind == "log" {
                    LOG_DAMAGED
                } else {
                    SEND_REFUSED
                };
                (self.log)(&format!(
                    "memoryd: {} cannot be sent ({kind}); it is marked failed; {pointer}",
                    ledger::label(&next.item)
                ));
                self.record(Record::Failed {
                    time: Timestamp::now(),
                    item: next.item,
                    status: None,
                    kind: kind.to_string(),
                })
            }
        }
    }

    /// The oldest waiting item whose workspace's key was not refused. A retraction, which
    /// nothing decides any more, waits.
    fn next(&self) -> Option<Next> {
        let ledger = self.ledger.lock().unwrap_or_else(|p| p.into_inner());
        ledger.outbox().into_iter().find_map(|(item, action)| {
            let (workspace, job) = match (&item, action) {
                (Item::Turn { session, turn }, Action::Turn { workspace, speaker }) => (
                    workspace,
                    Job::Turn {
                        honcho: honcho_session(ledger.decisions(), session, *turn),
                        session: session.clone(),
                        turn: *turn,
                        speaker,
                    },
                ),
                (
                    Item::Edit { .. },
                    Action::Conclude {
                        workspace,
                        peer,
                        session,
                        text,
                    },
                ) => (
                    workspace,
                    Job::Conclude {
                        peer,
                        session,
                        text,
                    },
                ),
                (
                    Item::Edit { .. },
                    Action::Unconclude {
                        workspace,
                        peer,
                        text,
                    },
                ) => (
                    workspace,
                    Job::Unconclude {
                        id: ledger.conclusion(workspace, &peer, &text).map(|(_, id)| id),
                    },
                ),
                (Item::Edit { id, .. }, Action::Fleet { session, text }) => (
                    Workspace::Fleet,
                    Job::Fleet {
                        session,
                        edit: id.clone(),
                        text,
                    },
                ),
                _ => return None,
            };
            if self.refused.contains(&workspace) {
                return None;
            }
            Some(Next {
                item: item.clone(),
                workspace,
                job,
            })
        })
    }

    /// Send one item, by what it is.
    fn send(&mut self, next: &Next) -> Result<Vec<String>, Unsent> {
        let honcho = |e| Unsent::Honcho(e);
        match &next.job {
            Job::Turn {
                session,
                turn,
                speaker,
                honcho: name,
            } => self.send_turn(next.workspace, session, *turn, speaker, name.as_ref()),
            Job::Conclude {
                peer,
                session,
                text,
            } => edits::conclude(&mut self.clients, next.workspace, peer, session, text)
                .map_err(honcho),
            Job::Unconclude { id } => {
                edits::unconclude(&self.clients, next.workspace, id.as_deref()).map_err(honcho)
            }
            Job::Fleet {
                session,
                edit,
                text,
            } => {
                let messages = edits::fleet_batch(text, &self.agent, session, edit);
                if messages.len() > MAX_BATCH {
                    return Err(Unsent::Local("too-long"));
                }
                edits::fleet(&mut self.clients, session, &messages).map_err(honcho)
            }
        }
    }

    /// Read the turn again, make sure of its Honcho session, and add its messages as one batch.
    fn send_turn(
        &mut self,
        workspace: Workspace,
        session: &SessionId,
        turn: u64,
        speaker: &str,
        honcho: Option<&HonchoSession>,
    ) -> Result<Vec<String>, Unsent> {
        let path = log_path(&self.sessions, session);
        let read = read_turn(&path, turn).map_err(Unsent::Unreadable)?;
        self.readable(&path);
        let (text, reply) = read.ok_or(Unsent::Local("log"))?;
        let honcho = honcho.ok_or(Unsent::Local("log"))?;
        let messages = batch(&text, &reply, speaker, &self.agent, session, turn);
        if messages.len() > MAX_BATCH {
            return Err(Unsent::Local("too-long"));
        }
        self.clients
            .ensure_session(workspace, honcho)
            .map_err(Unsent::Honcho)?;
        let client = self.clients.client(workspace);
        match client.add_messages(&honcho.name, &messages) {
            Ok(made) => Ok(ids(&made)),
            Err(e) => Err(Unsent::Honcho(classify(&e))),
        }
    }

    /// Honcho did not answer: log the outage's start once, and wait longer each time.
    fn unreachable(&mut self, error: &HonchoError) -> Step {
        let wait = match self.outage {
            Some(wait) => wait,
            None => {
                (self.log)(
                    format!("memoryd: {error}; sends wait and are retried; {UNREACHABLE}").as_str(),
                );
                status::update(&self.status, |s| s.outage_since = Some(Timestamp::now()));
                self.tuning.backoff
            }
        };
        self.outage = Some(wait.saturating_mul(2).min(self.tuning.backoff_max));
        Step::Backoff(wait)
    }

    /// A session log could not be read now: log it once, with the item's label and the error,
    /// never the text, and wait as for an outage. The item stays waiting.
    fn unreadable(&mut self, item: &Item, why: String) -> Step {
        let wait = match self.unreadable {
            Some(wait) => wait,
            None => {
                (self.log)(&format!(
                    "memoryd: {} cannot be read now ({why}); sends wait and are retried; {LOG_DAMAGED}",
                    ledger::label(item)
                ));
                status::update(&self.status, |s| {
                    s.log_unreadable_since = Some(Timestamp::now())
                });
                self.tuning.backoff
            }
        };
        self.unreadable = Some(wait.saturating_mul(2).min(self.tuning.backoff_max));
        Step::Backoff(wait)
    }

    /// The log at `path` read: if one could not be read before, log that it reads again.
    fn readable(&mut self, path: &std::path::Path) {
        if self.unreadable.take().is_some() {
            (self.log)(&format!(
                "memoryd: {} reads again; waiting sends go on",
                path.display()
            ));
            status::update(&self.status, |s| s.log_unreadable_since = None);
        }
    }

    /// Honcho answered: if that ends an outage, log it once, and decide what waited meanwhile.
    fn recovered(&mut self) {
        if self.outage.take().is_some() {
            (self.log)("memoryd: Honcho answers again; waiting sends go on");
            status::update(&self.status, |s| s.outage_since = None);
            self.retry();
        }
    }

    /// Task 09's walk, for the turns `brokerd` could not answer for.
    fn retry(&mut self) {
        let mut sync = self.sync.lock().unwrap_or_else(|p| p.into_inner());
        sync.retry(&*self.ask, Timestamp::now(), &*self.log);
    }

    /// Append the worker's record. A ledger that cannot be written stops the worker: the send
    /// would be repeated at every step.
    fn record(&mut self, record: Record) -> Step {
        let result = self
            .ledger
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .append(record);
        match result {
            Ok(()) => Step::Done,
            Err(e) => {
                (self.log)(
                    format!(
                        "memoryd: {e}; the Honcho worker stops until memoryd restarts; {}",
                        ledger::RUNBOOK
                    )
                    .as_str(),
                );
                self.stopped = true;
                Step::Stopped
            }
        }
    }
}