#![allow(clippy::type_complexity)] use std::marker::PhantomData; use std::num::TryFromIntError; use std::path::Path; use std::sync::Arc; use std::{fmt, io, str::FromStr, time}; use localtime::LocalTime; use sqlite as sql; use thiserror::Error; use crate::git; use crate::git::fmt::RefString; use crate::git::Oid; use crate::git::RefError; use crate::prelude::RepoId; use crate::sql::transaction; use crate::storage::RefUpdate; use super::{ Notification, NotificationId, NotificationKind, NotificationKindError, NotificationStatus, }; /// How long to wait for the database lock to be released before failing a read. const DB_READ_TIMEOUT: time::Duration = time::Duration::from_secs(3); /// How long to wait for the database lock to be released before failing a write. const DB_WRITE_TIMEOUT: time::Duration = time::Duration::from_secs(6); #[derive(Error, Debug)] pub enum Error { /// I/O error. #[error("i/o error: {0}")] Io(#[from] io::Error), /// An Internal error. #[error("internal error: {0}")] Internal(#[from] sql::Error), /// Timestamp error. #[error("invalid timestamp: {0}")] Timestamp(#[from] TryFromIntError), /// Invalid Git ref name. #[error("invalid ref name: {0}")] RefName(#[from] RefError), /// Invalid Git ref format. #[error("invalid ref format: {0}")] RefFormat(#[from] crate::git::fmt::Error), /// Invalid notification kind. #[error("invalid notification kind: {0}")] NotificationKind(#[from] NotificationKindError), /// Not found. #[error("notification {0} not found")] NotificationNotFound(NotificationId), /// Internal unit overflow. #[error("the unit overflowed")] UnitOverflow, } /// Read-only type witness. #[derive(Clone)] pub struct Read; /// Read-write type witness. #[derive(Clone)] pub struct Write; /// Notifications store. #[derive(Clone)] pub struct Store { db: Arc, marker: PhantomData, } impl fmt::Debug for Store { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "Store(..)") } } impl Store { const SCHEMA: &'static str = include_str!("schema.sql"); /// Same as [`Self::open`], but in read-only mode. This is useful to have multiple /// open databases, as no locking is required. pub fn reader>(path: P) -> Result { let mut db = sql::Connection::open_thread_safe_with_flags( path, sqlite::OpenFlags::new().with_read_only(), )?; db.set_busy_timeout(DB_READ_TIMEOUT.as_millis() as usize)?; db.execute(Self::SCHEMA)?; Ok(Self { db: Arc::new(db), marker: PhantomData, }) } /// Create a new in-memory address book. pub fn memory() -> Result { let db = sql::Connection::open_thread_safe_with_flags( ":memory:", sqlite::OpenFlags::new().with_read_only(), )?; db.execute(Self::SCHEMA)?; Ok(Self { db: Arc::new(db), marker: PhantomData, }) } } impl Store { const SCHEMA: &'static str = include_str!("schema.sql"); /// Open a policy store at the given path. Creates a new store if it /// doesn't exist. pub fn open>(path: P) -> Result { let mut db = sql::Connection::open_thread_safe(path)?; db.set_busy_timeout(DB_WRITE_TIMEOUT.as_millis() as usize)?; db.execute(Self::SCHEMA)?; Ok(Self { db: Arc::new(db), marker: PhantomData, }) } /// Create a new in-memory address book. pub fn memory() -> Result { let db = sql::Connection::open_thread_safe(":memory:")?; db.execute(Self::SCHEMA)?; Ok(Self { db: Arc::new(db), marker: PhantomData, }) } /// Get a read-only version of this store. pub fn read_only(self) -> Store { Store { db: self.db, marker: PhantomData, } } /// Set notification read status for the given notifications. pub fn set_status( &mut self, status: NotificationStatus, ids: &[NotificationId], ) -> Result { transaction(&self.db, |_| { let mut stmt = self.db.prepare( "UPDATE `repository-notifications` SET status = ?1 WHERE rowid = ?2", )?; for id in ids { stmt.bind((1, &status))?; stmt.bind((2, *id as i64))?; stmt.next()?; stmt.reset()?; } Ok(self.db.change_count() > 0) }) } /// Insert a notification. Resets the status to *unread* if it already exists. pub fn insert( &mut self, repo: &RepoId, update: &RefUpdate, timestamp: LocalTime, ) -> Result { let mut stmt = self.db.prepare( "INSERT INTO `repository-notifications` (repo, ref, old, new, timestamp) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT DO UPDATE SET old = ?3, new = ?4, timestamp = ?5, status = null", )?; let old = update.old().map(|o| o.to_string()); let new = update.new().map(|o| o.to_string()); stmt.bind((1, repo))?; stmt.bind((2, update.name().as_str()))?; stmt.bind((3, old.as_deref()))?; stmt.bind((4, new.as_deref()))?; stmt.bind((5, i64::try_from(timestamp.as_millis())?))?; stmt.next()?; Ok(self.db.change_count() > 0) } /// Delete the given notifications. pub fn clear(&mut self, ids: &[NotificationId]) -> Result { transaction(&self.db, |_| { let mut stmt = self .db .prepare("DELETE FROM `repository-notifications` WHERE rowid = ?")?; // N.b. we need to keep the count manually since the change count // will always be `1` because of each reset. let mut count = 0; for id in ids { stmt.bind((1, *id as i64))?; stmt.next()?; stmt.reset()?; count += self.db.change_count(); } Ok(count) }) } /// Delete all notifications of a repo. pub fn clear_by_repo(&mut self, repo: &RepoId) -> Result { let mut stmt = self .db .prepare("DELETE FROM `repository-notifications` WHERE repo = ?")?; stmt.bind((1, repo))?; stmt.next()?; Ok(self.db.change_count()) } /// Delete all notifications from all repos. pub fn clear_all(&mut self) -> Result { self.db .prepare("DELETE FROM `repository-notifications`")? .next()?; Ok(self.db.change_count()) } } /// `Read` methods for `Store`. This implies that a /// `Store` can access these functions as well. impl Store { /// Get a specific notification. pub fn get(&self, id: NotificationId) -> Result { let mut stmt = self.db.prepare( "SELECT rowid, repo, ref, old, new, status, timestamp FROM `repository-notifications` WHERE rowid = ?", )?; stmt.bind((1, id as i64))?; if let Some(Ok(row)) = stmt.into_iter().next() { return parse::notification(row); } Err(Error::NotificationNotFound(id)) } /// Get all notifications. pub fn all(&self) -> Result> + '_, Error> { let stmt = self.db.prepare( "SELECT rowid, repo, ref, old, new, status, timestamp FROM `repository-notifications` ORDER BY timestamp DESC", )?; Ok(stmt.into_iter().map(move |row| { let row = row?; parse::notification(row) })) } // Get notifications that were created between the given times: `since <= t < until`. pub fn by_timestamp( &self, since: LocalTime, until: LocalTime, ) -> Result> + '_, Error> { let mut stmt = self.db.prepare( "SELECT rowid, repo, ref, old, new, status, timestamp FROM `repository-notifications` WHERE timestamp >= ?1 AND timestamp < ?2 ORDER BY timestamp", )?; let since = i64::try_from(since.as_millis())?; let until = i64::try_from(until.as_millis())?; stmt.bind((1, since))?; stmt.bind((2, until))?; Ok(stmt.into_iter().map(move |row| { let row = row?; parse::notification(row) })) } /// Get notifications by repo. pub fn by_repo( &self, repo: &RepoId, order_by: &str, ) -> Result> + '_, Error> { let mut stmt = self.db.prepare(format!( "SELECT rowid, repo, ref, old, new, status, timestamp FROM `repository-notifications` WHERE repo = ? ORDER BY {order_by} DESC", ))?; stmt.bind((1, repo))?; Ok(stmt.into_iter().map(move |row| { let row = row?; parse::notification(row) })) } /// Get the total notification count. pub fn count(&self) -> Result { let stmt = self .db .prepare("SELECT COUNT(*) FROM `repository-notifications`")?; let count: i64 = stmt .into_iter() .next() .expect("COUNT will always return a single row")? .read(0); let count: usize = count.try_into().map_err(|_| Error::UnitOverflow)?; Ok(count) } /// Get the total notification count by repos. pub fn counts_by_repo( &self, ) -> Result> + '_, Error> { let stmt = self.db.prepare( "SELECT repo, COUNT(*) as count FROM `repository-notifications` GROUP BY repo", )?; Ok(stmt.into_iter().map(|row| { let row = row?; let count = row.try_read::("count")? as usize; let repo = row.try_read::("repo")?; Ok((repo, count)) })) } /// Get the notification count for the given repo. pub fn count_by_repo(&self, repo: &RepoId) -> Result { let mut stmt = self .db .prepare("SELECT COUNT(*) FROM `repository-notifications` WHERE repo = ?")?; stmt.bind((1, repo))?; let count: i64 = stmt .into_iter() .next() .expect("COUNT will always return a single row")? .read(0); let count: usize = count.try_into().map_err(|_| Error::UnitOverflow)?; Ok(count) } } mod parse { use super::*; pub fn notification(row: sql::Row) -> Result { let id = row.try_read::("rowid")? as NotificationId; let repo = row.try_read::("repo")?; let refstr = row.try_read::<&str, _>("ref")?; let status = row.try_read::("status")?; let old = row .try_read::, _>("old")? .map(|oid| { Oid::from_str(oid).map_err(|e| { Error::Internal(sql::Error { code: None, message: Some(format!("sql: invalid oid in `old` column: {oid:?}: {e}")), }) }) }) .unwrap_or(Ok(git::raw::Oid::zero().into()))?; let new = row .try_read::, _>("new")? .map(|oid| { Oid::from_str(oid).map_err(|e| { Error::Internal(sql::Error { code: None, message: Some(format!("sql: invalid oid in `new` column: {oid:?}: {e}")), }) }) }) .unwrap_or(Ok(git::raw::Oid::zero().into()))?; let update = RefUpdate::from(RefString::try_from(refstr)?, old, new); let (namespace, qualified) = git::parse_ref(refstr)?; let timestamp = row.try_read::("timestamp")?; let timestamp = LocalTime::from_millis(timestamp as u128); let qualified = qualified.to_owned(); let kind = NotificationKind::try_from(qualified.clone())?; Ok(Notification { id, repo, update, remote: namespace, qualified, status, kind, timestamp, }) } } #[cfg(test)] #[allow(clippy::unwrap_used)] mod test { use crate::git::fmt::{qualified, refname}; use crate::{cob, node::NodeId, test::arbitrary}; use super::*; #[test] fn test_clear() { let mut db = Store::open(":memory:").unwrap(); let repo = arbitrary::gen::(1); let old = arbitrary::oid(); let time = LocalTime::from_millis(32188142); let master = arbitrary::oid(); for i in 0..3 { let update = RefUpdate::Updated { name: format!("refs/heads/feature/{i}").try_into().unwrap(), old, new: master, }; assert!(db.insert(&repo, &update, time).unwrap()); } assert_eq!(db.count().unwrap(), 3); assert_eq!(db.count_by_repo(&repo).unwrap(), 3); db.clear_by_repo(&repo).unwrap(); assert_eq!(db.count().unwrap(), 0); assert_eq!(db.count_by_repo(&repo).unwrap(), 0); } #[test] fn test_counts_by_repo() { let mut db = Store::open(":memory:").unwrap(); let repo1 = arbitrary::gen::(1); let repo2 = arbitrary::gen::(1); let oid = arbitrary::oid(); let time = LocalTime::from_millis(32188142); let update1 = RefUpdate::Created { name: refname!("refs/heads/feature/1"), oid, }; let update2 = RefUpdate::Created { name: refname!("refs/heads/feature/2"), oid, }; let update3 = RefUpdate::Created { name: refname!("refs/heads/feature/3"), oid, }; assert!(db.insert(&repo1, &update1, time).unwrap()); assert!(db.insert(&repo1, &update2, time).unwrap()); assert!(db.insert(&repo2, &update3, time).unwrap()); let counts = db .counts_by_repo() .unwrap() .collect::, _>>() .unwrap(); assert_eq!(counts.get(&repo1).unwrap(), &2); assert_eq!(counts.get(&repo2).unwrap(), &1); } #[test] fn test_branch_notifications() { let repo = arbitrary::gen::(1); let old = arbitrary::oid(); let master = arbitrary::oid(); let other = arbitrary::oid(); let time1 = LocalTime::from_millis(32188142); let time2 = LocalTime::from_millis(32189874); let time3 = LocalTime::from_millis(32189879); let mut db = Store::open(":memory:").unwrap(); let update1 = RefUpdate::Updated { name: refname!("refs/heads/master"), old, new: master, }; let update2 = RefUpdate::Created { name: refname!("refs/heads/other"), oid: other, }; let update3 = RefUpdate::Deleted { name: refname!("refs/heads/dev"), oid: other, }; assert!(db.insert(&repo, &update1, time1).unwrap()); assert!(db.insert(&repo, &update2, time2).unwrap()); assert!(db.insert(&repo, &update3, time3).unwrap()); let mut notifs = db.by_repo(&repo, "timestamp").unwrap(); assert_eq!( notifs.next().unwrap().unwrap(), Notification { id: 3, repo, remote: None, qualified: qualified!("refs/heads/dev"), update: update3, kind: NotificationKind::Branch { name: refname!("dev") }, status: NotificationStatus::Unread, timestamp: time3, } ); assert_eq!( notifs.next().unwrap().unwrap(), Notification { id: 2, repo, remote: None, qualified: qualified!("refs/heads/other"), update: update2, kind: NotificationKind::Branch { name: refname!("other") }, status: NotificationStatus::Unread, timestamp: time2, } ); assert_eq!( notifs.next().unwrap().unwrap(), Notification { id: 1, repo, remote: None, qualified: qualified!("refs/heads/master"), update: update1, kind: NotificationKind::Branch { name: refname!("master") }, status: NotificationStatus::Unread, timestamp: time1, } ); assert!(notifs.next().is_none()); } #[test] fn test_notification_status() { let repo = arbitrary::gen::(1); let oid = arbitrary::oid(); let time = LocalTime::from_millis(32188142); let mut db = Store::open(":memory:").unwrap(); let update1 = RefUpdate::Created { name: refname!("refs/heads/feature/1"), oid, }; let update2 = RefUpdate::Created { name: refname!("refs/heads/feature/2"), oid, }; let update3 = RefUpdate::Created { name: refname!("refs/heads/feature/3"), oid, }; assert!(db.insert(&repo, &update1, time).unwrap()); assert!(db.insert(&repo, &update2, time).unwrap()); assert!(db.insert(&repo, &update3, time).unwrap()); assert!(db .set_status(NotificationStatus::ReadAt(time), &[1, 2, 3]) .unwrap()); let mut notifs = db.by_repo(&repo, "timestamp").unwrap(); assert_eq!( notifs.next().unwrap().unwrap().status, NotificationStatus::ReadAt(time), ); assert_eq!( notifs.next().unwrap().unwrap().status, NotificationStatus::ReadAt(time), ); assert_eq!( notifs.next().unwrap().unwrap().status, NotificationStatus::ReadAt(time), ); } #[test] fn test_duplicate_notifications() { let repo = arbitrary::gen::(1); let old = arbitrary::oid(); let master1 = arbitrary::oid(); let master2 = arbitrary::oid(); let time1 = LocalTime::from_millis(32188142); let time2 = LocalTime::from_millis(32189874); let mut db = Store::open(":memory:").unwrap(); let update1 = RefUpdate::Updated { name: refname!("refs/heads/master"), old, new: master1, }; let update2 = RefUpdate::Updated { name: refname!("refs/heads/master"), old: master1, new: master2, }; assert!(db.insert(&repo, &update1, time1).unwrap()); assert!(db .set_status(NotificationStatus::ReadAt(time1), &[1]) .unwrap()); assert!(db.insert(&repo, &update2, time2).unwrap()); let mut notifs = db.by_repo(&repo, "timestamp").unwrap(); assert_eq!( notifs.next().unwrap().unwrap(), Notification { id: 1, repo, remote: None, qualified: qualified!("refs/heads/master"), update: update2, kind: NotificationKind::Branch { name: refname!("master") }, // Status is reset to "unread". status: NotificationStatus::Unread, timestamp: time2, } ); assert!(notifs.next().is_none()); } #[test] fn test_cob_notifications() { let repo = arbitrary::gen::(1); let old = arbitrary::oid(); let new = arbitrary::oid(); let timestamp = LocalTime::from_millis(32189874); let nid: NodeId = "z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi" .parse() .unwrap(); let mut db = Store::open(":memory:").unwrap(); let qualified = qualified!("refs/cobs/xyz.radicle.issue/d87dcfe8c2b3200e78b128d9b959cfdf7063fefe"); let namespaced = qualified.with_namespace((&nid).into()); let update = RefUpdate::Updated { name: namespaced.to_ref_string(), old, new, }; assert!(db.insert(&repo, &update, timestamp).unwrap()); let mut notifs = db.by_repo(&repo, "timestamp").unwrap(); assert_eq!( notifs.next().unwrap().unwrap(), Notification { id: 1, repo, remote: Some( "z6MknSLrJoTcukLrE435hVNQT4JUhbvWLX4kUzqkEStBU8Vi" .parse() .unwrap() ), qualified, update, kind: NotificationKind::Cob { typed_id: cob::TypedId { type_name: cob::issue::TYPENAME.clone(), id: "d87dcfe8c2b3200e78b128d9b959cfdf7063fefe".parse().unwrap(), }, }, status: NotificationStatus::Unread, timestamp, } ); assert!(notifs.next().is_none()); } }