use std::collections::HashMap; use std::sync::{Arc, RwLock}; use uuid::Uuid; use crate::models::{ Goal, ImportProfile, ImportSession, Movie, MovieProfile, ProfileField, RefreshSession, RemoteWatchlistEntry, Review, User, UserSettings, WatchEvent, WatchlistEntry, WebhookToken, }; use crate::value_objects::SocialIdentity; /// One `processed_events` row. pub(super) struct LedgerRow { pub(super) done: bool, pub(super) started_at: chrono::NaiveDateTime, pub(super) processed_at: chrono::NaiveDateTime, } #[derive(Debug, Clone, PartialEq, Eq)] pub(super) enum FollowState { Pending, Accepted, } /// Tables whose rows the schema removes when a user row is deleted /// (`ON DELETE CASCADE` on a column referencing `users(id)`), and which /// `InMemoryStore` models. The sqlite adapter's cascade test asserts the same list. pub const USER_CASCADE_TABLES: &[&str] = &[ "watchlist_entries", "goals", "user_settings", "user_profile_fields", "webhook_tokens", "watch_events", "ap_followers", "ap_following", "blocked_actors", ]; /// Tables that cascade from a user but whose aggregate this store does not hold /// (they belong to the federation crate). pub const USER_CASCADE_TABLES_OUTSIDE_STORE: &[&str] = &["ap_local_actors"]; /// Tables whose rows the schema removes when a movie row is deleted. pub const MOVIE_CASCADE_TABLES: &[&str] = &[ "watchlist_entries", "movie_profiles", "movie_genres", "movie_keywords", "movie_cast", "movie_crew", ]; /// One in-memory implementation of every persistence port, mirroring the /// relational constraints of the sqlite schema (cascades, uniqueness, FK /// restrictions). Share one `Arc` across all port slots. pub struct InMemoryStore { pub(super) movies: RwLock>, pub(super) movie_profiles: RwLock>, pub(super) reviews: RwLock>, pub(super) users: RwLock>, /// Ids of `users` rows carrying a `deleted_at` tombstone. pub(super) tombstoned: RwLock>, /// Lower-cased `retired_usernames` (the column is NOCASE). pub(super) retired_usernames: RwLock>, pub(super) profile_fields: RwLock>>, pub(super) user_settings: RwLock>, pub(super) refresh_sessions: RwLock>, pub(super) watchlist: RwLock>, pub(super) remote_watchlist: RwLock>, pub(super) goals: RwLock>, pub(super) watch_events: RwLock>, pub(super) webhook_tokens: RwLock>, pub(super) import_sessions: RwLock>, pub(super) import_profiles: RwLock>, pub(super) follows: RwLock>, pub(super) blocked: RwLock>, pub(super) processed_events: RwLock>, pub(super) recorded_events: RwLock>, } impl InMemoryStore { pub fn new() -> Arc { Arc::new(Self { movies: RwLock::default(), movie_profiles: RwLock::default(), reviews: RwLock::default(), users: RwLock::default(), tombstoned: RwLock::default(), retired_usernames: RwLock::default(), profile_fields: RwLock::default(), user_settings: RwLock::default(), refresh_sessions: RwLock::default(), watchlist: RwLock::default(), remote_watchlist: RwLock::default(), goals: RwLock::default(), watch_events: RwLock::default(), webhook_tokens: RwLock::default(), import_sessions: RwLock::default(), import_profiles: RwLock::default(), follows: RwLock::default(), blocked: RwLock::default(), processed_events: RwLock::default(), recorded_events: RwLock::default(), }) } pub fn with_remote_watchlist(entries: Vec) -> Arc { let store = Self::new(); *store.remote_watchlist.write().unwrap() = entries; store } /// Events recorded atomically with a change (the outbox rows the sqlite /// adapter would have written in the same transaction), in write order. pub fn recorded_events(&self) -> Vec { self.recorded_events.read().unwrap().clone() } pub(super) fn record_events(&self, events: &[crate::events::DomainEvent]) { self.recorded_events .write() .unwrap() .extend_from_slice(events); } pub fn movie_count(&self) -> usize { self.movies.read().unwrap().len() } pub fn movie_profile_count(&self) -> usize { self.movie_profiles.read().unwrap().len() } pub fn review_count(&self) -> usize { self.reviews.read().unwrap().len() } pub fn user_count(&self) -> usize { self.users.read().unwrap().len() } pub fn watchlist_count(&self) -> usize { self.watchlist.read().unwrap().len() } pub fn goal_count(&self) -> usize { self.goals.read().unwrap().len() } pub fn import_session_count(&self) -> usize { self.import_sessions.read().unwrap().len() } pub fn processed_event_count(&self) -> usize { self.processed_events.read().unwrap().len() } /// Inserts a ledger row with explicit timestamps (retention and lease tests). pub fn seed_processed_event( &self, id: &crate::events::EventId, handler: &str, done: bool, at: chrono::NaiveDateTime, ) { self.processed_events.write().unwrap().insert( (id.value(), handler.to_owned()), LedgerRow { done, started_at: at, processed_at: at, }, ); } pub fn refresh_session_count(&self) -> usize { self.refresh_sessions.read().unwrap().len() } /// Inserts a user bypassing the uniqueness checks of `UserRepository::save`. pub fn seed_user(&self, user: User) { self.users.write().unwrap().insert(user.id().value(), user); } /// Row count of the sqlite table of the same name, for the tables this /// store models. Panics on a table the store does not hold. pub fn rows_in(&self, table: &str) -> usize { match table { "watchlist_entries" => self.watchlist_count(), "goals" => self.goal_count(), "user_settings" => self.user_settings.read().unwrap().len(), "user_profile_fields" => self .profile_fields .read() .unwrap() .values() .map(Vec::len) .sum(), "webhook_tokens" => self.webhook_tokens.read().unwrap().len(), "watch_events" => self.watch_events.read().unwrap().len(), "ap_following" => self.follows.read().unwrap().len(), "ap_followers" => self .follows .read() .unwrap() .iter() .filter(|(_, target, _)| matches!(target, SocialIdentity::Local(_))) .count(), "blocked_actors" => self.blocked.read().unwrap().len(), "movie_profiles" => self.movie_profile_count(), "movie_genres" => self.sum_profiles(|p| p.genres.len()), "movie_keywords" => self.sum_profiles(|p| p.keywords.len()), "movie_cast" => self.sum_profiles(|p| p.cast.len()), "movie_crew" => self.sum_profiles(|p| p.crew.len()), other => panic!("InMemoryStore does not model table {other}"), } } fn sum_profiles(&self, f: impl Fn(&MovieProfile) -> usize) -> usize { self.movie_profiles.read().unwrap().values().map(f).sum() } }