neiam /fieri-forever

//! Live updates for outings, over SSE.
//!
//! A change to an outing is announced with Postgres `NOTIFY outing_changed,
//! '<outing id>'`. Every replica `LISTEN`s and rebroadcasts ids on an
//! in-process channel, which each open SSE stream filters for its outing. The
//! browser reacts by re-fetching the outing panel, so the event carries no
//! data — it only says "look again".
//!
//! Going through Postgres rather than straight onto the channel is what makes
//! this work with more than one pod: a vote handled by one replica reaches
//! people whose stream is held by another.
use std::time::Duration;

use sea_orm::sqlx::postgres::{PgListener, PgPool};
use sea_orm::{ConnectionTrait, DatabaseConnection, DbBackend, DbErr, Statement};
use tokio::sync::broadcast;
use tracing::{info, warn};
use uuid::Uuid;

const CHANNEL: &str = "outing_changed";

#[derive(Clone)]
pub struct Live {
    tx: broadcast::Sender<Uuid>,
}

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

impl Live {
    pub fn new() -> Self {
        // Lagging receivers just miss ids; the 30 s poll on the page covers it.
        let (tx, _) = broadcast::channel(256);
        Self { tx }
    }

    pub fn subscribe(&self) -> broadcast::Receiver<Uuid> {
        self.tx.subscribe()
    }

    /// Relay `NOTIFY`s onto the channel for as long as the process runs,
    /// reconnecting when the listener's connection drops.
    pub fn spawn_listener(&self, pool: PgPool) {
        let tx = self.tx.clone();
        tokio::spawn(async move {
            loop {
                match listen(&pool, &tx).await {
                    Ok(()) => {}
                    Err(e) => warn!(error = %e, "outing listener failed; retrying"),
                }
                tokio::time::sleep(Duration::from_secs(2)).await;
            }
        });
    }
}

async fn listen(pool: &PgPool, tx: &broadcast::Sender<Uuid>) -> Result<(), sea_orm::sqlx::Error> {
    let mut listener = PgListener::connect_with(pool).await?;
    listener.listen(CHANNEL).await?;
    info!(channel = CHANNEL, "listening for outing changes");
    loop {
        let n = listener.recv().await?;
        if let Ok(id) = n.payload().parse() {
            // No subscribers is not an error: nobody's watching right now.
            let _ = tx.send(id);
        }
    }
}

/// Tell every replica's watchers that this outing changed.
pub async fn notify(db: &impl ConnectionTrait, outing_id: Uuid) -> Result<(), DbErr> {
    db.execute(Statement::from_sql_and_values(
        DbBackend::Postgres,
        "SELECT pg_notify($1, $2)",
        [CHANNEL.into(), outing_id.to_string().into()],
    ))
    .await?;
    Ok(())
}

pub fn pool(db: &DatabaseConnection) -> PgPool {
    db.get_postgres_connection_pool().clone()
}