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::sync::atomic::{AtomicBool, Ordering};
21use std::time::Duration;
22
23use sqlx::PgPool;
24use sqlx::postgres::PgListener;
25use tokio::task::JoinHandle;
26use tracing::{debug, error, info, warn};
27
28use super::repository::EventOutboxRepository;
29use super::routing::{EventRouter, OUTBOX_CHANNEL, OutboxChannel};
30use systemprompt_identifiers::{EventOutboxId, UserId};
31use systemprompt_models::{A2AEvent, AgUiEvent, AnalyticsEvent, SystemEvent};
32
33const OUTBOX_RETENTION: Duration = Duration::from_secs(3600);
34const PRUNE_INTERVAL: Duration = Duration::from_secs(300);
35const RETRY_MIN: Duration = Duration::from_secs(1);
36const RETRY_MAX: Duration = Duration::from_secs(60);
37
38// Why: Postgres `read_only_sql_transaction`, raised by `LISTEN` on a standby.
39const READ_ONLY_SQL_TRANSACTION: &str = "25006";
40
41static LISTENING: AtomicBool = AtomicBool::new(true);
42
43#[must_use]
44pub fn is_listening() -> bool {
45    LISTENING.load(Ordering::Relaxed)
46}
47
48fn is_read_only_standby(err: &sqlx::Error) -> bool {
49    match err {
50        sqlx::Error::Database(db) => db.code().as_deref() == Some(READ_ONLY_SQL_TRANSACTION),
51        _ => false,
52    }
53}
54
55#[derive(Debug, Clone)]
56pub struct PostgresEventBridge {
57    pool: PgPool,
58    outbox: EventOutboxRepository,
59}
60
61impl PostgresEventBridge {
62    #[must_use]
63    pub fn new(pool: PgPool) -> Self {
64        Self {
65            outbox: EventOutboxRepository::new(pool.clone()),
66            pool,
67        }
68    }
69
70    pub fn start(self) -> JoinHandle<()> {
71        EventRouter::install_relay(self.pool.clone());
72        tokio::spawn(async move {
73            self.run().await;
74        })
75    }
76
77    async fn open_listener(&self) -> Result<PgListener, sqlx::Error> {
78        let mut listener = PgListener::connect_with(&self.pool).await?;
79        listener.listen(OUTBOX_CHANNEL).await?;
80        Ok(listener)
81    }
82
83    fn report_listener_failure(err: &sqlx::Error, retry_in: Duration) {
84        let retry_in_secs = retry_in.as_secs();
85        if is_read_only_standby(err) {
86            error!(
87                error = %err,
88                channel = OUTBOX_CHANNEL,
89                retry_in_secs,
90                "event bridge: pool is a read-only standby; LISTEN/NOTIFY requires the primary. \
91                 Point the write pool (`database_write_url` secret, or `DATABASE_WRITE_URL` with \
92                 the env secrets source) at the primary and restart"
93            );
94        } else {
95            error!(
96                error = %err,
97                channel = OUTBOX_CHANNEL,
98                retry_in_secs,
99                "event bridge: failed to open Postgres listener; retrying"
100            );
101        }
102    }
103
104    async fn run(self) {
105        let mut prune_tick = tokio::time::interval(PRUNE_INTERVAL);
106        prune_tick.tick().await;
107        let mut backoff = RETRY_MIN;
108
109        loop {
110            let mut listener = match self.open_listener().await {
111                Ok(listener) => listener,
112                Err(e) => {
113                    LISTENING.store(false, Ordering::Relaxed);
114                    Self::report_listener_failure(&e, backoff);
115                    tokio::time::sleep(backoff).await;
116                    backoff = (backoff * 2).min(RETRY_MAX);
117                    continue;
118                },
119            };
120            backoff = RETRY_MIN;
121            LISTENING.store(true, Ordering::Relaxed);
122            info!(
123                channel = OUTBOX_CHANNEL,
124                "event bridge: listening for cross-replica events"
125            );
126
127            loop {
128                tokio::select! {
129                    notification = listener.recv() => match notification {
130                        Ok(notification) => {
131                            self.deliver(notification.payload()).await;
132                        },
133                        Err(e) => {
134                            warn!(error = %e, "event bridge: listener connection lost; reconnecting");
135                            break;
136                        },
137                    },
138                    _ = prune_tick.tick() => {
139                        self.prune().await;
140                    },
141                }
142            }
143        }
144    }
145
146    async fn deliver(&self, row_id: &str) {
147        let id = EventOutboxId::new(row_id);
148        let row = match self.outbox.find(&id).await {
149            Ok(Some(row)) => row,
150            Ok(None) => {
151                debug!(row_id, "event bridge: outbox row already pruned; skipping");
152                return;
153            },
154            Err(e) => {
155                error!(error = %e, row_id, "event bridge: failed to load outbox row");
156                return;
157            },
158        };
159
160        let Some(channel) = OutboxChannel::parse(&row.channel) else {
161            error!(channel = %row.channel, row_id, "event bridge: unknown outbox channel");
162            return;
163        };
164        Self::fan_in(channel, &row.user_id, row.payload).await;
165    }
166
167    pub(super) async fn fan_in(
168        channel: OutboxChannel,
169        user_id: &UserId,
170        // JSON: outbox payload is polymorphic by channel; decoded into the
171        // matching typed event immediately below.
172        payload: serde_json::Value,
173    ) {
174        match channel {
175            OutboxChannel::AgUi => match serde_json::from_value::<AgUiEvent>(payload) {
176                Ok(event) => {
177                    EventRouter::route_agui_local(user_id, event).await;
178                },
179                Err(e) => error!(error = %e, "event bridge: failed to decode AG-UI event"),
180            },
181            OutboxChannel::A2A => match serde_json::from_value::<A2AEvent>(payload) {
182                Ok(event) => {
183                    EventRouter::route_a2a_local(user_id, event).await;
184                },
185                Err(e) => error!(error = %e, "event bridge: failed to decode A2A event"),
186            },
187            OutboxChannel::System => match serde_json::from_value::<SystemEvent>(payload) {
188                Ok(event) => {
189                    EventRouter::route_system_local(user_id, event).await;
190                },
191                Err(e) => error!(error = %e, "event bridge: failed to decode system event"),
192            },
193            OutboxChannel::Analytics => match serde_json::from_value::<AnalyticsEvent>(payload) {
194                Ok(event) => {
195                    EventRouter::route_analytics_local(user_id, event).await;
196                },
197                Err(e) => error!(error = %e, "event bridge: failed to decode analytics event"),
198            },
199        }
200    }
201
202    async fn prune(&self) {
203        let cutoff = chrono::Utc::now()
204            - chrono::Duration::from_std(OUTBOX_RETENTION)
205                .unwrap_or_else(|_| chrono::Duration::seconds(3600));
206        match self.outbox.prune(cutoff).await {
207            Ok(deleted) => {
208                if deleted > 0 {
209                    debug!(deleted, "event bridge: pruned expired outbox rows");
210                }
211            },
212            Err(e) => error!(error = %e, "event bridge: outbox prune failed"),
213        }
214    }
215}
systemprompt_events/services/bridge.rs

systemprompt_events/services/
bridge.rs
