Feat/indexer sse

crates/indexer/Cargo.toml

@@ -20,6 +20,7 @@ thiserror = "2"
 serde_json = "1"
 
 axum = "0.8.8"
+futures-util = { version = "0.3", default-features = false, features = ["std"] }
 reqwest = "0.13.1"
 tracing = "0.1"
 tracing-log = "0.2"

crates/indexer/README.md

@@ -186,6 +186,41 @@ cargo build -p lending-indexer --no-default-features
 > [!TIP]
 > When running the API **directly** on port 8000, use paths without the `/api` prefix (e.g. `http://localhost:8000/offers`). The OpenAPI `servers` entry uses `/api` for deployments where nginx proxies `/api/*` to the backend. Swagger UI on a direct run still lists `/api/...` in "Try it out" unless you select or override the server URL.
 
+### Server-Sent Events
+
+`GET /events` opens a long-lived SSE stream (`text/event-stream`).
+
+When the indexer commits on-chain updates, it issues Postgres `NOTIFY` on channel `lending_indexer_events` with a JSON `IndexerEvent` payload. The API process listens and fans events out to all SSE subscribers.
+
+Event types:
+
+| SSE `event:` | When emitted | Suggested client action |
+| :--- | :--- | :--- |
+| `block_indexed` | After each indexed block | Refetch lists using `not_expired`, overviews |
+| `factory_created` | New issuance factory indexed | Refetch `GET /factories/{id}` or `/factories/by-script` |
+| `offer_created` | New offer indexed (`pending`) | Refetch `GET /borrowers/offers` when `borrower_script_pubkey` matches; or `GET /offers/{id}` |
+| `offer_status_updated` | Offer status transition | Refetch offer details and role-specific lists |
+
+Examples:
+
+```
+event: block_indexed
+data: {"type":"block_indexed","height":2500001}
+
+event: factory_created
+data: {"type":"factory_created","id":"a1b2c3d4-e5f6-7890-abcd-ef1234567890","height":2500001,"factory_auth_script_pubkey":"52ac…"}
+
+event: offer_created
+data: {"type":"offer_created","id":"42","issuance_factory_id":"…","height":2500001,"created_at_txid":"aabb…","borrower_script_pubkey":"52ac…"}
+
+event: offer_status_updated
+data: {"type":"offer_status_updated","id":"42","status":"active","height":2500005}
+```
+
+Clients should treat events as signals to refetch REST resources rather than as full state snapshots. Keep-alive comments are sent periodically so proxies do not close idle connections. If an nginx (or similar) reverse proxy sits in front of the API, disable response buffering for this path.
+
+API and indexer may run as separate processes (`RUN_MODE=api` / `RUN_MODE=indexer`); events cross the process boundary via Postgres LISTEN/NOTIFY.
+
 ### Identifiers
 
 - **Offer `id`**: internal auto-increment integer (`BIGINT`), serialized in JSON responses as a **decimal string** (e.g. `"1"`). Assigned by PostgreSQL on insert. Used in nested `offer_id` fields and `GET /offers/by-script` responses. Not present on-chain. The path parameter in `GET /offers/{id}` is a numeric ID (e.g. `/offers/42`).

crates/indexer/src/api/events/bus.rs

@@ -0,0 +1,33 @@
+use tokio::sync::broadcast;
+
+use crate::events::IndexerEvent;
+
+const EVENT_BUS_CAPACITY: usize = 64;
+
+#[derive(Clone, Debug)]
+pub struct EventBus {
+    sender: broadcast::Sender<IndexerEvent>,
+}
+
+impl EventBus {
+    pub fn new() -> Self {
+        let (sender, _) = broadcast::channel(EVENT_BUS_CAPACITY);
+
+        Self { sender }
+    }
+
+    pub fn publish(&self, event: IndexerEvent) {
+        // No active subscribers is fine — `send` returns Err in that case.
+        let _ = self.sender.send(event);
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<IndexerEvent> {
+        self.sender.subscribe()
+    }
+}
+
+impl Default for EventBus {
+    fn default() -> Self {
+        Self::new()
+    }
+}

crates/indexer/src/api/events/handlers.rs

@@ -0,0 +1,61 @@
+use std::convert::Infallible;
+use std::sync::Arc;
+use std::time::Duration;
+
+use axum::extract::State;
+use axum::response::sse::{Event, KeepAlive, Sse};
+use futures_util::stream::{Stream, unfold};
+use tokio::sync::broadcast::error::RecvError;
+
+use crate::api::AppState;
+
+#[utoipa::path(
+    get,
+    path = "/events",
+    tag = "events",
+    operation_id = "subscribe_indexer_events",
+    responses(
+        (
+            status = 200,
+            description = "Server-Sent Events stream. Event types: `block_indexed`, `factory_created`, `offer_created`, `offer_status_updated`. Each `data` field carries a tagged JSON `IndexerEvent`. Clients should refetch REST resources on receipt.",
+            content_type = "text/event-stream"
+        ),
+    )
+)]
+#[tracing::instrument(name = "Subscribing to indexer events", skip(state))]
+pub async fn subscribe_events(
+    State(state): State<Arc<AppState>>,
+) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
+    let receiver = state.events.subscribe();
+
+    let stream = unfold(receiver, |mut receiver| async move {
+        loop {
+            match receiver.recv().await {
+                Ok(event) => {
+                    let sse_event = Event::default()
+                        .event(event.sse_event_name())
+                        .json_data(&event);
+
+                    match sse_event {
+                        Ok(sse_event) => return Some((Ok(sse_event), receiver)),
+                        Err(error) => {
+                            tracing::error!(?error, "Failed to serialize SSE event");
+                            continue;
+                        }
+                    }
+                }
+                Err(RecvError::Lagged(skipped)) => {
+                    tracing::warn!(skipped, "SSE subscriber lagged; continuing");
+                    continue;
+                }
+                Err(RecvError::Closed) => return None,
+            }
+        }
+    });
+
+    Sse::new(stream).keep_alive(
+        KeepAlive::new()
+            .interval(Duration::from_secs(15))
+            .text("keep-alive"),
+    )
+}

crates/indexer/src/api/events/listener.rs

@@ -0,0 +1,50 @@
+use std::time::Duration;
+
+use sqlx::PgPool;
+use sqlx::postgres::PgListener;
+
+use crate::api::events::EventBus;
+use crate::events::{INDEXER_EVENTS_CHANNEL, IndexerEvent};
+
+const RECONNECT_DELAY: Duration = Duration::from_secs(2);
+
+pub fn spawn_indexer_events_listener(db: PgPool, events: EventBus) {
+    tokio::spawn(async move {
+        loop {
+            if let Err(error) = listen_loop(&db, &events).await {
+                tracing::error!(
+                    ?error,
+                    "Indexer events LISTEN loop failed; reconnecting in {:?}",
+                    RECONNECT_DELAY
+                );
+                tokio::time::sleep(RECONNECT_DELAY).await;
+            }
+        }
+    });
+}
+
+async fn listen_loop(db: &PgPool, events: &EventBus) -> Result<(), sqlx::Error> {
+    let mut listener = PgListener::connect_with(db).await?;
+
+    listener.listen(INDEXER_EVENTS_CHANNEL).await?;
+    tracing::info!(
+        channel = INDEXER_EVENTS_CHANNEL,
+        "Listening for indexer event notifications"
+    );
+
+    loop {
+        let notification = listener.recv().await?;
+        match serde_json::from_str::<IndexerEvent>(notification.payload()) {
+            Ok(event) => {
+                events.publish(event);
+            }
+            Err(error) => {
+                tracing::warn!(
+                    payload = notification.payload(),
+                    ?error,
+                    "Ignoring invalid indexer event notification payload"
+                );
+            }
+        }
+    }
+}

crates/indexer/src/api/events/mod.rs

@@ -0,0 +1,8 @@
+mod bus;
+pub(crate) mod handlers;
+mod listener;
+mod routes;
+
+pub use bus::EventBus;
+pub use listener::spawn_indexer_events_listener;
+pub use routes::routes;

crates/indexer/src/api/events/routes.rs

@@ -0,0 +1,11 @@
+use std::sync::Arc;
+
+use axum::{Router, routing::get};
+
+use crate::api::AppState;
+
+use super::handlers;
+
+pub fn routes() -> Router<Arc<AppState>> {
+    Router::new().route("/events", get(handlers::subscribe_events))
+}

crates/indexer/src/api/mod.rs

@@ -2,6 +2,7 @@ mod borrowers;
 mod db;
 mod dto;
 mod error;
+mod events;
 mod factories;
 mod health;
 mod lenders;

crates/indexer/src/api/openapi/doc.rs

@@ -6,6 +6,7 @@ use utoipa_swagger_ui::{Config, SwaggerUi};
 use crate::api::borrowers::dto::BorrowerOverview;
 use crate::api::borrowers::handlers as borrower_handlers;
 use crate::api::dto::AssetAmount;
+use crate::api::events::handlers as event_handlers;
 use crate::api::factories::dto::{
     FactoryAuthUtxoDto, FactoryDetailsResponse, FactoryProgramUtxoDto,
 };
@@ -19,6 +20,7 @@ use crate::api::offers::dto::{
 };
 use crate::api::offers::handlers as offer_handlers;
 use crate::api::params::{OfferSortBy, SortDir};
+use crate::events::IndexerEvent;
 use crate::models::{FactoryStatus, OfferStatus, ParticipantType, UtxoType};
 
 use super::schemas::{ErrorBody, ErrorResponse, OfferDetailsResponseSchema};
@@ -44,6 +46,7 @@ use super::schemas::{ErrorBody, ErrorResponse, OfferDetailsResponseSchema};
         lender_handlers::list_offers_by_script,
         factory_handlers::get_by_script,
         factory_handlers::get_by_id,
+        event_handlers::subscribe_events,
         health::health,
         health::ready,
     ),
@@ -57,6 +60,7 @@ use super::schemas::{ErrorBody, ErrorResponse, OfferDetailsResponseSchema};
         FactoryProgramUtxoDto,
         FactoryStatus,
         HealthResponse,
+        IndexerEvent,
         LenderOverview,
         OfferDetailsResponseSchema,
         OfferListItemShort,
@@ -77,6 +81,7 @@ use super::schemas::{ErrorBody, ErrorResponse, OfferDetailsResponseSchema};
         (name = "borrowers", description = "Borrower queries"),
         (name = "lenders", description = "Lender queries"),
         (name = "factories", description = "Issuance factory queries"),
+        (name = "events", description = "Server-Sent Events for indexer updates"),
         (name = "health", description = "Liveness and readiness checks"),
     )
 )]
@@ -111,6 +116,7 @@ mod tests {
         assert!(paths.contains_key("/factories/{id}"));
         assert!(paths.contains_key("/health"));
         assert!(paths.contains_key("/ready"));
+        assert!(paths.contains_key("/events"));
     }
 
     #[test]

crates/indexer/src/api/server.rs

@@ -8,6 +8,7 @@ use tower_http::request_id::{self, MakeRequestUuid, RequestId};
 use tower_http::trace::TraceLayer;
 
 use crate::api::borrowers;
+use crate::api::events::{self, EventBus, spawn_indexer_events_listener};
 use crate::api::factories;
 use crate::api::health;
 use crate::api::lenders;
@@ -16,10 +17,17 @@ use crate::api::openapi;
 use crate::api::state::AppState;
 
 pub async fn run_server(listener: TcpListener, db_pool: PgPool) {
-    let state = Arc::new(AppState { db: db_pool });
+    let events = EventBus::new();
+    spawn_indexer_events_listener(db_pool.clone(), events.clone());
+
+    let state = Arc::new(AppState {
+        db: db_pool,
+        events,
+    });
 
     let app = Router::new()
         .merge(health::routes())
+        .merge(events::routes())
         .merge(borrowers::routes())
         .merge(lenders::routes())
         .merge(factories::routes())

crates/indexer/src/api/state.rs

@@ -1,5 +1,8 @@
 use sqlx::PgPool;
 
+use crate::api::events::EventBus;
+
 pub struct AppState {
     pub db: PgPool,
+    pub events: EventBus,
 }

crates/indexer/src/events/mod.rs

@@ -0,0 +1,5 @@
+mod notify;
+mod types;
+
+pub use notify::notify_indexer_event;
+pub use types::{INDEXER_EVENTS_CHANNEL, IndexerEvent};

crates/indexer/src/events/notify.rs

@@ -0,0 +1,30 @@
+use crate::db::DbTx;
+
+use super::types::{INDEXER_EVENTS_CHANNEL, IndexerEvent};
+
+#[tracing::instrument(name = "Notifying indexer event", skip(sql_tx, event))]
+pub async fn notify_indexer_event(
+    sql_tx: &mut DbTx<'_>,
+    event: &IndexerEvent,
+) -> Result<(), sqlx::Error> {
+    let payload = serde_json::to_string(event).map_err(|error| {
+        tracing::error!(?error, "Failed to serialize indexer event");
+        sqlx::Error::Encode(error.into())
+    })?;
+
+    sqlx::query("SELECT pg_notify($1, $2)")
+        .bind(INDEXER_EVENTS_CHANNEL)
+        .bind(payload)
+        .execute(&mut **sql_tx)
+        .await
+        .map_err(|error| {
+            tracing::error!(
+                ?error,
+                event_type = event.sse_event_name(),
+                "Failed to pg_notify"
+            );
+            error
+        })?;
+
+    Ok(())
+}