//! 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::{Path, PathBuf};
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;

use bxcore::sanitize::neutralise;
use bxcore::{SessionId, SessionStateReply, Timestamp};
use roncho::MessageCreate;
use serde_json::{Map, json};

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

pub use crate::honcho::start::{Opened, open, spawn};

/// The most characters Honcho takes in one message (recorded: 25,000 is taken, 25,001 refused).
pub const MAX_PART: usize = 25_000;
/// The most messages the worker puts in one batch. A turn that needs more is marked failed.
pub const MAX_BATCH: usize = 100;

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";

/// The worker's waits. Tests shorten them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Tuning {
    /// The first wait in an outage; each next one doubles, up to `backoff_max`.
    pub backoff: Duration,
    pub backoff_max: Duration,
    /// How long the worker sleeps with nothing to send, unless a turn ends first.
    pub idle: Duration,
}

impl Tuning {
    pub const DEFAULT: Tuning = Tuning {
        backoff: Duration::from_secs(5),
        backoff_max: Duration::from_secs(600),
        idle: Duration::from_secs(30),
    };
}

/// Wakes the worker from its idle sleep when a turn ends.
#[derive(Debug, Default)]
pub struct Wake {
    woken: Mutex<bool>,
    ready: Condvar,
}

impl Wake {
    /// Wake the worker if it sleeps, or make its next sleep return at once.
    pub fn notify(&self) {
        let mut woken = self.woken.lock().unwrap_or_else(|p| p.into_inner());
        *woken = true;
        self.ready.notify_one();
    }

    /// Sleep until `notify`, or at most `timeout`.
    pub fn wait(&self, timeout: Duration) {
        let woken = self.woken.lock().unwrap_or_else(|p| p.into_inner());
        let (mut woken, _) = self
            .ready
            .wait_timeout_while(woken, timeout, |woken| !*woken)
            .unwrap_or_else(|p| p.into_inner());
        *woken = false;
    }
}

/// 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),
}

/// 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>,
    /// Workspaces whose key was refused: nothing more is sent to them until restart.
    refused: HashSet<Workspace>,
    stopped: bool,
}

/// `text` in parts of at most `MAX_PART` characters, in order. Empty text has no part.
pub fn parts(text: &str) -> Vec<String> {
    let chars: Vec<char> = text.chars().collect();
    chars
        .chunks(MAX_PART)
        .map(|chunk| chunk.iter().collect())
        .collect()
}

/// Turn `turn`'s user text and final reply, read again from the log at `path`. `None` when the log
/// does not read, or has no such turn, or the turn no reply.
pub fn read_turn(path: &Path, turn: u64) -> Option<(String, String)> {
    let records = read_log(path).ok()?;
    let records = turn_records(&records, turn)?;
    let (text, _) = user_text(records);
    let reply = final_reply(records)?;
    Some((text.to_string(), reply.to_string()))
}

/// A turn's messages, in one batch: the speaker's text, then the agent's reply, each neutralised
/// and in parts of at most `MAX_PART` characters, each part with the turn's metadata.
pub fn batch(
    text: &str,
    reply: &str,
    speaker: &str,
    agent: &str,
    session: &SessionId,
    turn: u64,
) -> Vec<MessageCreate> {
    let mut metadata = Map::new();
    metadata.insert("agent".to_string(), json!(agent));
    metadata.insert("boxmaker_session".to_string(), json!(session.as_str()));
    metadata.insert("turn".to_string(), json!(turn));
    let mut messages = Vec::new();
    for (content, peer) in [(text, speaker), (reply, agent)] {
        for part in parts(&neutralise(content)) {
            messages.push(MessageCreate::new(part, peer).with_metadata(metadata.clone()));
        }
    }
    messages
}

/// An item as the log lines name it: `turn 3 of session mm-a`, `part 0 of edit 2026-09-28-0001`.
fn label(item: &Item) -> String {
    match item {
        Item::Turn { session, turn } => format!("turn {turn} of session {}", session.as_str()),
        Item::Edit { id, part } => format!("part {part} of edit {id}"),
    }
}

/// The ids of the messages Honcho made, in order.
fn ids(messages: &[roncho::Message]) -> Vec<String> {
    messages.iter().map(|m| m.id.clone()).collect()
}

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) = {
            let sync = sync.lock().unwrap_or_else(|p| p.into_inner());
            (sync.shared_ledger(), sync.sessions().to_path_buf())
        };
        Worker {
            sync,
            ledger,
            sessions,
            clients,
            wake,
            ask,
            log,
            tuning,
            agent: agent.to_string(),
            outage: None,
            refused: HashSet::new(),
            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);
                (self.log)(&format!(
                    "memoryd: Honcho refused the key of workspace {} (HTTP {status}); nothing more is sent to it until memoryd restarts; {KEY_REFUSED}",
                    self.clients.client(next.workspace).workspace_id()
                ));
                Step::Done
            }
            Err(Unsent::Honcho(e)) => {
                self.recovered();
                (self.log)(&format!(
                    "memoryd: {} was not sent: {e}; it is marked failed; {SEND_REFUSED}",
                    label(&next.item)
                ));
                self.record(Record::Failed {
                    time: Timestamp::now(),
                    item: next.item,
                    status: e.status(),
                    kind: e.kind().to_string(),
                })
            }
            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}",
                    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 (text, reply) =
            read_turn(&log_path(&self.sessions, session), turn).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(),
                );
                self.tuning.backoff
            }
        };
        self.outage = Some(wait.saturating_mul(2).min(self.tuning.backoff_max));
        Step::Backoff(wait)
    }

    /// 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");
            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
            }
        }
    }
}