Something went wrong. Try again.
Streamplace AtmosphereConf 2026 VOD JAM
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518use sqlx::{FromRow, SqlitePool, sqlite::SqliteConnectOptions, sqlite::SqlitePoolOptions};use std::sync::OnceLock;use tokio::sync::OnceCell;
/// Direction of a fetch cursor for Bluesky firehose pagination.#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum FetchDirection { Forward, Reverse,}
impl FetchDirection { fn as_str(self) -> &'static str { match self { FetchDirection::Forward => "forward", FetchDirection::Reverse => "reverse", } }}
/// A cursor position for fetching chat messages from the firehose.#[derive(Debug, Clone)]pub struct FetchCursor { pub cursor: Option<String>, pub boundary_ms: i64,}
/// A chat message stored in the local SQLite cache.#[derive(Debug, Clone, FromRow)]pub struct CachedChatMessage { pub streamer_did: String, pub author_did: String, pub rkey: String, pub created_at_ms: i64, pub text: String, pub reply_parent_uri: Option<String>, pub reply_root_uri: Option<String>, pub facets_json: Option<String>,}
/// SQLite-backed cache for chat messages and fetch cursors.pub struct ChatCache { pool: SqlitePool, schema_init: OnceCell<()>,}
impl ChatCache { /// Open (or create) a SQLite database at `path`. /// /// The connection is established lazily; schema creation is deferred until /// the first operation via [`ensure_schema`]. pub fn open(path: &str) -> Result<Self, sqlx::Error> { let options = SqliteConnectOptions::new() .filename(path) .create_if_missing(true) .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
let pool = SqlitePoolOptions::new().connect_lazy_with(options);
Ok(Self { pool, schema_init: OnceCell::new(), }) }
/// Ensure the database schema has been created. Called before every /// operation so that the first caller initialises it exactly once. async fn ensure_schema(&self) -> Result<(), sqlx::Error> { self.schema_init .get_or_try_init(|| async { sqlx::query( r#" CREATE TABLE IF NOT EXISTS messages ( streamer_did TEXT NOT NULL, author_did TEXT NOT NULL, rkey TEXT NOT NULL, created_at_ms INTEGER NOT NULL, text TEXT NOT NULL, reply_parent_uri TEXT, reply_root_uri TEXT, facets_json TEXT, PRIMARY KEY (author_did, rkey) ) "#, ) .execute(&self.pool) .await?;
sqlx::query( r#" CREATE INDEX IF NOT EXISTS messages_streamer_time ON messages (streamer_did, created_at_ms) "#, ) .execute(&self.pool) .await?;
sqlx::query( r#" CREATE TABLE IF NOT EXISTS fetch_cursors ( streamer_did TEXT NOT NULL, direction TEXT NOT NULL CHECK(direction IN ('forward', 'reverse')), cursor TEXT, boundary_ms INTEGER NOT NULL, PRIMARY KEY (streamer_did, direction) ) "#, ) .execute(&self.pool) .await?;
sqlx::query( r#" CREATE TABLE IF NOT EXISTS video_catalog ( id INTEGER PRIMARY KEY DEFAULT 1, data_json TEXT NOT NULL, updated_at INTEGER NOT NULL ) "#, ) .execute(&self.pool) .await?;
Ok(()) }) .await .copied() }
/// Bulk-insert messages, silently ignoring duplicates (same author_did + rkey). pub async fn store_messages(&self, messages: &[CachedChatMessage]) -> Result<(), sqlx::Error> { self.ensure_schema().await?;
let mut tx = self.pool.begin().await?; for msg in messages { sqlx::query( r#" INSERT OR IGNORE INTO messages (streamer_did, author_did, rkey, created_at_ms, text, reply_parent_uri, reply_root_uri, facets_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?) "#, ) .bind(&msg.streamer_did) .bind(&msg.author_did) .bind(&msg.rkey) .bind(msg.created_at_ms) .bind(&msg.text) .bind(&msg.reply_parent_uri) .bind(&msg.reply_root_uri) .bind(&msg.facets_json) .execute(&mut *tx) .await?; } tx.commit().await?; Ok(()) }
/// Retrieve all messages for `streamer_did` whose timestamp falls within /// `[start_ms, end_ms]`, ordered chronologically. pub async fn get_messages( &self, streamer_did: &str, start_ms: i64, end_ms: i64, ) -> Result<Vec<CachedChatMessage>, sqlx::Error> { self.ensure_schema().await?;
sqlx::query_as::<_, CachedChatMessage>( r#" SELECT streamer_did, author_did, rkey, created_at_ms, text, reply_parent_uri, reply_root_uri, facets_json FROM messages WHERE streamer_did = ? AND created_at_ms BETWEEN ? AND ? ORDER BY created_at_ms ASC "#, ) .bind(streamer_did) .bind(start_ms) .bind(end_ms) .fetch_all(&self.pool) .await }
/// Retrieve the stored fetch cursor for `(streamer_did, direction)`. pub async fn get_fetch_cursor( &self, streamer_did: &str, direction: FetchDirection, ) -> Result<Option<FetchCursor>, sqlx::Error> { self.ensure_schema().await?;
let row: Option<(Option<String>, i64)> = sqlx::query_as( r#" SELECT cursor, boundary_ms FROM fetch_cursors WHERE streamer_did = ? AND direction = ? "#, ) .bind(streamer_did) .bind(direction.as_str()) .fetch_optional(&self.pool) .await?;
Ok(row.map(|(cursor, boundary_ms)| FetchCursor { cursor, boundary_ms, })) }
/// Upsert the fetch cursor for `(streamer_did, direction)`. pub async fn set_fetch_cursor( &self, streamer_did: &str, direction: FetchDirection, cursor: Option<&str>, boundary_ms: i64, ) -> Result<(), sqlx::Error> { self.ensure_schema().await?;
sqlx::query( r#" INSERT INTO fetch_cursors (streamer_did, direction, cursor, boundary_ms) VALUES (?, ?, ?, ?) ON CONFLICT(streamer_did, direction) DO UPDATE SET cursor = excluded.cursor, boundary_ms = excluded.boundary_ms "#, ) .bind(streamer_did) .bind(direction.as_str()) .bind(cursor) .bind(boundary_ms) .execute(&self.pool) .await?;
Ok(()) }
/// Store the full video catalog as a JSON blob, replacing any previous entry. pub async fn store_video_catalog<T: serde::Serialize>( &self, videos: &[T], ) -> Result<(), sqlx::Error> { self.ensure_schema().await?;
let json = serde_json::to_string(videos).map_err(|e| sqlx::Error::Protocol(e.to_string()))?; let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs() as i64;
sqlx::query( r#" INSERT INTO video_catalog (id, data_json, updated_at) VALUES (1, ?, ?) ON CONFLICT(id) DO UPDATE SET data_json = excluded.data_json, updated_at = excluded.updated_at "#, ) .bind(&json) .bind(now) .execute(&self.pool) .await?;
Ok(()) }
/// Load the cached video catalog, if any. Returns `(videos, updated_at_unix_secs)`. pub async fn load_video_catalog<T: serde::de::DeserializeOwned>( &self, ) -> Result<Option<(Vec<T>, i64)>, sqlx::Error> { self.ensure_schema().await?;
let row: Option<(String, i64)> = sqlx::query_as( r#" SELECT data_json, updated_at FROM video_catalog WHERE id = 1 "#, ) .fetch_optional(&self.pool) .await?;
match row { Some((json, updated_at)) => { let videos: Vec<T> = serde_json::from_str(&json) .map_err(|e| sqlx::Error::Protocol(e.to_string()))?; Ok(Some((videos, updated_at))) } None => Ok(None), } }}
// ---------------------------------------------------------------------------// Global accessor// ---------------------------------------------------------------------------
static CACHE: OnceLock<ChatCache> = OnceLock::new();
/// Return a reference to the process-wide [`ChatCache`].////// Panics on first call if the database cannot be opened (e.g. bad path or/// permissions).pub(crate) fn shared_cache() -> &'static ChatCache { CACHE.get_or_init(|| { let dir = std::env::var("VODPLACE_CACHE_DIR").unwrap_or_else(|_| "./data".to_string()); std::fs::create_dir_all(&dir).expect("failed to create cache directory"); let path = format!("{dir}/cache.db"); ChatCache::open(&path).expect("failed to open chat message cache database") })}
// ---------------------------------------------------------------------------// Tests// ---------------------------------------------------------------------------
#[cfg(test)]mod tests { use super::*;
/// Create an in-memory SQLite chat cache for tests. /// /// Uses `max_connections(1)` with an eager connect so all operations share /// the same connection and see the same in-memory database. A lazy pool would /// open separate connections, each with an empty DB, causing schema-not-found /// failures after `ensure_schema` runs on one connection. async fn test_cache() -> ChatCache { let pool = SqlitePoolOptions::new() .max_connections(1) .connect("sqlite::memory:") .await .expect("failed to open in-memory test database"); let cache = ChatCache { pool, schema_init: OnceCell::new(), }; cache.ensure_schema().await.expect("schema init failed"); cache }
fn test_message(rkey: &str, created_at_ms: i64) -> CachedChatMessage { CachedChatMessage { streamer_did: "did:plc:streamer1".to_string(), author_did: "did:plc:author1".to_string(), rkey: rkey.to_string(), created_at_ms, text: format!("message at {created_at_ms}"), reply_parent_uri: None, reply_root_uri: None, facets_json: None, } }
#[tokio::test] async fn store_and_retrieve_messages_same_window() { let cache = test_cache().await;
let messages = vec![ test_message("rkey1", 1000), test_message("rkey2", 2000), test_message("rkey3", 3000), ];
cache.store_messages(&messages).await.expect("store failed");
let result = cache .get_messages("did:plc:streamer1", 1000, 3000) .await .expect("get failed");
assert_eq!(result.len(), 3); assert_eq!(result[0].rkey, "rkey1"); assert_eq!(result[1].rkey, "rkey2"); assert_eq!(result[2].rkey, "rkey3"); }
#[tokio::test] async fn query_subset_time_window() { let cache = test_cache().await;
let messages = vec![ test_message("rkey1", 1000), test_message("rkey2", 2000), test_message("rkey3", 3000), test_message("rkey4", 4000), ];
cache.store_messages(&messages).await.expect("store failed");
// Ask for only the middle two let result = cache .get_messages("did:plc:streamer1", 2000, 3000) .await .expect("get failed");
assert_eq!(result.len(), 2); assert_eq!(result[0].rkey, "rkey2"); assert_eq!(result[1].rkey, "rkey3"); }
#[tokio::test] async fn set_and_get_forward_cursor() { let cache = test_cache().await;
cache .set_fetch_cursor( "did:plc:streamer1", FetchDirection::Forward, Some("cursor-abc"), 5000, ) .await .expect("set failed");
let got = cache .get_fetch_cursor("did:plc:streamer1", FetchDirection::Forward) .await .expect("get failed") .expect("cursor not found");
assert_eq!(got.cursor.as_deref(), Some("cursor-abc")); assert_eq!(got.boundary_ms, 5000); }
#[tokio::test] async fn set_and_get_reverse_cursor() { let cache = test_cache().await;
cache .set_fetch_cursor( "did:plc:streamer1", FetchDirection::Reverse, Some("cursor-xyz"), 1000, ) .await .expect("set failed");
let got = cache .get_fetch_cursor("did:plc:streamer1", FetchDirection::Reverse) .await .expect("get failed") .expect("cursor not found");
assert_eq!(got.cursor.as_deref(), Some("cursor-xyz")); assert_eq!(got.boundary_ms, 1000); }
#[tokio::test] async fn forward_and_reverse_cursors_are_independent() { let cache = test_cache().await;
cache .set_fetch_cursor( "did:plc:streamer1", FetchDirection::Forward, Some("fwd-cursor"), 9000, ) .await .expect("set forward failed");
cache .set_fetch_cursor( "did:plc:streamer1", FetchDirection::Reverse, Some("rev-cursor"), 1000, ) .await .expect("set reverse failed");
let fwd = cache .get_fetch_cursor("did:plc:streamer1", FetchDirection::Forward) .await .expect("get forward failed") .expect("forward cursor not found");
let rev = cache .get_fetch_cursor("did:plc:streamer1", FetchDirection::Reverse) .await .expect("get reverse failed") .expect("reverse cursor not found");
assert_eq!(fwd.cursor.as_deref(), Some("fwd-cursor")); assert_eq!(fwd.boundary_ms, 9000); assert_eq!(rev.cursor.as_deref(), Some("rev-cursor")); assert_eq!(rev.boundary_ms, 1000); }
#[tokio::test] async fn duplicate_messages_insert_or_ignore() { let cache = test_cache().await;
let messages = vec![test_message("rkey1", 1000)]; cache .store_messages(&messages) .await .expect("first store failed");
// Store the same message again — must not error and must not duplicate cache .store_messages(&messages) .await .expect("second store failed");
let result = cache .get_messages("did:plc:streamer1", 1000, 1000) .await .expect("get failed");
assert_eq!(result.len(), 1, "duplicate message was not ignored"); }}