//! The ledger: the audit writer and every session's state behind one lock, with the three steps
//! that hold it. Without the one lock, two results for one session could each read `private` and
//! the second write would put the taint back down; after any failed append or panic while the lock was held, the ledger refuses every later step, and only the next start's check puts a partial line on disk right.

use std::sync::{Mutex, MutexGuard};

use proto::{
    ApprovalAnswer, AuditEvent, CallId, DataClass, DecisionRecord, DenyReason, GrantProblem, Mode,
    SessionId, Timestamp, ToolRequest, ToolResponse,
};

use crate::approvals::Verdict;
use crate::audit::{AuditError, Writer};
use crate::grants::GrantSet;
use crate::policy::{Ask, Denial, Label, SessionState};
use crate::state::StateStore;

/// The message `finish` returns when the raised taint and the record are not both on disk.
pub const NOT_RECORDED: &str = "the result could not be recorded";
/// After a panic while the lock was held: every call is denied until brokerd is restarted.
pub const POISONED: &str = "brokerd: a thread panicked while holding the ledger; every call is denied until brokerd is restarted\nsee docs/runbook.md#audit-unavailable";
/// After a failed append: every call is denied until brokerd is restarted.
pub const STOPPED: &str = "brokerd: an earlier audit write failed; every call is denied until brokerd is restarted\nsee docs/runbook.md#audit-unavailable";

/// What `grants::load` returns: a valid set, or the list of problems that made it invalid.
pub type Grants = Result<GrantSet, Vec<GrantProblem>>;

/// The audit writer behind one lock, so a step that fails to record refuses every later step.
pub trait AuditSink: Send {
    fn append(&mut self, time: Timestamp, event: AuditEvent) -> Result<u64, AuditError>;
}

impl AuditSink for Writer {
    fn append(&mut self, time: Timestamp, event: AuditEvent) -> Result<u64, AuditError> {
        Writer::append(self, time, event)
    }
}

/// A call the broker is finishing: the session, the call, the decision it was approved under, and
/// the label the result carries.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Call {
    pub session: SessionId,
    pub call: CallId,
    pub decision: u64,
    pub label: Label,
}

impl Call {
    /// The call and label of an approved decision, with the seq it was approved under.
    pub fn of(decision: &crate::policy::Decision, seq: u64) -> Call {
        let request = decision.request();
        Call {
            session: request.session.clone(),
            call: request.call,
            decision: seq,
            label: decision.label(),
        }
    }
}

/// How a call was decided.
#[derive(Debug)]
pub enum Decided {
    Allowed {
        decision: crate::policy::Decision,
        seq: u64,
    },
    Ask {
        ask: Ask,
        seq: u64,
        state: SessionState,
    },
    Denied(DenyReason),
}

/// The value a decided call carries past the record, or the denial reason when there is none.
enum RecordPayload {
    Allowed(crate::policy::Decision),
    Ask(Ask),
    Denial(Denial),
    Simple(DenyReason),
}

/// An owner's answer to a pending call.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Answer {
    Approved {
        by: Option<String>,
    },
    Refused {
        by: Option<String>,
        reason: Option<String>,
    },
    Expired,
}

/// What an approval leaves: the verdict (run or deny) and the record written.
#[derive(Debug)]
pub struct Answered {
    pub verdict: Verdict,
    pub outcome: DecisionRecord,
}

struct Inner {
    audit: Box<dyn AuditSink>,
    state: StateStore,
    stopped: bool,
}

/// The audit writer and every session's state behind one lock.
pub struct Ledger {
    inner: Mutex<Inner>,
    log: Box<dyn Fn(&str) + Send + Sync>,
}

impl Ledger {
    /// A ledger over the audit writer, the session store, and a sink for its notices.
    pub fn new(
        audit: Box<dyn AuditSink>,
        state: StateStore,
        log: Box<dyn Fn(&str) + Send + Sync>,
    ) -> Ledger {
        Ledger {
            inner: Mutex::new(Inner {
                audit,
                state,
                stopped: false,
            }),
            log,
        }
    }

    /// The lock, `None` when the step cannot proceed: poisoned → `POISONED`, stopped → `STOPPED`.
    fn lock(&self) -> Option<MutexGuard<'_, Inner>> {
        let guard = match self.inner.lock() {
            Ok(guard) => guard,
            Err(_) => {
                (self.log)(POISONED);
                return None;
            }
        };
        if guard.stopped {
            (self.log)(STOPPED);
            return None;
        }
        Some(guard)
    }

    /// Read a session's state, logging and returning "unknown" when it cannot be read.
    fn state_read(
        &self,
        guard: &MutexGuard<Inner>,
        session: &SessionId,
    ) -> Result<SessionState, ()> {
        match guard.state.read(session) {
            Ok(state) => Ok(state),
            Err(e) => {
                (self.log)(&format!("brokerd: {e}"));
                Err(())
            }
        }
    }

    /// Append a record, stopping the ledger on failure and logging why. `Ok(seq)` or `Err`.
    fn append(
        &self,
        guard: &mut MutexGuard<Inner>,
        now: Timestamp,
        event: AuditEvent,
    ) -> Result<u64, ()> {
        match guard.audit.append(now, event) {
            Ok(seq) => Ok(seq),
            Err(e) => {
                guard.stopped = true;
                (self.log)(&format!("brokerd: {e}"));
                Err(())
            }
        }
    }

    /// The refusal `finish` returns when the result could not be recorded.
    fn not_recorded() -> ToolResponse {
        ToolResponse::Failed {
            message: NOT_RECORDED.to_string(),
        }
    }

    /// The answer when the audit is unavailable: deny the verdict and record the denial.
    fn audit_unavailable() -> Answered {
        Answered {
            verdict: Verdict::Denied(DenyReason::AuditUnavailable),
            outcome: DecisionRecord::Denied {
                reason: DenyReason::AuditUnavailable,
            },
        }
    }

    /// A denial verdict and its matching record, for one reason.
    fn denied(reason: DenyReason) -> (Verdict, DecisionRecord) {
        (Verdict::Denied(reason), DecisionRecord::Denied { reason })
    }

    /// The outcome of an approved re-decision: the named grant's mode, else allowed.
    fn outcome_for(&self, decision: &crate::policy::Decision, grants: &Grants) -> DecisionRecord {
        let mode = match grants {
            Ok(set) => set
                .grants()
                .iter()
                .find(|g| g.id == decision.grant())
                .map(|g| g.grant.mode),
            Err(_) => None,
        };
        match mode {
            Some(Mode::Ask) => DecisionRecord::Ask {},
            _ => DecisionRecord::Allowed {},
        }
    }

    /// Decide and record a call.
    pub fn decide(&self, request: ToolRequest, grants: &Grants, now: Timestamp) -> Decided {
        let session = request.session.clone();
        let call = request.call;
        let tool = request.tool.clone();
        let arguments = request.arguments.clone();

        // 1. Lock fails → denied, nothing written.
        let mut guard = match self.lock() {
            Some(guard) => guard,
            None => return Decided::Denied(DenyReason::AuditUnavailable),
        };

        // 2. Read the state, keeping the Result. The recorded state is "unknown" when unreadable.
        let state = self.state_read(&guard, &session);
        let recorded = match &state {
            Ok(state) => *state,
            Err(()) => SessionState {
                taint: DataClass::Secret,
                untrusted: true,
            },
        };

        // 3. The outcome, first that applies.
        let (outcome, grant, grant_sha256, payload) = match grants {
            Err(_) => (
                DecisionRecord::Denied {
                    reason: DenyReason::GrantsInvalid,
                },
                None,
                None,
                RecordPayload::Simple(DenyReason::GrantsInvalid),
            ),
            Ok(set) => match state {
                Err(()) => (
                    DecisionRecord::Denied {
                        reason: DenyReason::StateUnreadable,
                    },
                    None,
                    None,
                    RecordPayload::Simple(DenyReason::StateUnreadable),
                ),
                Ok(state) => match crate::policy::decide(request, set, state, now) {
                    crate::policy::Outcome::Allowed(decision) => (
                        DecisionRecord::Allowed {},
                        Some(decision.grant().to_string()),
                        Some(decision.grant_sha256()),
                        RecordPayload::Allowed(decision),
                    ),
                    crate::policy::Outcome::Ask(ask) => (
                        DecisionRecord::Ask {},
                        Some(ask.grant().to_string()),
                        Some(ask.grant_sha256()),
                        RecordPayload::Ask(ask),
                    ),
                    crate::policy::Outcome::Denied(denial) => (
                        DecisionRecord::Denied {
                            reason: denial.reason,
                        },
                        denial.grant.clone(),
                        denial.grant_sha256,
                        RecordPayload::Denial(denial),
                    ),
                },
            },
        };

        // 4. Write the record, then return the verdict.
        let event = AuditEvent::Decision {
            session,
            call,
            tool,
            arguments,
            outcome,
            grant,
            grant_sha256,
            taint: recorded.taint,
            untrusted: recorded.untrusted,
        };
        match self.append(&mut guard, now, event) {
            Ok(seq) => match payload {
                RecordPayload::Simple(reason) => Decided::Denied(reason),
                RecordPayload::Allowed(decision) => Decided::Allowed { decision, seq },
                RecordPayload::Ask(ask) => Decided::Ask {
                    ask,
                    seq,
                    state: recorded,
                },
                RecordPayload::Denial(denial) => Decided::Denied(denial.reason),
            },
            Err(()) => Decided::Denied(DenyReason::AuditUnavailable),
        }
    }

    /// Re-decide an approval and record it.
    pub fn answer(
        &self,
        ask: Ask,
        decision: u64,
        answer: Answer,
        grants: &Grants,
        now: Timestamp,
    ) -> Answered {
        // 1. Lock fails → step fails.
        let mut guard = match self.lock() {
            Some(guard) => guard,
            None => return Self::audit_unavailable(),
        };

        // 2. Read the state of the ask's session.
        let session = ask.request().session.clone();
        let call = ask.request().call;
        let state = self.state_read(&guard, &session);
        let recorded = match &state {
            Ok(state) => *state,
            Err(()) => SessionState {
                taint: DataClass::Secret,
                untrusted: true,
            },
        };

        // 3. The answer and its fields.
        let (answer, by, reason) = match &answer {
            Answer::Approved { by } => (ApprovalAnswer::Approved, by.clone(), None),
            Answer::Refused { by, reason } => (ApprovalAnswer::Refused, by.clone(), reason.clone()),
            Answer::Expired => (ApprovalAnswer::Expired, None, None),
        };

        // 4. The re-decision (approved only); every other answer is a denial.
        let redecide = match (answer, grants) {
            (ApprovalAnswer::Approved, Ok(set)) => match &state {
                Ok(state) => crate::policy::redecide(ask, set, *state, now),
                Err(()) => Err(Denial::new(DenyReason::StateUnreadable)),
            },
            _ => Err(Denial::new(DenyReason::GrantsInvalid)),
        };

        // 5. The grant fields, then the verdict and outcome together.
        let (grant, grant_sha256) = match &redecide {
            Ok(decision) => (
                Some(decision.grant().to_string()),
                Some(decision.grant_sha256()),
            ),
            Err(_) => (None, None),
        };
        let (verdict, outcome) = match (answer, redecide) {
            (ApprovalAnswer::Approved, Ok(decision)) => {
                let outcome = self.outcome_for(&decision, grants);
                (Verdict::Run(Box::new(decision)), outcome)
            }
            (ApprovalAnswer::Approved, Err(denial)) => Self::denied(denial.reason),
            (ApprovalAnswer::Refused, _) => Self::denied(DenyReason::ApprovalRefused),
            (ApprovalAnswer::Expired, _) => Self::denied(DenyReason::ApprovalExpired),
        };

        // 6. Record the approval, or step fails.
        match self.append(
            &mut guard,
            now,
            AuditEvent::Approval {
                session,
                call,
                decision,
                answer,
                by,
                post: None,
                reason,
                outcome: outcome.clone(),
                grant,
                grant_sha256,
                taint: recorded.taint,
                untrusted: recorded.untrusted,
            },
        ) {
            Ok(_) => Answered { verdict, outcome },
            Err(()) => Self::audit_unavailable(),
        }
    }

    /// Raise the state, hash the response, and record a result, or stop the ledger on failure.
    fn record_result(
        &self,
        guard: &mut MutexGuard<'_, Inner>,
        call: &Call,
        now: Timestamp,
        status: proto::ResultStatus,
        taint_after: DataClass,
        response: &ToolResponse,
    ) -> ToolResponse {
        let (sha256, bytes) = match response {
            ToolResponse::Result { content, .. } => match proto::sha256(content.as_bytes()) {
                Ok(sha) => (sha, u64::try_from(content.len()).unwrap_or(u64::MAX)),
                Err(_) => return Self::not_recorded(),
            },
            ToolResponse::Failed { message } => match proto::sha256(message.as_bytes()) {
                Ok(sha) => (sha, u64::try_from(message.len()).unwrap_or(u64::MAX)),
                Err(_) => return Self::not_recorded(),
            },
            _ => return Self::not_recorded(),
        };
        let (class, untrusted, truncated) = match response {
            ToolResponse::Result {
                class,
                untrusted,
                truncated,
                ..
            } => (*class, *untrusted, *truncated),
            ToolResponse::Failed { .. } => (call.label.class, call.label.untrusted, false),
            _ => return Self::not_recorded(),
        };
        let event = AuditEvent::Result {
            session: call.session.clone(),
            call: call.call,
            decision: call.decision,
            status,
            class,
            untrusted,
            truncated,
            bytes,
            sha256,
            taint_after,
        };
        match self.append(guard, now, event) {
            Ok(_) => response.clone(),
            Err(()) => Self::not_recorded(),
        }
    }

    /// Raise the state and record a result.
    pub fn finish(&self, call: &Call, response: ToolResponse, now: Timestamp) -> ToolResponse {
        // 1. Lock fails → cannot record.
        let mut guard = match self.lock() {
            Some(guard) => guard,
            None => return Self::not_recorded(),
        };

        match &response {
            // 3. A result raises the state, then is recorded.
            ToolResponse::Result { .. } => {
                let raised = match self.state_read(&guard, &call.session) {
                    Ok(state) => match guard.state.raise(&call.session, state, call.label) {
                        Ok(raised) => raised,
                        Err(e) => {
                            (self.log)(&format!("brokerd: {e}"));
                            return Self::not_recorded();
                        }
                    },
                    Err(()) => return Self::not_recorded(),
                };
                self.record_result(
                    &mut guard,
                    call,
                    now,
                    proto::ResultStatus::Result,
                    raised.taint,
                    &response,
                )
            }
            // 4. A failure changes no state, recorded by its message.
            ToolResponse::Failed { .. } => {
                let taint_after = match self.state_read(&guard, &call.session) {
                    Ok(state) => state.taint,
                    Err(()) => DataClass::Secret,
                };
                self.record_result(
                    &mut guard,
                    call,
                    now,
                    proto::ResultStatus::Failed,
                    taint_after,
                    &response,
                )
            }
            // 5. The runner never returns these; nothing to record.
            ToolResponse::Denied { .. } | ToolResponse::PendingApproval { .. } => {
                Self::not_recorded()
            }
        }
    }
}