//! Node-side data sources the gRPC service serves from (m11p2).
//!
//! The `WalShipping` service needs two read-only views of the embedding
//! node's state that tidal-net itself cannot own:
//!
//! - [`AppliedSource`]: the node's contiguous applied seqno per source shard,
//!   piggybacked on every `ShipSegmentResponse` so a leader's ship queue can
//!   fold the follower's own progress into its acked frontier.
//! - [`SegmentSource`]: read-back of the node's durable WAL batches, serving
//!   the `StreamSegments` catch-up path (a follower that reports applied=N
//!   pulls a stream from N+1).
//!
//! Both are wired by the embedding application (tidal-server hands in
//! adapters over its `TidalDb`); `None` preserves the pre-m11p2 behavior
//! (acks carry `applied_seqno = 0`, `StreamSegments` answers `Unimplemented`).

use std::path::PathBuf;
use std::sync::Arc;

use tidaldb::replication::shard::ShardId;

/// One contiguous run of encoded WAL batches served to a catch-up stream.
#[derive(Debug, Clone)]
pub struct SegmentChunk {
    /// Concatenated encoded batches, byte-identical to the leader's disk.
    pub bytes: Vec<u8>,
    /// First WAL seqno covered.
    pub first_seq: u64,
    /// Last WAL seqno covered.
    pub last_seq: u64,
    /// Records covered (blob batches count 1 each).
    pub event_count: u64,
}

/// The node's per-source-shard applied seqno, for ack piggybacking.
pub trait AppliedSource: Send + Sync + 'static {
    /// The contiguous applied seqno for `source_shard` (0 = nothing applied).
    fn applied_seqno(&self, source_shard: ShardId) -> u64;
}

/// Leader-side consumer of follower durable-frontier reports (m11p3).
///
/// The `ReportApplied` handler invokes it with the reporter's shard and its
/// contiguous durably-applied seqno for THIS node's stream — the input that
/// advances the quorum commit index. Wired late (the embedding application
/// builds its ship queue after the transport), hence the `OnceLock` cell on
/// [`ServingSources`].
pub trait AppliedSink: Send + Sync + 'static {
    /// Fold a follower's durable mark (monotonic; stale or unknown-peer
    /// reports must be ignored by the implementation). `reporter_term` is
    /// the reporter's current term (m11p4): the implementation must fold
    /// ONLY when it matches the commit index's activation term — the gate
    /// runs under the index's own lock so a leadership change can never
    /// race a fold (design-review C4/C8/C12).
    fn peer_applied(&self, peer: ShardId, applied: u64, reporter_term: u64);
}

/// Why a segment read-back could not serve a catch-up request.
///
/// The split is the wire contract (m11p4): the `StreamSegments` handler maps
/// [`Unavailable`](Self::Unavailable) to `FAILED_PRECONDITION` ("segments not
/// available from seq N; snapshot required") and [`Failed`](Self::Failed) to
/// `INTERNAL` (transient; the follower's retry timer re-pulls). Without the
/// distinction, a leader whose WAL predates this binary's segment format
/// (rolling upgrade) answered with a generic internal error and the follower
/// could not tell "retry later" from "this log will never serve you".
#[derive(Debug, Clone)]
pub enum SegmentReadError {
    /// The requested range can NEVER be served from this node's WAL — its
    /// segments carry a format this binary cannot read (written by another
    /// tidalDB version), or the node has no durable log to serve. The
    /// follower needs a snapshot (m11p5) or an operator reseed.
    Unavailable {
        /// Human-readable cause, forwarded verbatim in the status message.
        detail: String,
    },
    /// A transient read/validation failure; retrying can succeed.
    Failed {
        /// Human-readable cause, forwarded verbatim in the status message.
        detail: String,
    },
}

impl std::fmt::Display for SegmentReadError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Unavailable { detail } => write!(f, "unavailable: {detail}"),
            Self::Failed { detail } => write!(f, "read failed: {detail}"),
        }
    }
}

/// Read-back over the node's durable WAL for the catch-up stream.
///
/// Implementations are synchronous (they read segment files); the service
/// calls them through `spawn_blocking`, never on the reactor.
pub trait SegmentSource: Send + Sync + 'static {
    /// The shard whose stream this node serves (its own).
    fn source_shard(&self) -> ShardId;

    /// The stream's baseline: the WAL seqno at which this node's stream
    /// started (0 = from the beginning; non-zero after a promote). Seqnos at
    /// or below it are never served — they are pre-stream history.
    fn stream_baseline(&self) -> u64;

    /// The current flushed frontier (the stream's end at snapshot time).
    fn flushed_seq(&self) -> u64;

    /// Collect encoded batches covering seqnos `>= from_seq`, bounded by
    /// `max_events` / `max_bytes`. Empty means caught up. A chunk whose
    /// `first_seq > from_seq` means the segments below it were compacted.
    ///
    /// # Errors
    ///
    /// [`SegmentReadError::Unavailable`] when the range can never be served
    /// from this log (snapshot required); [`SegmentReadError::Failed`] for a
    /// transient read/validation failure.
    fn collect_from(
        &self,
        from_seq: u64,
        max_events: u64,
        max_bytes: usize,
    ) -> Result<Vec<SegmentChunk>, SegmentReadError>;
}

/// A staged snapshot artifact ready to stream (m11p5 §2), from
/// [`SnapshotSource::stage`].
///
/// When `needed` is `false` the live
/// WAL can still serve the puller's `from_seqno` — the joiner just streams,
/// and `files`/`root` are empty/unused. When `true`, `files` is the complete
/// manifest (relative paths, sizes, content hashes) under `root`; the service
/// streams each file's bytes off `spawn_blocking` reads.
///
/// The BLAKE3 in each manifest entry is the end-to-end integrity contract:
/// computed ONCE when the artifact is staged, forwarded verbatim on the wire,
/// and verified by the installing puller (the server never recomputes it per
/// chunk — the manifest hash is authoritative).
#[derive(Debug, Clone)]
pub struct SnapshotStaging {
    /// `false` = no snapshot needed, just stream from `from_seqno`.
    pub needed: bool,
    /// The artifact's recovered WAL tail: the seqno the snapshot is valid at
    /// (the puller advances its frontier to this, resumes from `+ 1`). 0 when
    /// `needed` is `false`.
    pub snapshot_seq: u64,
    /// The staging directory the manifest's relative paths resolve against.
    pub root: PathBuf,
    /// The complete manifest: `(relative_path, size, blake3)` per file. Empty
    /// when `needed` is `false`.
    pub files: Vec<(String, u64, [u8; 32])>,
}

/// Why a snapshot could not be staged for a `FetchSnapshot` request.
///
/// The split is the wire contract (m11p5 §2): [`Busy`](Self::Busy) maps to a
/// **retryable** `UNAVAILABLE` (a concurrent stage is mid-flight; a second
/// joiner shares the in-flight artifact moments later — NEVER
/// `FAILED_PRECONDITION`, which would mis-route the joiner into the reseed
/// class), and [`Failed`](Self::Failed) maps to `INTERNAL` (staging hit a real
/// fault — disk, checkpoint, copy — the puller retries on the next seed loop).
#[derive(Debug, Clone)]
pub enum SnapshotStageError {
    /// A stage is already in flight (single-flight artifact, §2 caching): the
    /// puller retries and shares the result. Wire: retryable `UNAVAILABLE`.
    Busy,
    /// Staging failed for a concrete reason. Wire: `INTERNAL`.
    Failed(String),
}

impl std::fmt::Display for SnapshotStageError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Busy => write!(f, "snapshot staging busy (single-flight; retry)"),
            Self::Failed(detail) => write!(f, "snapshot staging failed: {detail}"),
        }
    }
}

/// A node-side staged snapshot artifact, served by the `FetchSnapshot` stream
/// (m11p5 §2).
///
/// Late-bound like [`SegmentSource`]: the embedding application
/// wires its `TidalDb`-backed adapter after the transport. `None` = the RPC
/// answers `Unimplemented` (a pre-m11p5 peer, or a bare transport).
///
/// Implementations are synchronous (they checkpoint + copy + read files); the
/// service calls [`stage`](Self::stage) on the reactor (it must be cheap —
/// reuse an already-staged artifact) and reads file bytes through
/// `spawn_blocking`.
pub trait SnapshotSource: Send + Sync + 'static {
    /// Stage (or reuse) the snapshot artifact that covers `from_seqno`, and
    /// pin its retention until [`release`](Self::release) (§2.1 per-consumer
    /// pin grace). Returns the manifest the service streams.
    ///
    /// # Errors
    ///
    /// [`SnapshotStageError::Busy`] when a stage is already in flight (the
    /// puller retries — retryable `UNAVAILABLE`); [`SnapshotStageError::Failed`]
    /// when staging hit a concrete fault (`INTERNAL`).
    fn stage(&self, from_seqno: u64) -> Result<SnapshotStaging, SnapshotStageError>;

    /// Signal that a `FetchSnapshot` stream finished (success OR client drop):
    /// the service calls this exactly once per [`stage`](Self::stage) that
    /// returned a `needed=true` staging, releasing that consumer's retention
    /// pin (§2.1). A `needed=false` staging holds no pin and is not released.
    ///
    /// Idempotent — the node-side impl (stage B) reconciles double calls.
    fn release(&self);
}

/// Late-bound consumer of a follower's typed "snapshot-required" catch-up
/// refusal (m11p5 §2.4).
///
/// When the puller's `StreamSegments` open is refused with `FAILED_PRECONDITION`
/// carrying the `x-tidal-catchup: snapshot-required` trailer, the transport
/// invokes this so the node can durably latch its `reseed_required` marker (the
/// reseed runs on the NEXT boot, never via live surgery). Absent (bare
/// transports, pre-m11p5) → today's behavior: log + the standing retry timer.
///
/// Wired like [`AppliedSink`] (an `OnceLock` on [`ServingSources`]): the node
/// builds it after the transport.
pub trait SnapshotRequiredSink: Send + Sync + 'static {
    /// The catch-up stream from `shard` (starting at `from_seqno`) was refused
    /// `snapshot-required`. Latch the reseed marker; the retry timer keeps its
    /// standing wake-up regardless (the m11p4 re-arm-on-skip liveness fix).
    fn snapshot_required(&self, shard: ShardId, from_seqno: u64);
}

/// The heartbeat exchange verdict an [`ElectionHooks`] implementation
/// returns: the responder's (possibly just-raised) term and whether the
/// sender's leadership assertion was accepted.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct HeartbeatExchange {
    /// The responder's current term.
    pub term: u64,
    /// Whether the sender's term was accepted as current leadership.
    pub accepted: bool,
}

/// The leader-heartbeat payload a follower folds (m11p4 + m12p5).
///
/// Grouped into a struct so new heartbeat-carried state is an ADDITIVE field
/// rather than a wider positional signature on [`ElectionHooks::on_heartbeat`]
/// — and so the several `u64`s (`term`, `stream_baseline`, `leader_last_seq`)
/// can't be silently transposed at a call site.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct HeartbeatContext {
    /// The leader's election term.
    pub term: u64,
    /// The leader's region id.
    pub leader_region: u16,
    /// The term's immutable activation baseline (the stream's term-start point).
    pub stream_baseline: u64,
    /// The leader's election-time log position `(prev_log_term, prev_log_seq)`.
    pub prev_log: tidaldb::replication::LogPosition,
    /// The leader's LIVE flushed frontier at heartbeat time (m12p5): the
    /// high-water-mark every follower converges to, in the same stream numbering
    /// as a follower's per-shard `applied_seqno`. `0` = a pre-m12p5 leader
    /// conveyed no frontier (the heartbeat readiness drive is skipped).
    pub leader_last_seq: u64,
}

/// Why an inbound leadership-stamped payload was refused (m11p4).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ClaimRejection {
    /// The claim's term is below this node's (or this node is quarantined):
    /// permanent for that leadership — the sender steps down or stays fenced.
    Stale {
        /// This node's current term.
        current_term: u64,
    },
    /// The claim's term is NEW to this node and the term-join check (which
    /// only a heartbeat, carrying the leader's election-time position, can
    /// run) has not happened yet. TRANSIENT: the sender retries after the
    /// next heartbeat round joins the term.
    JoinPending,
}

/// Node-side election integration (m11p4), late-bound like [`AppliedSink`].
///
/// The gRPC service consults it to FENCE inbound traffic (stale-term ships,
/// stream chunks and frontier reports never reach the apply path) and to
/// answer the election RPCs. Absent hooks = the pre-m11p4 behavior: every
/// inbound is accepted, vote RPCs answer `Unimplemented` — exactly what a
/// bare transport (tests, in-process harnesses) wants.
///
/// Implementations own the election state machine and its durable hard
/// state; every method that can adopt a term MUST persist before returning
/// (the reply is the externally visible action).
pub trait ElectionHooks: Send + Sync + 'static {
    /// This node's `(current_term, region)` — stamped on outbound responses
    /// and served stream chunks.
    fn self_claim(&self) -> (u64, u16);

    /// An inbound leadership-stamped data payload (live ship or catch-up
    /// chunk) covering seqnos from `first_seq`. `Ok(())` = accept (the claim
    /// was current; leader contact was recorded; any higher term was
    /// adopted-and-persisted before return). `Err(current_term)` = the
    /// payload is fenced; the caller rejects it with `FAILED_PRECONDITION`
    /// carrying the current term.
    ///
    /// # Errors
    ///
    /// [`ClaimRejection::Stale`] when `term` is below this node's current
    /// term or this node is quarantined (divergent suffix — no data-plane
    /// participation until reseeded); [`ClaimRejection::JoinPending`] when
    /// the term is new and only the next heartbeat (which carries the
    /// leader's election-time position) can run the divergence check —
    /// transient, the sender retries.
    fn observe_leader_claim(
        &self,
        term: u64,
        leader_region: u16,
        first_seq: u64,
    ) -> Result<(), ClaimRejection>;

    /// A leader heartbeat (payload: [`HeartbeatContext`]): term + leadership +
    /// the term's activation baseline + the leader's election-time log position
    /// `(prev_log_term, prev_log_seq)` + the leader's LIVE flushed frontier
    /// (`leader_last_seq`). Drives the failure detector, the term-join
    /// divergence check, and (m12p5) the sticky readiness latch — a caught-up
    /// joiner converges from the heartbeat (which flows on an idle cluster),
    /// not only from observed ship traffic. `leader_last_seq == 0` means a
    /// pre-m12p5 leader conveyed no frontier (the readiness drive is skipped).
    fn on_heartbeat(&self, hb: HeartbeatContext) -> HeartbeatExchange;

    /// A pre-vote or vote request. The grant (and any term adoption) is
    /// durable before this returns.
    fn on_vote(&self, rpc: tidaldb::replication::VoteRpc) -> tidaldb::replication::VoteReply;

    /// The current leader sanctioned an immediate transfer election.
    /// Returns whether a candidacy started.
    fn on_timeout_now(&self, term: u64, leader_region: u16) -> bool;

    /// A response/request carried a term above ours outside the paths above
    /// (e.g. a stale-source `StreamSegments` request from a newer-term
    /// puller): adopt + persist + step down.
    fn on_observed_term(&self, term: u64);

    /// Whether a frontier report stamped `reporter_term` may fold into the
    /// quorum commit index / hint map (true iff it matches the current
    /// leadership term).
    fn report_term_acceptable(&self, reporter_term: u64) -> bool;

    /// The TYPED REMOVED SIGNAL (m11p5 §3.3): whether `region` is a `Removed`
    /// member in THIS node's APPLIED roster. The heartbeat / vote handlers call
    /// it with the inbound peer's region (`region_id` / `candidate_region`) and
    /// stamp the `removed` bit on the response.
    ///
    /// This is the out-of-band delivery channel for a removed node that MISSED
    /// the `Removed` record in the stream (it was down during the removal-
    /// delivery grace, §3.3): it learns of its removal from any voter's
    /// heartbeat / vote refusal rather than the log, flips readiness to 503, and
    /// suppresses campaigning — WITHOUT latching a reseed marker (a remove is
    /// not a reseed). Defaults to `false` (the era-0 / pre-membership node has
    /// no `Removed` members and a bare transport has no roster).
    fn is_removed_member(&self, _region: u16) -> bool {
        false
    }
}

/// The leader-side outcome of a [`JoinHooks::join`] (m11p5 §3.3).
///
/// Mirrors the `JoinResponse` proto, but in domain terms so tidal-net never
/// reaches into the engine's membership types. The transport adapter maps this
/// 1:1 onto the wire response.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JoinOutcome {
    /// `false` = refused; `refusal_reason` says why and (for a non-leader seed)
    /// the `leader_*` fields name where to re-target.
    pub accepted: bool,
    /// Human-readable refusal cause (empty when `accepted`).
    pub refusal_reason: String,
    /// The id the joiner was assigned (or its existing id on a re-join).
    pub assigned_id: u16,
    /// The current leadership term.
    pub term: u64,
    /// The leader's region name + advertised addresses (for re-targeting and
    /// the joiner's direct dial). Empty when no leader is known.
    pub leader_region: String,
    pub leader_grpc_addr: String,
    pub leader_http_addr: String,
    /// The full roster after the join: `(id, name, grpc_addr, http_addr, role)`
    /// where `role` is the `MemberRole` discriminant byte (0/1/2).
    pub members: Vec<MemberInfo>,
    /// The conf version the roster is at.
    pub membership_version: u64,
}

/// One member in a [`JoinOutcome`] roster (the domain shape of the
/// `MemberInfo` proto).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MemberInfo {
    pub id: u16,
    pub name: String,
    pub grpc_addr: String,
    pub http_addr: String,
    /// `MemberRole` discriminant byte: 0 Voter, 1 Learner, 2 Removed.
    pub role: u8,
}

/// The joiner's request, as the leader-side hook sees it (m11p5 §3.3).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JoinAsk {
    pub name: String,
    pub grpc_addr: String,
    pub http_addr: String,
    /// The joiner binary's capability bit-field.
    pub capabilities: u64,
}

/// Node-side `JoinCluster` integration (m11p5 §3.3), late-bound like
/// [`ElectionHooks`] (the `OnceLock` on [`ServingSources`]).
///
/// The gRPC service consults it to answer a join: a non-leader refuses with a
/// leader hint, the leader assigns a permanent id, appends a Learner record,
/// waits for same-term quorum commit, and returns the roster. Absent =
/// `JoinCluster` answers `Unimplemented` (a pre-m11p5 peer, or a bare
/// transport).
///
/// The implementation owns the membership runtime; this trait keeps the engine
/// types out of tidal-net (the adapter lives in tidal-server).
pub trait JoinHooks: Send + Sync + 'static {
    /// Handle a join. The hook BLOCKS on the bounded same-term commit wait —
    /// the service calls it on `spawn_blocking`, never the reactor.
    fn join(&self, ask: JoinAsk) -> JoinOutcome;
}

/// The optional node-side sources handed to [`crate::GrpcTransport`].
#[derive(Clone, Default)]
pub struct ServingSources {
    /// Applied-seqno reader for ack piggybacking (`None` = acks carry 0).
    pub applied: Option<Arc<dyn AppliedSource>>,
    /// WAL read-back for `StreamSegments` (`None` = catch-up unimplemented).
    pub segments: Option<Arc<dyn SegmentSource>>,
    /// Staged-snapshot source for `FetchSnapshot` (m11p5). A `OnceLock`
    /// because the node wires its `TidalDb`-backed adapter after the
    /// transport (the snapshot artifact needs the live engine + data dir).
    /// `None`/unset = `FetchSnapshot` answers `Unimplemented`.
    pub snapshots: Arc<std::sync::OnceLock<Arc<dyn SnapshotSource>>>,
    /// Late-bound consumer of typed `snapshot-required` catch-up refusals
    /// (m11p5 §2.4). Unset = log + retry-timer only (pre-m11p5 behavior).
    pub snapshot_required: Arc<std::sync::OnceLock<Arc<dyn SnapshotRequiredSink>>>,
    /// Consumer of follower durable-frontier reports (m11p3). A `OnceLock`
    /// because the embedding application can only build it AFTER the
    /// transport exists (the ship queue takes the transport): fill it via
    /// [`ServingSources::set_applied_sink`] once available. Reports arriving
    /// before it is set fold into the transport's hint map only.
    pub applied_sink: Arc<std::sync::OnceLock<Arc<dyn AppliedSink>>>,
    /// Election integration (m11p4), late-bound for the same reason as
    /// `applied_sink` (the election driver is built after the transport).
    /// Absent = pre-m11p4 behavior (no fencing, vote RPCs unimplemented).
    pub election: Arc<std::sync::OnceLock<Arc<dyn ElectionHooks>>>,
    /// `JoinCluster` integration (m11p5 §3.3), late-bound: the membership
    /// runtime is built after the transport. Absent = `JoinCluster` answers
    /// `Unimplemented`.
    pub join: Arc<std::sync::OnceLock<Arc<dyn JoinHooks>>>,
}

impl ServingSources {
    /// Late-bind the applied-report sink (idempotent; first set wins).
    pub fn set_applied_sink(&self, sink: Arc<dyn AppliedSink>) {
        let _ = self.applied_sink.set(sink);
    }

    /// Late-bind the election hooks (idempotent; first set wins).
    pub fn set_election_hooks(&self, hooks: Arc<dyn ElectionHooks>) {
        let _ = self.election.set(hooks);
    }

    /// Late-bind the snapshot source for `FetchSnapshot` (idempotent; first
    /// set wins). Until set, `FetchSnapshot` answers `Unimplemented`.
    pub fn set_snapshot_source(&self, source: Arc<dyn SnapshotSource>) {
        let _ = self.snapshots.set(source);
    }

    /// Late-bind the `snapshot-required` refusal sink (idempotent; first set
    /// wins). Until set, a `snapshot-required` trailer is log + retry only.
    pub fn set_snapshot_required_sink(&self, sink: Arc<dyn SnapshotRequiredSink>) {
        let _ = self.snapshot_required.set(sink);
    }

    /// Late-bind the `JoinCluster` hooks (idempotent; first set wins). Until
    /// set, `JoinCluster` answers `Unimplemented`.
    pub fn set_join_hooks(&self, hooks: Arc<dyn JoinHooks>) {
        let _ = self.join.set(hooks);
    }
}

impl std::fmt::Debug for ServingSources {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ServingSources")
            .field("applied", &self.applied.is_some())
            .field("segments", &self.segments.is_some())
            .field("snapshots", &self.snapshots.get().is_some())
            .field("applied_sink", &self.applied_sink.get().is_some())
            .field("election", &self.election.get().is_some())
            .field("snapshot_required", &self.snapshot_required.get().is_some())
            .field("join", &self.join.get().is_some())
            .finish()
    }
}