use chrono::Utc; use error_stack::ResultExt as _; use sqlx::sqlite::SqlitePool; use sqlx::Row; use tracing::debug; use crate::errors::{Error, Result}; /// Sentinel password hash for OIDC-provisioned users. It is deliberately not a /// valid PHC string, so `password_auth::verify_password` always rejects it and /// the account is reachable only via SSO. const OIDC_NO_PASSWORD: &str = "!oidc:no-password-login"; /// Application database for storing Kobo sync state, reading progress, /// shelves, and authentication tokens. #[derive(Debug)] pub struct AppDb { pool: SqlitePool, } /// A user record. #[derive(Debug, Clone)] pub struct UserRecord { pub id: i64, pub username: String, pub password_hash: String, /// OIDC `sub` claim this account is linked to, if any. `None` for /// password-only users. pub oidc_sub: Option, } /// A reading state record. #[derive(Debug, Clone)] pub struct ReadingStateRecord { pub id: i64, pub user_id: i64, pub book_id: i64, pub last_modified: String, pub priority_timestamp: String, /// Device token that last wrote this state, so its own progress is never /// echoed back to it on the next sync. pub last_writer_token_id: Option, } /// A bookmark record. #[derive(Debug, Clone)] pub struct BookmarkRecord { pub id: i64, pub reading_state_id: i64, pub last_modified: String, pub location_source: Option, pub location_type: Option, pub location_value: Option, pub progress_percent: Option, pub content_source_progress_percent: Option, } /// A statistics record. #[derive(Debug, Clone)] pub struct StatisticsRecord { pub id: i64, pub reading_state_id: i64, pub last_modified: String, pub remaining_time_minutes: Option, pub spent_reading_minutes: Option, } /// A read status record. #[derive(Debug, Clone)] pub struct ReadBookRecord { pub id: i64, pub book_id: i64, pub user_id: i64, pub read_status: i32, pub last_modified: String, pub last_time_started_reading: Option, pub times_started_reading: i32, } /// An archived book record. #[derive(Debug, Clone)] pub struct ArchivedBookRecord { pub id: i64, pub user_id: i64, pub book_id: i64, pub is_archived: bool, pub last_modified: String, } /// A shelf record. #[derive(Debug, Clone)] pub struct ShelfRecord { pub id: i64, pub uuid: String, pub name: String, pub user_id: i64, pub kobo_sync: bool, pub created: String, pub last_modified: String, } /// An auth token record. #[derive(Debug, Clone)] pub struct AuthTokenRecord { pub id: i64, pub auth_token: String, pub user_id: i64, /// Serial of a device bound to this token via `device_registrations`, if /// any. Only populated by `list_auth_tokens`; other lookups leave it `None`. pub device_serial: Option, } /// A sync point — a frozen snapshot of a user's library state at a moment in time. #[derive(Debug, Clone)] pub struct SyncPointRow { pub id: String, pub user_id: i64, pub auth_token_id: i64, pub created_at: String, } /// A book row inside a sync point snapshot. #[derive(Debug, Clone)] pub struct SyncPointBookRow { pub sync_point_id: String, pub book_id: i64, pub book_uuid: String, pub book_last_modified: String, pub book_timestamp: String, pub file_format: String, pub file_size: i64, pub reading_state_last_modified: Option, pub is_archived: bool, } /// A shelf row inside a sync point snapshot. #[derive(Debug, Clone)] pub struct SyncPointShelfRow { pub sync_point_id: String, pub shelf_id: i64, pub shelf_uuid: String, pub shelf_name: String, pub created: String, pub last_modified: String, } /// Input for inserting a `sync_point_books` row during snapshot creation. #[derive(Debug, Clone)] pub struct SyncPointBookInsert { pub book_id: i64, pub book_uuid: String, pub book_last_modified: String, pub book_timestamp: String, pub file_format: String, pub file_size: i64, pub reading_state_last_modified: Option, pub is_archived: bool, } impl AppDb { pub async fn connect(db_path: &std::path::Path) -> Result { let url = format!("sqlite://{}?mode=rwc", db_path.display()); let pool = SqlitePool::connect(&url) .await .change_context(Error::Database) .attach_with(|| format!("Failed to connect to app DB at {}", db_path.display()))?; let db = Self { pool }; db.run_migrations().await?; Ok(db) } /// Returns a reference to the underlying connection pool. pub fn pool(&self) -> &SqlitePool { &self.pool } /// Create all required tables if they don't exist. /// SQLite only executes one statement per `execute()`, so we run them individually. async fn run_migrations(&self) -> Result<()> { let statements = [ r#"CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL )"#, r#"CREATE TABLE IF NOT EXISTS auth_tokens ( id INTEGER PRIMARY KEY AUTOINCREMENT, auth_token TEXT NOT NULL UNIQUE, user_id INTEGER NOT NULL )"#, r#"CREATE TABLE IF NOT EXISTS reading_states ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, book_id INTEGER NOT NULL, last_modified TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), priority_timestamp TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), UNIQUE(user_id, book_id) )"#, r#"CREATE TABLE IF NOT EXISTS bookmarks ( id INTEGER PRIMARY KEY AUTOINCREMENT, reading_state_id INTEGER NOT NULL REFERENCES reading_states(id) ON DELETE CASCADE, last_modified TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), location_source TEXT, location_type TEXT, location_value TEXT, progress_percent REAL, content_source_progress_percent REAL, UNIQUE(reading_state_id) )"#, r#"CREATE TABLE IF NOT EXISTS statistics ( id INTEGER PRIMARY KEY AUTOINCREMENT, reading_state_id INTEGER NOT NULL REFERENCES reading_states(id) ON DELETE CASCADE, last_modified TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), remaining_time_minutes INTEGER, spent_reading_minutes INTEGER, UNIQUE(reading_state_id) )"#, r#"CREATE TABLE IF NOT EXISTS read_books ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id INTEGER NOT NULL, user_id INTEGER NOT NULL, read_status INTEGER NOT NULL DEFAULT 0, last_modified TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), last_time_started_reading TEXT, times_started_reading INTEGER NOT NULL DEFAULT 0, UNIQUE(user_id, book_id) )"#, r#"CREATE TABLE IF NOT EXISTS archived_books ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, book_id INTEGER NOT NULL, is_archived BOOLEAN NOT NULL DEFAULT 0, last_modified TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), UNIQUE(user_id, book_id) )"#, r#"CREATE TABLE IF NOT EXISTS shelves ( id INTEGER PRIMARY KEY AUTOINCREMENT, uuid TEXT NOT NULL UNIQUE, name TEXT NOT NULL, user_id INTEGER NOT NULL, kobo_sync BOOLEAN NOT NULL DEFAULT 0, created TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), last_modified TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) )"#, r#"CREATE TABLE IF NOT EXISTS shelf_books ( id INTEGER PRIMARY KEY AUTOINCREMENT, shelf_id INTEGER NOT NULL REFERENCES shelves(id) ON DELETE CASCADE, book_id INTEGER NOT NULL, date_added TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), UNIQUE(shelf_id, book_id) )"#, r#"CREATE TABLE IF NOT EXISTS sync_points ( id TEXT PRIMARY KEY, user_id INTEGER NOT NULL, auth_token_id INTEGER NOT NULL, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')) )"#, r#"CREATE INDEX IF NOT EXISTS idx_sync_points_user_token ON sync_points(user_id, auth_token_id)"#, r#"CREATE INDEX IF NOT EXISTS idx_sync_points_created ON sync_points(created_at)"#, r#"CREATE TABLE IF NOT EXISTS sync_point_books ( sync_point_id TEXT NOT NULL REFERENCES sync_points(id) ON DELETE CASCADE, book_id INTEGER NOT NULL, book_uuid TEXT NOT NULL, book_last_modified TEXT NOT NULL, book_timestamp TEXT NOT NULL, file_format TEXT NOT NULL, file_size INTEGER NOT NULL, reading_state_last_modified TEXT, is_archived INTEGER NOT NULL DEFAULT 0, synced INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (sync_point_id, book_id) )"#, r#"CREATE INDEX IF NOT EXISTS idx_spb_synced ON sync_point_books(sync_point_id, synced)"#, r#"CREATE TABLE IF NOT EXISTS sync_point_shelves ( sync_point_id TEXT NOT NULL REFERENCES sync_points(id) ON DELETE CASCADE, shelf_id INTEGER NOT NULL, shelf_uuid TEXT NOT NULL, shelf_name TEXT NOT NULL, created TEXT NOT NULL, last_modified TEXT NOT NULL, synced INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (sync_point_id, shelf_id) )"#, r#"CREATE TABLE IF NOT EXISTS sync_point_shelf_books ( sync_point_id TEXT NOT NULL, shelf_id INTEGER NOT NULL, book_uuid TEXT NOT NULL, PRIMARY KEY (sync_point_id, shelf_id, book_uuid), FOREIGN KEY (sync_point_id, shelf_id) REFERENCES sync_point_shelves(sync_point_id, shelf_id) ON DELETE CASCADE )"#, // Drop legacy tables superseded by the snapshot model. r#"DROP TABLE IF EXISTS synced_books"#, r#"DROP TABLE IF EXISTS shelf_archives"#, // Maps a device serial number to its auth token so the token-free // `device_auth` endpoint (which receives the serial but no Bearer) // can return a JWT carrying the right token. Bound on first sync by // claiming a pending registration (below). r#"CREATE TABLE IF NOT EXISTS device_registrations ( serial TEXT NOT NULL PRIMARY KEY, auth_token TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) )"#, // A token "awaiting a device". Setup inserts one; the next // `device_auth` with an unbound serial claims it, binding // serial → auth_token in `device_registrations`. r#"CREATE TABLE IF NOT EXISTS pending_registrations ( auth_token TEXT NOT NULL PRIMARY KEY, user_id INTEGER NOT NULL, created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) )"#, // The unified book catalog: the single source of truth for book // existence and metadata. Rows are ingested from a Calibre library // (source='calibre', id = Calibre books.id preserved) or created by // the upload portal (source='upload', id >= UPLOAD_ID_BASE). Every // per-user table keys off `book_id`, which references `books.id`. r#"CREATE TABLE IF NOT EXISTS books ( id INTEGER PRIMARY KEY, uuid TEXT NOT NULL UNIQUE, title TEXT NOT NULL, authors TEXT NOT NULL DEFAULT '[]', publisher TEXT, language TEXT, description TEXT, series_name TEXT, series_index REAL, pubdate TEXT, file_format TEXT NOT NULL, file_size INTEGER NOT NULL, source TEXT NOT NULL, locator TEXT NOT NULL, original_filename TEXT, content_hash TEXT, has_cover INTEGER NOT NULL DEFAULT 0, owner_user_id INTEGER, timestamp TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), last_modified TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) )"#, r#"CREATE INDEX IF NOT EXISTS idx_books_source ON books(source)"#, r#"CREATE INDEX IF NOT EXISTS idx_books_content_hash ON books(owner_user_id, content_hash)"#, // Global server settings as a simple key/value store. Values are // TEXT; booleans are stored as "1"/"0". An absent key means "use the // built-in default", so upgrades preserve prior behavior. r#"CREATE TABLE IF NOT EXISTS settings ( key TEXT NOT NULL PRIMARY KEY, value TEXT NOT NULL )"#, // Per-user override of a book's sync eligibility. A present row wins // over `books.default_sync`; absence falls back to that default. r#"CREATE TABLE IF NOT EXISTS book_sync_prefs ( user_id INTEGER NOT NULL, book_id INTEGER NOT NULL, enabled INTEGER NOT NULL, PRIMARY KEY (user_id, book_id) )"#, // Override cover for a book, stored on disk as {cover_dir}/{book_id}.jpg. // A present row means "this book has an override cover"; `version` is // bumped on each replacement to bust the browser and device caches. // Kept in its own table (not on the `books` row) so it survives the // startup Calibre re-ingest, which deletes and reinserts calibre rows. r#"CREATE TABLE IF NOT EXISTS book_covers ( book_id INTEGER NOT NULL PRIMARY KEY, version INTEGER NOT NULL )"#, // Per-book metadata overrides. A non-NULL column replaces the value on // the `books` row when read (COALESCE at query time); NULL falls back // to the source value. Kept in its own table (not on the `books` row) // so edits survive the startup Calibre re-ingest, which deletes and // reinserts calibre rows. `authors` is a JSON array (same encoding as // `books.authors`). r#"CREATE TABLE IF NOT EXISTS book_meta_overrides ( book_id INTEGER NOT NULL PRIMARY KEY, title TEXT, authors TEXT, series_name TEXT, series_index REAL, description TEXT )"#, // Soft-deleted books. A present row hides the book from the catalog // and from every user's sync (removed from the device on next sync), // without dropping the `books` row or its files. Kept in its own // table (keyed by the preserved `books.id`) so it survives the // startup Calibre re-ingest — the only way to keep a Calibre book // hidden, since its row is rebuilt from the read-only metadata.db. r#"CREATE TABLE IF NOT EXISTS deleted_books ( book_id INTEGER NOT NULL PRIMARY KEY, deleted_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) )"#, // Server-side sync cursor, one row per device token. The firmware // keeps the sync token in `user.SyncContinuationToken` and clears it // between syncs, so a device routinely arrives with no token at all; // without a server-held baseline every sync degrades to a full sync. r#"CREATE TABLE IF NOT EXISTS device_sync_state ( auth_token_id INTEGER NOT NULL PRIMARY KEY, last_sync_point_id TEXT, ongoing_sync_point_id TEXT, updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) )"#, ]; for sql in &statements { sqlx::query(*sql) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to run migration")?; } // Link OIDC identities to web users. Added out-of-band because SQLite's // `ALTER TABLE ADD COLUMN` has no `IF NOT EXISTS` and this runner // re-executes on every boot. self.add_column_if_missing("users", "oidc_sub", "TEXT") .await?; sqlx::query( "CREATE UNIQUE INDEX IF NOT EXISTS idx_users_oidc_sub \ ON users(oidc_sub) WHERE oidc_sub IS NOT NULL", ) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to create oidc_sub index")?; // A book's default sync eligibility, applied to any user without an // explicit `book_sync_prefs` row. `DEFAULT 1` keeps every existing book // syncable across the upgrade. self.add_column_if_missing("books", "default_sync", "INTEGER NOT NULL DEFAULT 1") .await?; // Stable per-book identifier from the uploaded EPUB's `urn:uuid:` // `` (Calibre writes one). Used to dedupe re-uploads of // the same book: Calibre re-exports change the file bytes (so the // content hash differs every send) but keep this identifier constant. self.add_column_if_missing("books", "source_uuid", "TEXT") .await?; // Device token whose `PUT .../state` last changed this reading state. // The firmware fetches sync before pushing its own progress, so echoing // a device's own state back hands it a stale position stamped with a // fresher `LastModified` — which it then adopts, losing the real page. self.add_column_if_missing("reading_states", "last_writer_token_id", "INTEGER") .await?; sqlx::query( "CREATE INDEX IF NOT EXISTS idx_books_source_uuid \ ON books(owner_user_id, source_uuid)", ) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to create source_uuid index")?; Ok(()) } /// Add a column to `table` when it is not already present. Idempotent guard /// around `ALTER TABLE ADD COLUMN`, which SQLite offers no `IF NOT EXISTS` /// for. `table`/`column`/`definition` are trusted, code-supplied literals. async fn add_column_if_missing( &self, table: &str, column: &str, definition: &str, ) -> Result<()> { let exists = sqlx::query(sqlx::AssertSqlSafe(format!( "SELECT 1 FROM pragma_table_info('{table}') WHERE name = ?" ))) .bind(column) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to inspect table columns")? .is_some(); if !exists { sqlx::query(sqlx::AssertSqlSafe(format!( "ALTER TABLE {table} ADD COLUMN {column} {definition}" ))) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to add column")?; } Ok(()) } // ---- User Operations ---- /// Find a user by username. pub async fn find_user_by_username(&self, username: &str) -> Result> { let row = sqlx::query( "SELECT id, username, password_hash, oidc_sub FROM users WHERE username = ?", ) .bind(username) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to find user by username")?; Ok(row.map(|r| UserRecord { id: r.get("id"), username: r.get("username"), password_hash: r.get("password_hash"), oidc_sub: r.get("oidc_sub"), })) } /// Find a user by ID. pub async fn find_user_by_id(&self, user_id: i64) -> Result> { let row = sqlx::query("SELECT id, username, password_hash, oidc_sub FROM users WHERE id = ?") .bind(user_id) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to find user by ID")?; Ok(row.map(|r| UserRecord { id: r.get("id"), username: r.get("username"), password_hash: r.get("password_hash"), oidc_sub: r.get("oidc_sub"), })) } /// Find a user by the OIDC `sub` claim their account is linked to. pub async fn find_user_by_oidc_sub(&self, oidc_sub: &str) -> Result> { let row = sqlx::query( "SELECT id, username, password_hash, oidc_sub FROM users WHERE oidc_sub = ?", ) .bind(oidc_sub) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to find user by oidc_sub")?; Ok(row.map(|r| UserRecord { id: r.get("id"), username: r.get("username"), password_hash: r.get("password_hash"), oidc_sub: r.get("oidc_sub"), })) } /// Create a new user with a pre-hashed password. pub async fn create_user(&self, username: &str, password_hash: &str) -> Result { let row = sqlx::query("INSERT INTO users (username, password_hash) VALUES (?, ?) RETURNING id") .bind(username) .bind(password_hash) .fetch_one(&self.pool) .await .change_context(Error::Database) .attach("Failed to create user")?; Ok(row.get("id")) } /// Create a user provisioned from an OIDC identity. The stored password hash /// is an unusable sentinel, so the account can only be reached via SSO — the /// password-login path rejects it (the sentinel is not a valid PHC hash). pub async fn create_oidc_user(&self, username: &str, oidc_sub: &str) -> Result { let row = sqlx::query( "INSERT INTO users (username, password_hash, oidc_sub) VALUES (?, ?, ?) RETURNING id", ) .bind(username) .bind(OIDC_NO_PASSWORD) .bind(oidc_sub) .fetch_one(&self.pool) .await .change_context(Error::Database) .attach("Failed to create OIDC user")?; Ok(row.get("id")) } /// Link an existing user to an OIDC `sub`, so future SSO logins resolve to /// this account (preserving its tokens and reading state). pub async fn link_oidc_sub(&self, user_id: i64, oidc_sub: &str) -> Result<()> { sqlx::query("UPDATE users SET oidc_sub = ? WHERE id = ?") .bind(oidc_sub) .bind(user_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to link OIDC sub")?; Ok(()) } /// Check if any users exist. pub async fn has_users(&self) -> Result { let row = sqlx::query("SELECT COUNT(*) as count FROM users") .fetch_one(&self.pool) .await .change_context(Error::Database) .attach("Failed to count users")?; let count: i64 = row.get("count"); Ok(count > 0) } // ---- Auth Token Operations ---- /// Look up a user by their auth token. pub async fn find_user_by_token(&self, token: &str) -> Result> { let row = sqlx::query("SELECT id, auth_token, user_id FROM auth_tokens WHERE auth_token = ?") .bind(token) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to find auth token")?; Ok(row.map(|r| AuthTokenRecord { id: r.get("id"), auth_token: r.get("auth_token"), user_id: r.get("user_id"), device_serial: None, })) } /// Mark a token as awaiting a device. The next `device_auth` with an unbound /// serial claims it. The `auth_token` must be owned by `user_id`; any prior /// pending row for the same token is replaced. pub async fn add_pending_registration(&self, user_id: i64, auth_token: &str) -> Result<()> { sqlx::query( r#"INSERT OR REPLACE INTO pending_registrations (auth_token, user_id) SELECT auth_token, user_id FROM auth_tokens WHERE auth_token = ? AND user_id = ?"#, ) .bind(auth_token) .bind(user_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to add pending registration")?; Ok(()) } /// Bind a device serial to the most recent pending token (within the last /// hour), consuming the pending row. Returns the bound token, or `None` when /// no pending registration is waiting. pub async fn claim_pending_registration(&self, serial: &str) -> Result> { let row = sqlx::query( r#"SELECT auth_token FROM pending_registrations WHERE created_at > strftime('%Y-%m-%dT%H:%M:%SZ', 'now', '-1 hour') ORDER BY created_at DESC LIMIT 1"#, ) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to read pending registration")?; let Some(token): Option = row.map(|r| r.get("auth_token")) else { return Ok(None); }; sqlx::query( r#"INSERT INTO device_registrations (serial, auth_token) VALUES (?, ?) ON CONFLICT(serial) DO UPDATE SET auth_token = excluded.auth_token"#, ) .bind(serial) .bind(&token) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to bind device serial")?; sqlx::query("DELETE FROM pending_registrations WHERE auth_token = ?") .bind(&token) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to clear pending registration")?; Ok(Some(token)) } /// Resolve a device serial number to its bound auth token, if any (and the /// token still exists). pub async fn find_token_by_serial(&self, serial: &str) -> Result> { let row = sqlx::query( r#"SELECT d.auth_token FROM device_registrations d JOIN auth_tokens t ON t.auth_token = d.auth_token WHERE d.serial = ?"#, ) .bind(serial) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to look up device serial")?; Ok(row.map(|r| r.get("auth_token"))) } /// Create an auth token for a user. Returns the new token's ID. pub async fn create_auth_token(&self, user_id: i64, token: &str) -> Result { let row = sqlx::query("INSERT INTO auth_tokens (auth_token, user_id) VALUES (?, ?) RETURNING id") .bind(token) .bind(user_id) .fetch_one(&self.pool) .await .change_context(Error::Database) .attach("Failed to create auth token")?; Ok(row.get("id")) } /// Delete all auth tokens for a user. pub async fn delete_auth_token(&self, user_id: i64) -> Result<()> { sqlx::query("DELETE FROM auth_tokens WHERE user_id = ?") .bind(user_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to delete auth token")?; Ok(()) } /// List all auth tokens for a user, each with the serial of any device /// currently bound to it (via `device_registrations`). pub async fn list_auth_tokens(&self, user_id: i64) -> Result> { let rows = sqlx::query( r#"SELECT t.id, t.auth_token, t.user_id, d.serial AS device_serial FROM auth_tokens t LEFT JOIN device_registrations d ON d.auth_token = t.auth_token WHERE t.user_id = ?"#, ) .bind(user_id) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to list auth tokens")?; Ok(rows .iter() .map(|r| AuthTokenRecord { id: r.get("id"), auth_token: r.get("auth_token"), user_id: r.get("user_id"), device_serial: r.get::, _>("device_serial"), }) .collect()) } /// The serial of a device bound to this token (scoped to the user), or /// `None` if the token is unbound or not owned by the user. Used to guard /// deletion of a token a Kobo device is actively syncing with. pub async fn token_device_serial(&self, token_id: i64, user_id: i64) -> Result> { let row = sqlx::query( r#"SELECT d.serial FROM auth_tokens t JOIN device_registrations d ON d.auth_token = t.auth_token WHERE t.id = ? AND t.user_id = ?"#, ) .bind(token_id) .bind(user_id) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to look up token device binding")?; Ok(row.map(|r| r.get("serial"))) } /// Delete a specific auth token by ID, scoped to a user. pub async fn delete_auth_token_by_id(&self, token_id: i64, user_id: i64) -> Result { let result = sqlx::query("DELETE FROM auth_tokens WHERE id = ? AND user_id = ?") .bind(token_id) .bind(user_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to delete auth token")?; Ok(result.rows_affected() > 0) } // ---- Reading State Operations ---- /// Get or create a reading state for a user/book. pub async fn get_or_create_reading_state( &self, user_id: i64, book_id: i64, ) -> Result { let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); sqlx::query( r#" INSERT OR IGNORE INTO reading_states (user_id, book_id, last_modified, priority_timestamp) VALUES (?, ?, ?, ?) "#, ) .bind(user_id) .bind(book_id) .bind(&now) .bind(&now) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to create reading state")?; let row = sqlx::query( "SELECT id, user_id, book_id, last_modified, priority_timestamp, last_writer_token_id \ FROM reading_states WHERE user_id = ? AND book_id = ?", ) .bind(user_id) .bind(book_id) .fetch_one(&self.pool) .await .change_context(Error::Database) .attach("Failed to get reading state")?; Ok(row_to_reading_state(&row)) } /// Look up the reading_states.last_modified fingerprint for a user's books. /// Used during snapshot creation. Returned map is keyed by book_id. pub async fn map_reading_state_last_modified( &self, user_id: i64, ) -> Result> { let rows = sqlx::query("SELECT book_id, last_modified FROM reading_states WHERE user_id = ?") .bind(user_id) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to map reading state last_modified")?; Ok(rows .iter() .map(|r| { ( r.get::("book_id"), r.get::("last_modified"), ) }) .collect()) } /// Look up the archive flag for a user's books. Returned map is keyed by book_id. pub async fn map_archived_books( &self, user_id: i64, ) -> Result> { let rows = sqlx::query("SELECT book_id, is_archived FROM archived_books WHERE user_id = ?") .bind(user_id) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to map archive flags")?; Ok(rows .iter() .map(|r| (r.get::("book_id"), r.get::("is_archived"))) .collect()) } /// Get bookmark for a reading state. pub async fn get_bookmark(&self, reading_state_id: i64) -> Result> { let row = sqlx::query( r#" SELECT id, reading_state_id, last_modified, location_source, location_type, location_value, progress_percent, content_source_progress_percent FROM bookmarks WHERE reading_state_id = ? "#, ) .bind(reading_state_id) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to get bookmark")?; Ok(row.map(|r| BookmarkRecord { id: r.get("id"), reading_state_id: r.get("reading_state_id"), last_modified: r.get("last_modified"), location_source: r.get("location_source"), location_type: r.get("location_type"), location_value: r.get("location_value"), progress_percent: r.get("progress_percent"), content_source_progress_percent: r.get("content_source_progress_percent"), })) } /// Get statistics for a reading state. pub async fn get_statistics(&self, reading_state_id: i64) -> Result> { let row = sqlx::query( r#" SELECT id, reading_state_id, last_modified, remaining_time_minutes, spent_reading_minutes FROM statistics WHERE reading_state_id = ? "#, ) .bind(reading_state_id) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to get statistics")?; Ok(row.map(|r| StatisticsRecord { id: r.get("id"), reading_state_id: r.get("reading_state_id"), last_modified: r.get("last_modified"), remaining_time_minutes: r.get("remaining_time_minutes"), spent_reading_minutes: r.get("spent_reading_minutes"), })) } /// Get read book record. pub async fn get_read_book( &self, user_id: i64, book_id: i64, ) -> Result> { let row = sqlx::query( r#" SELECT id, book_id, user_id, read_status, last_modified, last_time_started_reading, times_started_reading FROM read_books WHERE user_id = ? AND book_id = ? "#, ) .bind(user_id) .bind(book_id) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to get read book")?; Ok(row.map(|r| ReadBookRecord { id: r.get("id"), book_id: r.get("book_id"), user_id: r.get("user_id"), read_status: r.get("read_status"), last_modified: r.get("last_modified"), last_time_started_reading: r.get("last_time_started_reading"), times_started_reading: r.get("times_started_reading"), })) } /// Upsert a bookmark. pub async fn upsert_bookmark( &self, reading_state_id: i64, progress_percent: Option, content_source_progress_percent: Option, location_source: Option<&str>, location_type: Option<&str>, location_value: Option<&str>, ) -> Result<()> { let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); sqlx::query( r#" INSERT INTO bookmarks (reading_state_id, last_modified, progress_percent, content_source_progress_percent, location_source, location_type, location_value) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(reading_state_id) DO UPDATE SET last_modified = excluded.last_modified, progress_percent = COALESCE(excluded.progress_percent, bookmarks.progress_percent), content_source_progress_percent = COALESCE(excluded.content_source_progress_percent, bookmarks.content_source_progress_percent), location_source = COALESCE(excluded.location_source, bookmarks.location_source), location_type = COALESCE(excluded.location_type, bookmarks.location_type), location_value = COALESCE(excluded.location_value, bookmarks.location_value) "#, ) .bind(reading_state_id) .bind(&now) .bind(progress_percent) .bind(content_source_progress_percent) .bind(location_source) .bind(location_type) .bind(location_value) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to upsert bookmark")?; self.touch_reading_state(reading_state_id).await } /// Upsert statistics. pub async fn upsert_statistics( &self, reading_state_id: i64, spent_reading_minutes: Option, remaining_time_minutes: Option, ) -> Result<()> { let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); sqlx::query( r#" INSERT INTO statistics (reading_state_id, last_modified, spent_reading_minutes, remaining_time_minutes) VALUES (?, ?, ?, ?) ON CONFLICT(reading_state_id) DO UPDATE SET last_modified = excluded.last_modified, spent_reading_minutes = COALESCE(excluded.spent_reading_minutes, statistics.spent_reading_minutes), remaining_time_minutes = COALESCE(excluded.remaining_time_minutes, statistics.remaining_time_minutes) "#, ) .bind(reading_state_id) .bind(&now) .bind(spent_reading_minutes) .bind(remaining_time_minutes) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to upsert statistics")?; self.touch_reading_state(reading_state_id).await } /// Upsert read status. pub async fn upsert_read_status(&self, user_id: i64, book_id: i64, status: i32) -> Result<()> { let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); sqlx::query( r#" INSERT INTO read_books (user_id, book_id, read_status, last_modified, times_started_reading) VALUES (?, ?, ?, ?, 0) ON CONFLICT(user_id, book_id) DO UPDATE SET read_status = excluded.read_status, last_modified = excluded.last_modified, times_started_reading = CASE WHEN excluded.read_status = 2 THEN read_books.times_started_reading + 1 ELSE read_books.times_started_reading END, last_time_started_reading = CASE WHEN excluded.read_status = 2 THEN excluded.last_modified ELSE read_books.last_time_started_reading END "#, ) .bind(user_id) .bind(book_id) .bind(status) .bind(&now) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to upsert read status")?; Ok(()) } /// Update the last_modified timestamp on a reading state. async fn touch_reading_state(&self, reading_state_id: i64) -> Result<()> { let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); sqlx::query( "UPDATE reading_states SET last_modified = ?, priority_timestamp = ? WHERE id = ?", ) .bind(&now) .bind(&now) .bind(reading_state_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to touch reading state")?; Ok(()) } /// Record which device token last wrote a reading state. pub async fn set_reading_state_writer( &self, reading_state_id: i64, auth_token_id: i64, ) -> Result<()> { sqlx::query("UPDATE reading_states SET last_writer_token_id = ? WHERE id = ?") .bind(auth_token_id) .bind(reading_state_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to set reading state writer")?; Ok(()) } // ---- Archive Operations ---- /// Check if a book is archived for a user. pub async fn is_book_archived(&self, user_id: i64, book_id: i64) -> Result { let row = sqlx::query("SELECT is_archived FROM archived_books WHERE user_id = ? AND book_id = ?") .bind(user_id) .bind(book_id) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to check archive status")?; Ok(row .and_then(|r| r.get::, _>("is_archived")) .unwrap_or(false)) } /// Set archive status for a book. pub async fn set_book_archived( &self, user_id: i64, book_id: i64, is_archived: bool, ) -> Result<()> { let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); sqlx::query( r#" INSERT INTO archived_books (user_id, book_id, is_archived, last_modified) VALUES (?, ?, ?, ?) ON CONFLICT(user_id, book_id) DO UPDATE SET is_archived = excluded.is_archived, last_modified = excluded.last_modified "#, ) .bind(user_id) .bind(book_id) .bind(is_archived) .bind(&now) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to set archive status")?; Ok(()) } // ---- Settings & Sync Preferences ---- /// Read a boolean setting, returning `default` when the key is absent. pub async fn get_bool_setting(&self, key: &str, default: bool) -> Result { let row = sqlx::query("SELECT value FROM settings WHERE key = ?") .bind(key) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to read setting")?; Ok(row .map(|r| r.get::("value") == "1") .unwrap_or(default)) } /// Upsert a boolean setting, stored as `"1"`/`"0"`. pub async fn set_bool_setting(&self, key: &str, value: bool) -> Result<()> { sqlx::query( r#" INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value "#, ) .bind(key) .bind(if value { "1" } else { "0" }) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to write setting")?; Ok(()) } /// Set a user's per-book sync override. pub async fn set_book_sync_pref( &self, user_id: i64, book_id: i64, enabled: bool, ) -> Result<()> { sqlx::query( r#" INSERT INTO book_sync_prefs (user_id, book_id, enabled) VALUES (?, ?, ?) ON CONFLICT(user_id, book_id) DO UPDATE SET enabled = excluded.enabled "#, ) .bind(user_id) .bind(book_id) .bind(enabled) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to set book sync preference")?; Ok(()) } /// Record that a book's cover has been replaced by an override (stored on /// disk at `{cover_dir}/{book_id}.jpg`). Bumps the per-book cover `version` /// (busts the browser `?v=` and device `CoverImageId` caches) and touches /// `books.last_modified` so the next device sync re-emits the entitlement. /// Returns the new version. pub async fn set_book_cover(&self, book_id: i64) -> Result { let version: i64 = sqlx::query( r#" INSERT INTO book_covers (book_id, version) VALUES (?, 1) ON CONFLICT(book_id) DO UPDATE SET version = version + 1 RETURNING version "#, ) .bind(book_id) .fetch_one(&self.pool) .await .change_context(Error::Database) .attach("Failed to record book cover")? .get::("version"); sqlx::query( "UPDATE books SET last_modified = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", ) .bind(book_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to bump book last_modified after cover change")?; Ok(version) } /// Drop a book's override-cover record so it reverts to its source (Calibre / /// upload) cover. The caller removes the override file; this clears the /// `book_covers` row (so `cover_version` returns to 0 and `has_cover` reflects /// the source again) and bumps `books.last_modified` so the next device sync /// re-emits the entitlement with the bare-uuid `CoverImageId`. pub async fn remove_book_cover(&self, book_id: i64) -> Result<()> { sqlx::query("DELETE FROM book_covers WHERE book_id = ?") .bind(book_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to clear book cover override")?; sqlx::query( "UPDATE books SET last_modified = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", ) .bind(book_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to bump book last_modified after cover reset")?; Ok(()) } /// Store per-book metadata overrides (title, authors, series, description). /// Each argument replaces the source value on read; passing `None` clears /// that field's override so it falls back to the Calibre/upload value. The /// whole override row is rewritten each call. Touches `books.last_modified` /// so the next device sync re-emits the entitlement with the new metadata. #[allow(clippy::too_many_arguments)] pub async fn set_book_meta( &self, book_id: i64, title: Option<&str>, authors: Option<&str>, series_name: Option<&str>, series_index: Option, description: Option<&str>, ) -> Result<()> { sqlx::query( r#" INSERT INTO book_meta_overrides (book_id, title, authors, series_name, series_index, description) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(book_id) DO UPDATE SET title = excluded.title, authors = excluded.authors, series_name = excluded.series_name, series_index = excluded.series_index, description = excluded.description "#, ) .bind(book_id) .bind(title) .bind(authors) .bind(series_name) .bind(series_index) .bind(description) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to store book metadata override")?; sqlx::query( "UPDATE books SET last_modified = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?", ) .bind(book_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to bump book last_modified after metadata edit")?; Ok(()) } // ---- Shelf Operations ---- /// Get all shelves for a user. pub async fn get_shelves(&self, user_id: i64) -> Result> { let rows = sqlx::query( "SELECT id, uuid, name, user_id, kobo_sync, created, last_modified FROM shelves WHERE user_id = ?", ) .bind(user_id) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to get shelves")?; Ok(rows.iter().map(row_to_shelf).collect()) } /// Get book IDs in a shelf. pub async fn get_shelf_book_ids(&self, shelf_id: i64) -> Result> { let rows = sqlx::query("SELECT book_id FROM shelf_books WHERE shelf_id = ?") .bind(shelf_id) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to get shelf book IDs")?; Ok(rows.iter().map(|r| r.get::("book_id")).collect()) } /// Create a shelf. pub async fn create_shelf(&self, user_id: i64, uuid: &str, name: &str) -> Result<()> { let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); sqlx::query( "INSERT INTO shelves (uuid, name, user_id, kobo_sync, created, last_modified) VALUES (?, ?, ?, 1, ?, ?)", ) .bind(uuid) .bind(name) .bind(user_id) .bind(&now) .bind(&now) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to create shelf")?; Ok(()) } /// Find a shelf by UUID. pub async fn find_shelf_by_uuid(&self, uuid: &str) -> Result> { let row = sqlx::query( "SELECT id, uuid, name, user_id, kobo_sync, created, last_modified FROM shelves WHERE uuid = ?", ) .bind(uuid) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to find shelf")?; Ok(row.as_ref().map(row_to_shelf)) } /// Delete a shelf. The next snapshot diff surfaces it as a removed tag /// (the previous snapshot still has the shelf row; the new one does not). pub async fn delete_shelf(&self, user_id: i64, shelf_uuid: &str) -> Result<()> { sqlx::query("DELETE FROM shelves WHERE uuid = ? AND user_id = ?") .bind(shelf_uuid) .bind(user_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to delete shelf")?; Ok(()) } /// Rename a shelf. pub async fn rename_shelf(&self, shelf_uuid: &str, new_name: &str) -> Result<()> { let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); sqlx::query("UPDATE shelves SET name = ?, last_modified = ? WHERE uuid = ?") .bind(new_name) .bind(&now) .bind(shelf_uuid) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to rename shelf")?; Ok(()) } /// Add books to a shelf (by book_id from Calibre). Bumps the parent /// shelf's `last_modified` so the next snapshot detects the membership change. pub async fn add_books_to_shelf(&self, shelf_id: i64, book_ids: &[i64]) -> Result<()> { if book_ids.is_empty() { return Ok(()); } let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); for &book_id in book_ids { sqlx::query( "INSERT OR IGNORE INTO shelf_books (shelf_id, book_id, date_added) VALUES (?, ?, ?)", ) .bind(shelf_id) .bind(book_id) .bind(&now) .execute(&self.pool) .await .change_context(Error::Database) .attach_with(|| format!("Failed to add book {book_id} to shelf {shelf_id}"))?; } self.touch_shelf(shelf_id, &now).await } /// Remove books from a shelf. Bumps the parent shelf's `last_modified` so /// the next snapshot detects the membership change. pub async fn remove_books_from_shelf(&self, shelf_id: i64, book_ids: &[i64]) -> Result<()> { if book_ids.is_empty() { return Ok(()); } for &book_id in book_ids { sqlx::query("DELETE FROM shelf_books WHERE shelf_id = ? AND book_id = ?") .bind(shelf_id) .bind(book_id) .execute(&self.pool) .await .change_context(Error::Database) .attach_with(|| format!("Failed to remove book {book_id} from shelf {shelf_id}"))?; } let now = Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); self.touch_shelf(shelf_id, &now).await } async fn touch_shelf(&self, shelf_id: i64, now: &str) -> Result<()> { sqlx::query("UPDATE shelves SET last_modified = ? WHERE id = ?") .bind(now) .bind(shelf_id) .execute(&self.pool) .await .change_context(Error::Database) .attach_with(|| format!("Failed to touch shelf {shelf_id}"))?; Ok(()) } // ---- SyncPoint Operations ---- /// Insert a new sync point row. Returns the inserted row. pub async fn insert_sync_point( &self, id: &str, user_id: i64, auth_token_id: i64, ) -> Result { let now = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string(); sqlx::query( "INSERT INTO sync_points (id, user_id, auth_token_id, created_at) VALUES (?, ?, ?, ?)", ) .bind(id) .bind(user_id) .bind(auth_token_id) .bind(&now) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to insert sync point")?; Ok(SyncPointRow { id: id.to_string(), user_id, auth_token_id, created_at: now, }) } /// Read the server-held sync cursor for a device token, as /// `(last_sync_point_id, ongoing_sync_point_id)`. pub async fn get_device_sync_state( &self, auth_token_id: i64, ) -> Result<(Option, Option)> { let row = sqlx::query( "SELECT last_sync_point_id, ongoing_sync_point_id FROM device_sync_state WHERE auth_token_id = ?", ) .bind(auth_token_id) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to read device sync state")?; Ok(row .map(|r| (r.get("last_sync_point_id"), r.get("ongoing_sync_point_id"))) .unwrap_or((None, None))) } /// Write the server-held sync cursor for a device token. pub async fn upsert_device_sync_state( &self, auth_token_id: i64, last_sync_point_id: Option<&str>, ongoing_sync_point_id: Option<&str>, ) -> Result<()> { sqlx::query( r#" INSERT INTO device_sync_state (auth_token_id, last_sync_point_id, ongoing_sync_point_id, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(auth_token_id) DO UPDATE SET last_sync_point_id = excluded.last_sync_point_id, ongoing_sync_point_id = excluded.ongoing_sync_point_id, updated_at = excluded.updated_at "#, ) .bind(auth_token_id) .bind(last_sync_point_id) .bind(ongoing_sync_point_id) .bind(Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string()) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to write device sync state")?; Ok(()) } /// Look up a sync point by ID. pub async fn find_sync_point(&self, id: &str) -> Result> { let row = sqlx::query( "SELECT id, user_id, auth_token_id, created_at FROM sync_points WHERE id = ?", ) .bind(id) .fetch_optional(&self.pool) .await .change_context(Error::Database) .attach("Failed to find sync point")?; Ok(row.map(|r| SyncPointRow { id: r.get("id"), user_id: r.get("user_id"), auth_token_id: r.get("auth_token_id"), created_at: r.get("created_at"), })) } /// Delete a sync point and its child snapshot rows (cascading FKs). pub async fn delete_sync_point(&self, id: &str) -> Result<()> { sqlx::query("DELETE FROM sync_points WHERE id = ?") .bind(id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to delete sync point")?; Ok(()) } /// Delete sync points older than `cutoff` (ISO-8601 string). Used by the /// background GC task. Returns the number of rows removed. pub async fn gc_orphan_sync_points(&self, cutoff: &str) -> Result { let result = sqlx::query("DELETE FROM sync_points WHERE created_at < ?") .bind(cutoff) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to GC sync points")?; let rows = result.rows_affected(); if rows > 0 { debug!(rows = rows, cutoff = cutoff, "GC'd sync points"); } Ok(rows) } /// Insert a batch of book rows into a sync point. Chunks under SQLite's /// ~999 bound-parameter limit (9 columns × 100 rows = 900 params per stmt). pub async fn insert_sync_point_books( &self, sync_point_id: &str, books: &[SyncPointBookInsert], ) -> Result<()> { if books.is_empty() { return Ok(()); } const CHUNK: usize = 100; for chunk in books.chunks(CHUNK) { let placeholders = std::iter::repeat_n("(?, ?, ?, ?, ?, ?, ?, ?, ?, 0)", chunk.len()) .collect::>() .join(", "); let sql = format!( "INSERT INTO sync_point_books (sync_point_id, book_id, book_uuid, \ book_last_modified, book_timestamp, file_format, file_size, \ reading_state_last_modified, is_archived, synced) VALUES {placeholders}" ); let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)); for b in chunk { q = q .bind(sync_point_id) .bind(b.book_id) .bind(&b.book_uuid) .bind(&b.book_last_modified) .bind(&b.book_timestamp) .bind(&b.file_format) .bind(b.file_size) .bind(&b.reading_state_last_modified) .bind(b.is_archived); } q.execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to insert sync_point_books batch")?; } Ok(()) } /// Insert shelves into a sync point from the user's current shelves. pub async fn insert_sync_point_shelves_for_user( &self, sync_point_id: &str, user_id: i64, ) -> Result> { sqlx::query( r#" INSERT INTO sync_point_shelves (sync_point_id, shelf_id, shelf_uuid, shelf_name, created, last_modified, synced) SELECT ?, s.id, s.uuid, s.name, s.created, s.last_modified, 0 FROM shelves s WHERE s.user_id = ? AND s.kobo_sync = 1 "#, ) .bind(sync_point_id) .bind(user_id) .execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to insert sync_point_shelves")?; let rows = sqlx::query( r#" SELECT sync_point_id, shelf_id, shelf_uuid, shelf_name, created, last_modified FROM sync_point_shelves WHERE sync_point_id = ? "#, ) .bind(sync_point_id) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to fetch inserted sync_point_shelves")?; Ok(rows .iter() .map(|r| SyncPointShelfRow { sync_point_id: r.get("sync_point_id"), shelf_id: r.get("shelf_id"), shelf_uuid: r.get("shelf_uuid"), shelf_name: r.get("shelf_name"), created: r.get("created"), last_modified: r.get("last_modified"), }) .collect()) } /// Insert (sync_point_id, shelf_id, book_uuid) tuples for a snapshotted shelf. pub async fn insert_sync_point_shelf_books( &self, sync_point_id: &str, shelf_id: i64, book_uuids: &[String], ) -> Result<()> { if book_uuids.is_empty() { return Ok(()); } const CHUNK: usize = 200; for chunk in book_uuids.chunks(CHUNK) { let placeholders = std::iter::repeat_n("(?, ?, ?)", chunk.len()) .collect::>() .join(", "); let sql = format!( "INSERT INTO sync_point_shelf_books (sync_point_id, shelf_id, book_uuid) \ VALUES {placeholders}" ); let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)); for uuid in chunk { q = q.bind(sync_point_id).bind(shelf_id).bind(uuid); } q.execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to insert sync_point_shelf_books batch")?; } Ok(()) } /// Fetch the snapshotted book UUIDs for a shelf inside a sync point. pub async fn get_sync_point_shelf_book_uuids( &self, sync_point_id: &str, shelf_id: i64, ) -> Result> { let rows = sqlx::query( "SELECT book_uuid FROM sync_point_shelf_books \ WHERE sync_point_id = ? AND shelf_id = ? ORDER BY book_uuid", ) .bind(sync_point_id) .bind(shelf_id) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to fetch sync_point_shelf_books")?; Ok(rows .iter() .map(|r| r.get::("book_uuid")) .collect()) } // ---- SyncPoint diff queries (books) ---- /// `to` snapshot rows not present in `from` snapshot, unsynced, paginated. /// Returns (rows, has_next). pub async fn take_sync_point_books_added( &self, from: &str, to: &str, size: usize, ) -> Result<(Vec, bool)> { self.fetch_sync_point_books_page( r#" SELECT to_b.* FROM sync_point_books to_b WHERE to_b.sync_point_id = ?1 AND to_b.synced = 0 AND NOT EXISTS ( SELECT 1 FROM sync_point_books fb WHERE fb.sync_point_id = ?2 AND fb.book_id = to_b.book_id ) ORDER BY to_b.book_id LIMIT ?3 "#, to, from, size, ) .await } /// Books in both snapshots whose file/archive fingerprint differs. pub async fn take_sync_point_books_changed( &self, from: &str, to: &str, size: usize, ) -> Result<(Vec, bool)> { self.fetch_sync_point_books_page( r#" SELECT to_b.* FROM sync_point_books to_b JOIN sync_point_books fb ON fb.sync_point_id = ?2 AND fb.book_id = to_b.book_id WHERE to_b.sync_point_id = ?1 AND to_b.synced = 0 AND ( to_b.book_last_modified != fb.book_last_modified OR to_b.file_size != fb.file_size OR to_b.file_format != fb.file_format OR to_b.is_archived != fb.is_archived ) ORDER BY to_b.book_id LIMIT ?3 "#, to, from, size, ) .await } /// Books present in `from` but not `to`, unsynced. Read off the `from` snapshot. pub async fn take_sync_point_books_removed( &self, from: &str, to: &str, size: usize, ) -> Result<(Vec, bool)> { self.fetch_sync_point_books_page( r#" SELECT fb.* FROM sync_point_books fb WHERE fb.sync_point_id = ?1 AND fb.synced = 0 AND NOT EXISTS ( SELECT 1 FROM sync_point_books tb WHERE tb.sync_point_id = ?2 AND tb.book_id = fb.book_id ) ORDER BY fb.book_id LIMIT ?3 "#, from, to, size, ) .await } /// Books in both with file fingerprint identical, but reading-state fingerprint moved. pub async fn take_sync_point_books_progress_changed( &self, from: &str, to: &str, size: usize, ) -> Result<(Vec, bool)> { self.fetch_sync_point_books_page( r#" SELECT to_b.* FROM sync_point_books to_b JOIN sync_point_books fb ON fb.sync_point_id = ?2 AND fb.book_id = to_b.book_id WHERE to_b.sync_point_id = ?1 AND to_b.synced = 0 AND to_b.book_last_modified = fb.book_last_modified AND to_b.file_size = fb.file_size AND to_b.file_format = fb.file_format AND to_b.is_archived = fb.is_archived AND COALESCE(to_b.reading_state_last_modified,'') != COALESCE(fb.reading_state_last_modified,'') ORDER BY to_b.book_id LIMIT ?3 "#, to, from, size, ) .await } /// First-sync path: every row in the `to` snapshot, paginated. pub async fn take_sync_point_books_initial( &self, to: &str, size: usize, ) -> Result<(Vec, bool)> { let limit_plus_one = (size as i64) + 1; let rows = sqlx::query( r#" SELECT * FROM sync_point_books WHERE sync_point_id = ? AND synced = 0 ORDER BY book_id LIMIT ? "#, ) .bind(to) .bind(limit_plus_one) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to take initial sync_point_books page")?; let has_next = rows.len() as i64 > size as i64; let trimmed = rows.iter().take(size).map(row_to_sync_point_book).collect(); Ok((trimmed, has_next)) } async fn fetch_sync_point_books_page( &self, sql: &str, bind1: &str, bind2: &str, size: usize, ) -> Result<(Vec, bool)> { let limit_plus_one = (size as i64) + 1; let rows = sqlx::query(sqlx::AssertSqlSafe(sql)) .bind(bind1) .bind(bind2) .bind(limit_plus_one) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to fetch sync_point_books page")?; let has_next = rows.len() as i64 > size as i64; let trimmed = rows.iter().take(size).map(row_to_sync_point_book).collect(); Ok((trimmed, has_next)) } /// Mark a batch of book rows in a snapshot as synced. pub async fn mark_sync_point_books_synced( &self, sync_point_id: &str, book_ids: &[i64], ) -> Result<()> { if book_ids.is_empty() { return Ok(()); } const CHUNK: usize = 500; for chunk in book_ids.chunks(CHUNK) { let placeholders = std::iter::repeat_n("?", chunk.len()) .collect::>() .join(", "); let sql = format!( "UPDATE sync_point_books SET synced = 1 \ WHERE sync_point_id = ? AND book_id IN ({placeholders})" ); let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(sync_point_id); for &id in chunk { q = q.bind(id); } q.execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to mark sync_point_books synced")?; } Ok(()) } // ---- SyncPoint diff queries (shelves) ---- pub async fn take_sync_point_shelves_added( &self, from: &str, to: &str, size: usize, ) -> Result<(Vec, bool)> { self.fetch_sync_point_shelves_page( r#" SELECT to_s.* FROM sync_point_shelves to_s WHERE to_s.sync_point_id = ?1 AND to_s.synced = 0 AND NOT EXISTS ( SELECT 1 FROM sync_point_shelves fs WHERE fs.sync_point_id = ?2 AND fs.shelf_id = to_s.shelf_id ) ORDER BY to_s.shelf_id LIMIT ?3 "#, to, from, size, ) .await } pub async fn take_sync_point_shelves_changed( &self, from: &str, to: &str, size: usize, ) -> Result<(Vec, bool)> { self.fetch_sync_point_shelves_page( r#" SELECT to_s.* FROM sync_point_shelves to_s JOIN sync_point_shelves fs ON fs.sync_point_id = ?2 AND fs.shelf_id = to_s.shelf_id WHERE to_s.sync_point_id = ?1 AND to_s.synced = 0 AND to_s.last_modified != fs.last_modified ORDER BY to_s.shelf_id LIMIT ?3 "#, to, from, size, ) .await } pub async fn take_sync_point_shelves_removed( &self, from: &str, to: &str, size: usize, ) -> Result<(Vec, bool)> { self.fetch_sync_point_shelves_page( r#" SELECT fs.* FROM sync_point_shelves fs WHERE fs.sync_point_id = ?1 AND fs.synced = 0 AND NOT EXISTS ( SELECT 1 FROM sync_point_shelves ts WHERE ts.sync_point_id = ?2 AND ts.shelf_id = fs.shelf_id ) ORDER BY fs.shelf_id LIMIT ?3 "#, from, to, size, ) .await } pub async fn take_sync_point_shelves_initial( &self, to: &str, size: usize, ) -> Result<(Vec, bool)> { let limit_plus_one = (size as i64) + 1; let rows = sqlx::query( r#" SELECT * FROM sync_point_shelves WHERE sync_point_id = ? AND synced = 0 ORDER BY shelf_id LIMIT ? "#, ) .bind(to) .bind(limit_plus_one) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to take initial sync_point_shelves page")?; let has_next = rows.len() as i64 > size as i64; let trimmed = rows .iter() .take(size) .map(row_to_sync_point_shelf) .collect(); Ok((trimmed, has_next)) } async fn fetch_sync_point_shelves_page( &self, sql: &str, bind1: &str, bind2: &str, size: usize, ) -> Result<(Vec, bool)> { let limit_plus_one = (size as i64) + 1; let rows = sqlx::query(sqlx::AssertSqlSafe(sql)) .bind(bind1) .bind(bind2) .bind(limit_plus_one) .fetch_all(&self.pool) .await .change_context(Error::Database) .attach("Failed to fetch sync_point_shelves page")?; let has_next = rows.len() as i64 > size as i64; let trimmed = rows .iter() .take(size) .map(row_to_sync_point_shelf) .collect(); Ok((trimmed, has_next)) } pub async fn mark_sync_point_shelves_synced( &self, sync_point_id: &str, shelf_ids: &[i64], ) -> Result<()> { if shelf_ids.is_empty() { return Ok(()); } const CHUNK: usize = 500; for chunk in shelf_ids.chunks(CHUNK) { let placeholders = std::iter::repeat_n("?", chunk.len()) .collect::>() .join(", "); let sql = format!( "UPDATE sync_point_shelves SET synced = 1 \ WHERE sync_point_id = ? AND shelf_id IN ({placeholders})" ); let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)).bind(sync_point_id); for &id in chunk { q = q.bind(id); } q.execute(&self.pool) .await .change_context(Error::Database) .attach("Failed to mark sync_point_shelves synced")?; } Ok(()) } } /// Convert a sqlx Row into a ReadingStateRecord. fn row_to_reading_state(row: &sqlx::sqlite::SqliteRow) -> ReadingStateRecord { ReadingStateRecord { id: row.get("id"), user_id: row.get("user_id"), book_id: row.get("book_id"), last_modified: row.get("last_modified"), priority_timestamp: row.get("priority_timestamp"), last_writer_token_id: row.try_get("last_writer_token_id").ok().flatten(), } } /// Convert a sqlx Row into a ShelfRecord. fn row_to_shelf(row: &sqlx::sqlite::SqliteRow) -> ShelfRecord { ShelfRecord { id: row.get("id"), uuid: row.get("uuid"), name: row.get("name"), user_id: row.get("user_id"), kobo_sync: row.get("kobo_sync"), created: row.get("created"), last_modified: row.get("last_modified"), } } fn row_to_sync_point_book(row: &sqlx::sqlite::SqliteRow) -> SyncPointBookRow { SyncPointBookRow { sync_point_id: row.get("sync_point_id"), book_id: row.get("book_id"), book_uuid: row.get("book_uuid"), book_last_modified: row.get("book_last_modified"), book_timestamp: row.get("book_timestamp"), file_format: row.get("file_format"), file_size: row.get("file_size"), reading_state_last_modified: row.get("reading_state_last_modified"), is_archived: row.get("is_archived"), } } fn row_to_sync_point_shelf(row: &sqlx::sqlite::SqliteRow) -> SyncPointShelfRow { SyncPointShelfRow { sync_point_id: row.get("sync_point_id"), shelf_id: row.get("shelf_id"), shelf_uuid: row.get("shelf_uuid"), shelf_name: row.get("shelf_name"), created: row.get("created"), last_modified: row.get("last_modified"), } }