ai_crew_sync/
events.rs

1//! In-process event hub fed by Postgres LISTEN/NOTIFY.
2//!
3//! One background task holds a single LISTEN connection on `bus_events` and
4//! fans every payload out to in-process subscribers through a tokio broadcast
5//! channel. Consumers: the `wait_for_updates` tool (long-poll) and the webhook
6//! dispatcher. Payloads carry ids only; consumers resolve names against the
7//! database when they need them.
8
9use sqlx::{PgPool, postgres::PgListener};
10use tokio::sync::broadcast;
11use tokio_util::sync::CancellationToken;
12use uuid::Uuid;
13
14pub const PG_CHANNEL: &str = "bus_events";
15
16/// A parsed NOTIFY payload. Kept as loose JSON plus typed accessors so adding
17/// fields to the triggers never breaks older consumers.
18#[derive(Clone, Debug)]
19pub struct BusEvent(pub serde_json::Value);
20
21impl BusEvent {
22    pub fn kind(&self) -> &str {
23        self.0.get("kind").and_then(|v| v.as_str()).unwrap_or("")
24    }
25
26    fn uuid_field(&self, key: &str) -> Option<Uuid> {
27        self.0
28            .get(key)
29            .and_then(|v| v.as_str())
30            .and_then(|s| Uuid::parse_str(s).ok())
31    }
32
33    pub fn team_id(&self) -> Option<Uuid> {
34        self.uuid_field("team_id")
35    }
36
37    pub fn recipient_agent_id(&self) -> Option<Uuid> {
38        self.uuid_field("recipient_agent_id")
39    }
40
41    pub fn sender_agent_id(&self) -> Option<Uuid> {
42        self.uuid_field("sender_agent_id")
43    }
44
45    pub fn channel_id(&self) -> Option<Uuid> {
46        self.uuid_field("channel_id")
47    }
48
49    pub fn message_id(&self) -> Option<i64> {
50        self.0.get("id").and_then(|v| v.as_i64())
51    }
52
53    pub fn is_direct_message(&self) -> bool {
54        self.kind() == "message" && self.recipient_agent_id().is_some()
55    }
56
57    /// Is this event visible to `agent` of `team`? Direct messages are only
58    /// visible to their recipient (and sender); everything else is team-wide.
59    pub fn visible_to(&self, team_id: Uuid, agent_id: Uuid) -> bool {
60        if self.team_id() != Some(team_id) {
61            return false;
62        }
63        if self.is_direct_message() {
64            return self.recipient_agent_id() == Some(agent_id)
65                || self.sender_agent_id() == Some(agent_id);
66        }
67        true
68    }
69}
70
71#[derive(Clone)]
72pub struct EventHub {
73    tx: broadcast::Sender<BusEvent>,
74}
75
76impl Default for EventHub {
77    fn default() -> Self {
78        Self::new()
79    }
80}
81
82impl EventHub {
83    pub fn new() -> Self {
84        // 256 in-flight events is plenty; laggards get Lagged and resync from
85        // the database, which every consumer does anyway.
86        let (tx, _) = broadcast::channel(256);
87        Self { tx }
88    }
89
90    pub fn subscribe(&self) -> broadcast::Receiver<BusEvent> {
91        self.tx.subscribe()
92    }
93
94    pub fn publish(&self, event: BusEvent) {
95        // No receivers is fine: nobody is waiting right now.
96        let _ = self.tx.send(event);
97    }
98}
99
100/// Run the LISTEN loop until cancelled. Reconnects with backoff on failure so
101/// a Postgres restart degrades to polling latency instead of killing wakeups.
102pub async fn run_pg_listener(pool: PgPool, hub: EventHub, ct: CancellationToken) {
103    loop {
104        if ct.is_cancelled() {
105            return;
106        }
107        match PgListener::connect_with(&pool).await {
108            Ok(mut listener) => {
109                if let Err(e) = listener.listen(PG_CHANNEL).await {
110                    tracing::warn!(error = %e, "LISTEN failed; retrying");
111                } else {
112                    tracing::info!("event listener attached to '{PG_CHANNEL}'");
113                    loop {
114                        tokio::select! {
115                            _ = ct.cancelled() => return,
116                            recv = listener.try_recv() => match recv {
117                                Ok(Some(notification)) => {
118                                    match serde_json::from_str(notification.payload()) {
119                                        Ok(value) => hub.publish(BusEvent(value)),
120                                        Err(e) => tracing::warn!(
121                                            error = %e,
122                                            payload = notification.payload(),
123                                            "unparseable bus event"
124                                        ),
125                                    }
126                                }
127                                // None = connection dropped and was re-established;
128                                // notifications in between are lost, which consumers
129                                // tolerate by re-checking the database.
130                                Ok(None) => tracing::debug!("event listener reconnected"),
131                                Err(e) => {
132                                    tracing::warn!(error = %e, "event listener error");
133                                    break;
134                                }
135                            }
136                        }
137                    }
138                }
139            }
140            Err(e) => {
141                tracing::warn!(error = %e, "could not attach event listener; retrying");
142            }
143        }
144        tokio::select! {
145            _ = ct.cancelled() => return,
146            _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => {}
147        }
148    }
149}
ai_crew_sync/events.rs

ai_crew_sync/
events.rs
