batpak 0.10.0

use super::super::fanout::CommittedEventEnvelope;
use super::super::staging::{PreparedBatch, StagedCommittedEvent};
use super::{kind_to_raw, Notification, WriterCore};
use crate::store::index::{DiskPos, IndexEntry};
use crate::store::segment::sidx::SidxEntry;
use crate::store::stats::HlcPoint;
use crate::store::{AppendReceipt, EncodedBytes, ExtensionKey};
use std::collections::BTreeMap;

fn broadcast_all<T>(values: impl IntoIterator<Item = T>, mut broadcast: impl FnMut(&T)) -> usize {
    let mut count = 0usize;
    for value in values {
        count += 1;
        broadcast(&value);
    }
    count
}

#[derive(Clone, Copy)]
struct LanePublishPoint {
    publish_up_to: u64,
    frontier_point: HlcPoint,
}

fn lane_publish_points_from_notifications(
    notifications: &[Notification],
) -> BTreeMap<u32, LanePublishPoint> {
    let mut points = BTreeMap::new();
    for notification in notifications {
        let lane = notification.position.lane();
        let publish_up_to = notification.sequence.saturating_add(1);
        let frontier_point = HlcPoint {
            wall_ms: notification.position.wall_ms(),
            global_sequence: notification.sequence,
        };
        points
            .entry(lane)
            .and_modify(|current: &mut LanePublishPoint| {
                if publish_up_to > current.publish_up_to {
                    *current = LanePublishPoint {
                        publish_up_to,
                        frontier_point,
                    };
                }
            })
            .or_insert(LanePublishPoint {
                publish_up_to,
                frontier_point,
            });
    }
    points
}

pub(super) struct CommitArtifacts {
    pub(super) index_entry: IndexEntry,
    pub(super) sidx_entry: SidxEntry,
    pub(super) notification: Notification,
    pub(super) envelope: Option<CommittedEventEnvelope>,
}

#[derive(Clone, Copy)]
pub(super) struct CommitInternedIds {
    pub(super) entity_id: crate::store::index::interner::InternId,
    pub(super) scope_id: crate::store::index::interner::InternId,
}

impl CommitInternedIds {
    /// Intern a coordinate's entity and scope strings into compact ids. Returns
    /// [`crate::store::StoreError::InternerExhausted`] if the `u32` interner id
    /// domain is exhausted.
    pub(super) fn for_coord(
        index: &crate::store::index::StoreIndex,
        coord: &crate::coordinate::Coordinate,
    ) -> Result<Self, crate::store::StoreError> {
        Ok(Self {
            entity_id: index.interner.intern(coord.entity())?,
            scope_id: index.interner.intern(coord.scope())?,
        })
    }
}

pub(super) struct BatchCommitArtifacts {
    pub(super) entries: Vec<IndexEntry>,
    pub(super) sidx_entries: Vec<SidxEntry>,
    pub(super) notifications: Vec<Notification>,
    pub(super) envelopes: Vec<CommittedEventEnvelope>,
}

#[derive(Clone, Copy)]
pub(super) struct CommitFrameView<'a> {
    pub(super) payload_bytes: &'a [u8],
    pub(super) flags: u8,
    pub(super) receipt_extensions: &'a BTreeMap<ExtensionKey, EncodedBytes>,
    pub(super) emit_envelope: bool,
}

/// A reactor requested an envelope but the committed payload could not be
/// decoded into its plaintext form (most notably an encrypted append, whose
/// committed frame bytes are ciphertext under `payload-encryption`).
///
/// Surfaced by [`reactor_envelope`] as `Err(_)` — and logged LOUDLY by the
/// caller (mirroring `reactor_delivery::warn_shredded_reactor_delivery`) —
/// rather than swallowed to a bare `None`: the previous `.ok()` collapsed "no
/// reactor asked" and "the plaintext could not be built" into the same silent
/// absence, hiding a starved reactor behind a dropped `Result::Err`.
struct ReactorEnvelopeUnavailable;

/// Derive the in-process reactor [`CommittedEventEnvelope`] for a committed
/// event from its staged facts. The envelope carries the DECODED plaintext so an
/// in-process reactor ([`crate::store::Store::react_loop`]) is served without a
/// disk re-read.
///
/// Returns `Ok(None)` when no reactor subscriber wants an envelope, `Ok(Some(_))`
/// when the plaintext envelope was built, or `Err(ReactorEnvelopeUnavailable)`
/// when a reactor wanted one but the committed payload could not be decoded into
/// the plaintext envelope form. Pure over its inputs so the three outcomes are
/// unit-tested directly.
fn reactor_envelope(
    staged: &StagedCommittedEvent,
    notification: &Notification,
    frame: CommitFrameView<'_>,
) -> Result<Option<CommittedEventEnvelope>, ReactorEnvelopeUnavailable> {
    if !frame.emit_envelope {
        return Ok(None);
    }
    match staged.stored_event(frame.payload_bytes, frame.flags) {
        Ok(stored) => Ok(Some(CommittedEventEnvelope {
            notification: notification.clone(),
            stored,
        })),
        // The committed payload did not decode into a plaintext envelope. This
        // is expected for an encrypted append (the committed bytes are
        // ciphertext); it is a genuine fault for a non-canonical plaintext
        // payload. Either way the reactor cannot be served this event, so the
        // outcome is surfaced (and logged by the caller), never dropped.
        Err(_error) => Err(ReactorEnvelopeUnavailable),
    }
}

/// Emit the observable, structured warn for a committed event whose plaintext
/// reactor envelope could not be built. LOUD (this warn), never silent —
/// mirrors `reactor_delivery::warn_shredded_reactor_delivery`.
fn warn_reactor_envelope_unavailable(
    coord: &crate::coordinate::Coordinate,
    event_id: crate::id::EventId,
) {
    use crate::id::EntityIdType;
    tracing::warn!(
        target: "batpak::fanout",
        flow = "reactor",
        entity = coord.entity(),
        event_id = event_id.as_u128(),
        "reactor envelope unavailable: the committed payload could not be decoded into a \
         plaintext reactor envelope (an encrypted or non-canonical payload); the in-process \
         reactor is not served this event"
    );
}

impl BatchCommitArtifacts {
    pub(super) fn with_capacity(len: usize) -> Self {
        Self {
            entries: Vec::with_capacity(len),
            sidx_entries: Vec::with_capacity(len),
            notifications: Vec::with_capacity(len),
            envelopes: Vec::with_capacity(len),
        }
    }

    fn push(&mut self, committed: CommitArtifacts) {
        self.entries.push(committed.index_entry);
        self.sidx_entries.push(committed.sidx_entry);
        self.notifications.push(committed.notification);
        if let Some(envelope) = committed.envelope {
            self.envelopes.push(envelope);
        }
    }
}

impl WriterCore {
    pub(super) fn materialize_commit_artifacts(
        &self,
        staged: &StagedCommittedEvent,
        disk_pos: DiskPos,
        interned_ids: CommitInternedIds,
        frame: CommitFrameView<'_>,
    ) -> CommitArtifacts {
        let coord = staged.coord.clone();
        let position = staged.position();
        let notification = Notification {
            event_id: crate::id::EventId::from_u128(staged.meta.event_id),
            correlation_id: staged.meta.correlation_id,
            causation_id: staged.meta.causation_id,
            coord: coord.clone(),
            kind: staged.meta.kind,
            sequence: staged.meta.global_sequence,
            position,
        };
        let index_entry = IndexEntry {
            event_id: staged.meta.event_id,
            correlation_id: staged.meta.correlation_id,
            causation_id: staged.meta.causation_id,
            coord: coord.clone(),
            entity_id: interned_ids.entity_id,
            scope_id: interned_ids.scope_id,
            kind: staged.meta.kind,
            wall_ms: staged.timing.wall_ms,
            clock: staged.timing.clock,
            dag_lane: staged.timing.dag_lane,
            dag_depth: staged.timing.dag_depth,
            hash_chain: staged.hash_chain.clone(),
            disk_pos,
            global_sequence: staged.meta.global_sequence,
            receipt_extensions: frame.receipt_extensions.clone(),
        };
        let sidx_entry = SidxEntry {
            event_id: staged.meta.event_id,
            // Placeholder string-table slots: SidxEntryCollector::record rewrites
            // them to the correct entity/scope indexes when the footer is built.
            entity_idx: 0,
            scope_idx: 0,
            kind: kind_to_raw(staged.meta.kind),
            wall_ms: staged.timing.wall_ms,
            clock: staged.timing.clock,
            dag_lane: staged.timing.dag_lane,
            dag_depth: staged.timing.dag_depth,
            prev_hash: staged.hash_chain.prev_hash,
            event_hash: staged.hash_chain.event_hash,
            frame_offset: disk_pos.offset,
            frame_length: disk_pos.length,
            global_sequence: staged.meta.global_sequence,
            correlation_id: staged.meta.correlation_id,
            causation_id: staged.meta.causation_id.unwrap_or(0),
        };
        let envelope = match reactor_envelope(staged, &notification, frame) {
            Ok(envelope) => envelope,
            Err(ReactorEnvelopeUnavailable) => {
                warn_reactor_envelope_unavailable(&coord, notification.event_id);
                None
            }
        };
        CommitArtifacts {
            index_entry,
            sidx_entry,
            notification,
            envelope,
        }
    }

    /// STEP 12/14: Materialize all post-write views in one pass from the
    /// committed staged facts plus receipts. This is the product split over
    /// the same semantic source, so index/SIDX/notification/envelope derivation
    /// cannot silently drift apart.
    pub(super) fn materialize_batch_commit_artifacts(
        &self,
        prepared: &PreparedBatch,
        staged: &[StagedCommittedEvent],
        receipts: &[AppendReceipt],
    ) -> Result<BatchCommitArtifacts, crate::store::StoreError> {
        let emit_envelope = self.reactor_subscribers.has_subscribers();
        let mut artifacts = BatchCommitArtifacts::with_capacity(staged.len());
        let interned = prepared.interned_ids(&self.index)?;

        for (((item, staged), receipt), ids) in prepared
            .items()
            .iter()
            .zip(staged.iter())
            .zip(receipts.iter())
            .zip(interned.iter())
        {
            let committed = self.materialize_commit_artifacts(
                staged,
                receipt.disk_pos,
                CommitInternedIds {
                    entity_id: ids.entity_id,
                    scope_id: ids.scope_id,
                },
                CommitFrameView {
                    payload_bytes: item.payload_bytes(),
                    flags: item.options().flags,
                    receipt_extensions: &receipt.extensions,
                    emit_envelope,
                },
            );
            artifacts.push(committed);
        }

        Ok(artifacts)
    }

    pub(super) fn broadcast_commit_artifacts(
        &self,
        notifications: impl IntoIterator<Item = Notification>,
        envelopes: impl IntoIterator<Item = CommittedEventEnvelope>,
    ) {
        let push_notifications = broadcast_all(notifications, |notification| {
            self.subscribers.broadcast(notification)
        });
        let push_envelopes = broadcast_all(envelopes, |envelope| {
            self.reactor_subscribers.broadcast(envelope);
        });
        tracing::trace!(
            target: "batpak::fanout",
            push_notifications,
            push_envelopes,
            "commit fanout batch",
        );
    }

    /// Publishes the index boundary for an unfenced commit, then notifies
    /// subscribers. ORDER IS LOAD-BEARING: subscribers woken by the broadcast
    /// must already be able to observe the events as visible — if the
    /// broadcast ran before the publish, a subscriber could read the
    /// notification, query the store, and see the entry still hidden.
    ///
    /// Call-sites for unfenced commits use only this helper; the raw
    /// `publish` + `broadcast_commit_artifacts` calls are intentionally kept
    /// private to this module so this ordering contract cannot be swapped by
    /// mistake.
    #[inline]
    pub(super) fn publish_then_broadcast_unfenced(
        &mut self,
        publish_up_to: u64,
        frontier_point: HlcPoint,
        notifications: impl IntoIterator<Item = Notification>,
        envelopes: impl IntoIterator<Item = CommittedEventEnvelope>,
    ) -> Result<(), crate::store::StoreError> {
        let notifications: Vec<Notification> = notifications.into_iter().collect();
        let lane_points = lane_publish_points_from_notifications(&notifications);
        self.index.publish_on_lanes(
            publish_up_to,
            lane_points
                .iter()
                .map(|(lane, point)| (*lane, point.publish_up_to)),
            "publish_then_broadcast_unfenced",
        )?;
        self.broadcast_commit_artifacts(notifications, envelopes);
        let mut watermark = self.watermark_handle.lock();
        watermark.advance_visible_and_emitted(frontier_point);
        for (lane, point) in lane_points {
            watermark.advance_visible_and_emitted_on_lane(lane, point.frontier_point);
        }
        Ok(())
    }

    /// Finishes a visibility fence (publishes the hidden range), then notifies
    /// subscribers. Same ordering contract as
    /// [`publish_then_broadcast_unfenced`]: visibility must be established
    /// before the broadcast, or a subscriber could observe a notification for
    /// an entry that is still hidden.
    ///
    /// `publish_up_to` is an `Option<u64>` because a fence with no recorded
    /// progress (no fenced appends committed) finishes without advancing
    /// the visible watermark; the index's `finish_visibility_fence` accepts
    /// that shape directly.
    #[inline]
    pub(super) fn fence_finish_then_broadcast(
        &mut self,
        token: u64,
        publish_up_to: Option<u64>,
        frontier_point: Option<HlcPoint>,
        notifications: impl IntoIterator<Item = Notification>,
        envelopes: impl IntoIterator<Item = CommittedEventEnvelope>,
    ) -> Result<(), crate::store::StoreError> {
        let notifications: Vec<Notification> = notifications.into_iter().collect();
        let lane_points = lane_publish_points_from_notifications(&notifications);
        self.index.finish_visibility_fence_on_lanes(
            token,
            publish_up_to,
            lane_points
                .iter()
                .map(|(lane, point)| (*lane, point.publish_up_to)),
        )?;
        self.broadcast_commit_artifacts(notifications, envelopes);
        let mut watermark = self.watermark_handle.lock();
        if let Some(point) = frontier_point {
            watermark.advance_visible_and_emitted(point);
        }
        for (lane, point) in lane_points {
            watermark.advance_visible_and_emitted_on_lane(lane, point.frontier_point);
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::{
        broadcast_all, lane_publish_points_from_notifications, reactor_envelope, CommitFrameView,
        Notification, ReactorEnvelopeUnavailable,
    };
    use crate::coordinate::{Coordinate, DagPosition};
    use crate::event::{EventKind, HashChain};
    use crate::store::write::staging::{
        StagedCommitMeta, StagedCommitTiming, StagedCommittedEvent,
    };
    use crate::store::{EncodedBytes, ExtensionKey};
    use std::collections::BTreeMap;

    fn notification_for(lane: u32, sequence: u64, wall_ms: u64) -> Notification {
        Notification {
            event_id: crate::id::EventId::from_u128(0),
            correlation_id: 0,
            causation_id: None,
            coord: Coordinate::new("entity", "scope").expect("valid coordinate"),
            kind: EventKind::DATA,
            sequence,
            position: DagPosition::with_hlc(
                wall_ms,
                0,
                0,
                lane,
                u32::try_from(sequence).unwrap_or(u32::MAX),
            ),
        }
    }

    #[test]
    fn lane_publish_points_keep_first_on_equal_publish_up_to() {
        // Two notifications on the SAME lane with the SAME sequence (hence the
        // same `publish_up_to = sequence + 1`) but DIFFERENT wall_ms. The
        // `and_modify` only overwrites when the new `publish_up_to` is strictly
        // greater, so equal values must keep the FIRST point. The `> -> >=`
        // mutant would overwrite with the second (wall_ms = 222) instead.
        let notifications = vec![notification_for(7, 5, 111), notification_for(7, 5, 222)];

        let points = lane_publish_points_from_notifications(&notifications);
        let point = points.get(&7).expect("lane 7 must be present");

        assert_eq!(
            point.publish_up_to, 6,
            "PROPERTY: publish_up_to is sequence + 1"
        );
        assert_eq!(
            point.frontier_point.wall_ms, 111,
            "PROPERTY: equal publish_up_to must NOT overwrite the first lane point"
        );
    }

    #[test]
    fn lane_publish_points_advance_on_strictly_greater() {
        // Sanity companion: a strictly greater publish_up_to DOES overwrite, so
        // the test above is pinning the equality boundary, not blanket no-update.
        let notifications = vec![notification_for(3, 5, 111), notification_for(3, 9, 222)];

        let points = lane_publish_points_from_notifications(&notifications);
        let point = points.get(&3).expect("lane 3 must be present");

        assert_eq!(point.publish_up_to, 10);
        assert_eq!(point.frontier_point.wall_ms, 222);
    }

    fn staged_event() -> StagedCommittedEvent {
        StagedCommittedEvent::new(
            Coordinate::new("entity:react", "scope:react").expect("valid coordinate"),
            StagedCommitMeta::new(0xABCD, 1, None, EventKind::DATA, 7),
            StagedCommitTiming::new(1, 2, 3, 4, 5),
            HashChain {
                prev_hash: [0u8; 32],
                event_hash: [0u8; 32],
            },
        )
    }

    #[test]
    fn reactor_envelope_not_requested_when_no_subscribers() {
        // emit_envelope=false must short-circuit to NotRequested WITHOUT touching
        // the payload: even an undecodable payload yields NotRequested, proving
        // the envelope is only built when a reactor actually wants one.
        let staged = staged_event();
        let notification = notification_for(0, 1, 10);
        let ext: BTreeMap<ExtensionKey, EncodedBytes> = BTreeMap::new();
        let undecodable = [0xc1u8, 0xc1, 0xc1];
        let frame = CommitFrameView {
            payload_bytes: &undecodable,
            flags: 0,
            receipt_extensions: &ext,
            emit_envelope: false,
        };
        assert!(
            matches!(reactor_envelope(&staged, &notification, frame), Ok(None)),
            "no reactor subscriber must yield Ok(None)"
        );
    }

    #[test]
    fn reactor_envelope_available_for_decodable_payload() {
        // A reactor is subscribed and the committed payload decodes: the plaintext
        // envelope is built.
        let staged = staged_event();
        let notification = notification_for(0, 1, 10);
        let ext: BTreeMap<ExtensionKey, EncodedBytes> = BTreeMap::new();
        let payload =
            crate::encoding::to_bytes(&serde_json::json!({"k": 1})).expect("encode payload");
        let frame = CommitFrameView {
            payload_bytes: &payload,
            flags: 0,
            receipt_extensions: &ext,
            emit_envelope: true,
        };
        assert!(
            matches!(reactor_envelope(&staged, &notification, frame), Ok(Some(_))),
            "a decodable payload with a reactor subscribed must build an envelope"
        );
    }

    #[test]
    fn reactor_envelope_unavailable_is_surfaced_not_swallowed() {
        // REGRESSION (C3): the previous `.ok()` collapsed a `stored_event` decode
        // FAILURE and "no reactor asked" into the same silent `None`. With a
        // reactor subscribed (emit_envelope=true) but an undecodable committed
        // payload — as an encrypted append's ciphertext frame bytes are — the
        // outcome must be the distinct, observable Unavailable, NOT NotRequested
        // and NOT a swallowed Available.
        let staged = staged_event();
        let notification = notification_for(0, 1, 10);
        let ext: BTreeMap<ExtensionKey, EncodedBytes> = BTreeMap::new();
        let undecodable = [0xc1u8, 0xc1, 0xc1];
        let frame = CommitFrameView {
            payload_bytes: &undecodable,
            flags: 0,
            receipt_extensions: &ext,
            emit_envelope: true,
        };
        assert!(
            matches!(
                reactor_envelope(&staged, &notification, frame),
                Err(ReactorEnvelopeUnavailable)
            ),
            "an undecodable payload with a reactor subscribed must surface \
             ReactorEnvelopeUnavailable, not silently drop to Ok(None)"
        );
    }

    #[test]
    fn broadcast_all_counts_every_pushed_item() {
        let mut pushed = Vec::new();
        let count = broadcast_all([10, 20, 30], |item| pushed.push(*item));

        assert_eq!(
            count, 3,
            "PROPERTY: fanout telemetry count must advance once per pushed item"
        );
        assert_eq!(
            pushed,
            vec![10, 20, 30],
            "PROPERTY: count helper must still broadcast each item in order"
        );
    }
}