systemprompt_events/services/
bridge.rs

1//! Cross-replica event relay over Postgres `LISTEN`/`NOTIFY`.
2//!
3//! In a multi-replica deployment the in-process [`crate::EventRouter`]
4//! broadcasters only reach SSE connections held by the current process.
5//! [`PostgresEventBridge`] closes that gap: every replica runs one bridge
6//! task that `LISTEN`s on [`OUTBOX_CHANNEL`]. When any replica routes an
7//! event it appends a row to `event_outbox` and emits a `NOTIFY` carrying
8//! that row's id. Each bridge receives the notification, loads the row,
9//! deserializes the payload by its `channel`, and re-injects the event
10//! through the router's *local-only* path — which deliberately does **not**
11//! touch the outbox, so the relay cannot loop.
12//!
13//! The notification payload is only the row id (a UUID string) to stay
14//! well under Postgres' ~8 KB `NOTIFY` limit; the event body lives in the
15//! `jsonb` column.
16//!
17//! Copyright (c) systemprompt.io — Business Source License 1.1.
18//! See <https://systemprompt.io> for licensing details.
19
20use std::time::Duration;
21
22use sqlx::PgPool;
23use sqlx::postgres::PgListener;
24use tokio::task::JoinHandle;
25use tracing::{debug, error, info, warn};
26
27use super::repository::EventOutboxRepository;
28use super::routing::{EventRouter, OUTBOX_CHANNEL, OutboxChannel};
29use systemprompt_identifiers::{EventOutboxId, UserId};
30use systemprompt_models::{A2AEvent, AgUiEvent, AnalyticsEvent, SystemEvent};
31
32const OUTBOX_RETENTION: Duration = Duration::from_secs(3600);
33const PRUNE_INTERVAL: Duration = Duration::from_secs(300);
34
35#[derive(Debug, Clone)]
36pub struct PostgresEventBridge {
37    pool: PgPool,
38}
39
40impl PostgresEventBridge {
41    #[must_use]
42    pub const fn new(pool: PgPool) -> Self {
43        Self { pool }
44    }
45
46    /// Abort the returned handle to stop the relay.
47    pub fn start(self) -> JoinHandle<()> {
48        EventRouter::install_relay(self.pool.clone());
49        tokio::spawn(async move {
50            self.run().await;
51        })
52    }
53
54    async fn run(self) {
55        let mut prune_tick = tokio::time::interval(PRUNE_INTERVAL);
56        prune_tick.tick().await;
57
58        loop {
59            let mut listener = match PgListener::connect_with(&self.pool).await {
60                Ok(listener) => listener,
61                Err(e) => {
62                    error!(error = %e, "event bridge: failed to open Postgres listener; retrying");
63                    tokio::time::sleep(Duration::from_secs(5)).await;
64                    continue;
65                },
66            };
67            if let Err(e) = listener.listen(OUTBOX_CHANNEL).await {
68                error!(error = %e, channel = OUTBOX_CHANNEL, "event bridge: LISTEN failed; retrying");
69                tokio::time::sleep(Duration::from_secs(5)).await;
70                continue;
71            }
72            info!(
73                channel = OUTBOX_CHANNEL,
74                "event bridge: listening for cross-replica events"
75            );
76
77            loop {
78                tokio::select! {
79                    notification = listener.recv() => match notification {
80                        Ok(notification) => {
81                            self.deliver(notification.payload()).await;
82                        },
83                        Err(e) => {
84                            warn!(error = %e, "event bridge: listener connection lost; reconnecting");
85                            break;
86                        },
87                    },
88                    _ = prune_tick.tick() => {
89                        self.prune().await;
90                    },
91                }
92            }
93        }
94    }
95
96    async fn deliver(&self, row_id: &str) {
97        let repo = EventOutboxRepository::new(self.pool.clone());
98        let id = EventOutboxId::new(row_id);
99        let row = match repo.find(&id).await {
100            Ok(Some(row)) => row,
101            Ok(None) => {
102                debug!(row_id, "event bridge: outbox row already pruned; skipping");
103                return;
104            },
105            Err(e) => {
106                error!(error = %e, row_id, "event bridge: failed to load outbox row");
107                return;
108            },
109        };
110
111        let Some(channel) = OutboxChannel::parse(&row.channel) else {
112            error!(channel = %row.channel, row_id, "event bridge: unknown outbox channel");
113            return;
114        };
115        Self::fan_in(channel, &row.user_id, row.payload).await;
116    }
117
118    pub(super) async fn fan_in(
119        channel: OutboxChannel,
120        user_id: &UserId,
121        // JSON: outbox payload is polymorphic by channel; decoded into the
122        // matching typed event immediately below.
123        payload: serde_json::Value,
124    ) {
125        match channel {
126            OutboxChannel::AgUi => match serde_json::from_value::<AgUiEvent>(payload) {
127                Ok(event) => {
128                    EventRouter::route_agui_local(user_id, event).await;
129                },
130                Err(e) => error!(error = %e, "event bridge: failed to decode AG-UI event"),
131            },
132            OutboxChannel::A2A => match serde_json::from_value::<A2AEvent>(payload) {
133                Ok(event) => {
134                    EventRouter::route_a2a_local(user_id, event).await;
135                },
136                Err(e) => error!(error = %e, "event bridge: failed to decode A2A event"),
137            },
138            OutboxChannel::System => match serde_json::from_value::<SystemEvent>(payload) {
139                Ok(event) => {
140                    EventRouter::route_system_local(user_id, event).await;
141                },
142                Err(e) => error!(error = %e, "event bridge: failed to decode system event"),
143            },
144            OutboxChannel::Analytics => match serde_json::from_value::<AnalyticsEvent>(payload) {
145                Ok(event) => {
146                    EventRouter::route_analytics_local(user_id, event).await;
147                },
148                Err(e) => error!(error = %e, "event bridge: failed to decode analytics event"),
149            },
150        }
151    }
152
153    async fn prune(&self) {
154        let cutoff = chrono::Utc::now()
155            - chrono::Duration::from_std(OUTBOX_RETENTION)
156                .unwrap_or_else(|_| chrono::Duration::seconds(3600));
157        let repo = EventOutboxRepository::new(self.pool.clone());
158        match repo.prune(cutoff).await {
159            Ok(deleted) => {
160                if deleted > 0 {
161                    debug!(deleted, "event bridge: pruned expired outbox rows");
162                }
163            },
164            Err(e) => error!(error = %e, "event bridge: outbox prune failed"),
165        }
166    }
167}
systemprompt_events/services/bridge.rs

systemprompt_events/services/
bridge.rs
