huskies/server/src/pipeline_state/events.rs

//! Event bus for pipeline state transitions.

use chrono::{DateTime, Utc};
use std::sync::OnceLock;
use tokio::sync::broadcast;

use super::{PipelineEvent, Stage, StoryId};

// ── Static transition broadcast channel ─────────────────────────────────────

static TRANSITION_TX: OnceLock<broadcast::Sender<TransitionFired>> = OnceLock::new();

fn get_or_init_tx() -> &'static broadcast::Sender<TransitionFired> {
    TRANSITION_TX.get_or_init(|| {
        let (tx, _) = broadcast::channel(256);
        tx
    })
}

/// Subscribe to all pipeline stage transitions.
///
/// Every call to [`apply_transition`][super::apply_transition] broadcasts the
/// resulting [`TransitionFired`] on this channel.  Returns a new receiver that
/// replays events from the moment of subscription.  Lagged receivers silently
/// skip missed events — callers should handle
/// [`broadcast::error::RecvError::Lagged`].
pub fn subscribe_transitions() -> broadcast::Receiver<TransitionFired> {
    get_or_init_tx().subscribe()
}

/// Broadcast `fired` to all active transition subscribers.
///
/// Called from [`apply_transition`][super::apply] after writing the new stage
/// to the CRDT.  No-ops safely when there are no subscribers.
pub(super) fn try_broadcast(fired: &TransitionFired) {
    let _ = get_or_init_tx().send(fired.clone());
}

/// Fired when a pipeline stage transition completes.
#[derive(Debug, Clone)]
pub struct TransitionFired {
    pub story_id: StoryId,
    pub before: Stage,
    pub after: Stage,
    pub event: PipelineEvent,
    pub at: DateTime<Utc>,
}

/// Trait for side-effect handlers that react to pipeline transitions.
pub trait TransitionSubscriber: Send + Sync {
    fn name(&self) -> &'static str;
    fn on_transition(&self, fired: &TransitionFired);
}

/// Collects [`TransitionSubscriber`]s and dispatches [`TransitionFired`] events to each.
pub struct EventBus {
    subscribers: Vec<Box<dyn TransitionSubscriber>>,
}

impl EventBus {
    /// Create an empty event bus with no subscribers.
    pub fn new() -> Self {
        Self {
            subscribers: Vec::new(),
        }
    }

    /// Register a subscriber to receive all future transition events.
    pub fn subscribe<S: TransitionSubscriber + 'static>(&mut self, subscriber: S) {
        self.subscribers.push(Box::new(subscriber));
    }

    /// Fire a transition event, calling every registered subscriber in order.
    pub fn fire(&self, event: TransitionFired) {
        for sub in &self.subscribers {
            sub.on_transition(&event);
        }
    }
}

impl Default for EventBus {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::super::{BranchName, PlanState};
    use super::*;
    use std::num::NonZeroU32;

    fn nz(n: u32) -> NonZeroU32 {
        NonZeroU32::new(n).unwrap()
    }
    fn fb(name: &str) -> BranchName {
        BranchName(name.to_string())
    }
    fn sid(s: &str) -> StoryId {
        StoryId(s.to_string())
    }

    #[test]
    fn event_bus_fires_to_all_subscribers() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicU32, Ordering};

        struct CountingSub(Arc<AtomicU32>);
        impl TransitionSubscriber for CountingSub {
            fn name(&self) -> &'static str {
                "counter"
            }
            fn on_transition(&self, _: &TransitionFired) {
                self.0.fetch_add(1, Ordering::SeqCst);
            }
        }

        let counter = Arc::new(AtomicU32::new(0));
        let mut bus = EventBus::new();
        bus.subscribe(CountingSub(counter.clone()));
        bus.subscribe(CountingSub(counter.clone()));

        bus.fire(TransitionFired {
            story_id: StoryId("test".into()),
            before: Stage::Backlog,
            after: Stage::Coding {
                claim: None,
                plan: PlanState::Missing,
                retries: 0,
            },
            event: PipelineEvent::DepsMet,
            at: Utc::now(),
        });

        assert_eq!(counter.load(Ordering::SeqCst), 2);
    }

    // ── Bug 502 regression: agent field is not part of Stage ────────────

    #[test]
    fn bug_502_agent_not_in_stage() {
        // Bug 502 was caused by a coder agent being assigned to a story in
        // Merge stage. In the typed system, Stage has no `agent` field at all.
        // Agent assignment is per-node ExecutionState. This test documents that
        // the old failure mode is structurally impossible.
        let merge = Stage::Merge {
            feature_branch: BranchName("feature/story-1".into()),
            commits_ahead: NonZeroU32::new(3).unwrap(),
            claim: None,
            retries: 0,
            server_start_time: None,
        };
        // Stage::Merge has exactly two fields: feature_branch and commits_ahead.
        // There is no way to attach an agent name to it. The type system
        // prevents bug 502 by construction.
        assert!(matches!(merge, Stage::Merge { .. }));
    }

    // ── TransitionError Display ─────────────────────────────────────────
}