use std::collections::{HashMap, HashSet};
use rusqlite::{Connection, OptionalExtension, params};
use crate::db::connection::DbError;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StoredProfile {
pub uid: String,
pub rev: i64,
pub edited_at: i64,
pub doc: Option<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Saved {
pub rev: i64,
pub stored: bool,
}
fn next_rev(conn: &Connection, user: i64) -> Result<i64, DbError> {
Ok(conn.query_row(
"SELECT MAX(
COALESCE((SELECT MAX(rev) FROM dsp_profiles WHERE user_id = ?1), 0),
COALESCE((SELECT MAX(rev) FROM dsp_dismissed WHERE user_id = ?1), 0)
) + 1",
[user],
|r| r.get(0),
)?)
}
pub fn changes(
conn: &Connection,
user: i64,
since: i64,
) -> Result<(Vec<StoredProfile>, i64), DbError> {
let mut stmt = conn.prepare(
"SELECT uid, rev, edited_at, doc FROM dsp_profiles
WHERE user_id = ?1 AND rev > ?2 ORDER BY rev",
)?;
let rows = stmt
.query_map(params![user, since], |r| {
Ok(StoredProfile {
uid: r.get(0)?,
rev: r.get(1)?,
edited_at: r.get(2)?,
doc: r.get(3)?,
})
})?
.collect::<Result<Vec<_>, _>>()?;
let cursor = (next_rev(conn, user)? - 1).max(since);
Ok((rows, cursor))
}
pub fn save(
conn: &Connection,
user: i64,
uid: &str,
edited_at: i64,
doc: Option<&str>,
) -> Result<Saved, DbError> {
let existing: Option<(i64, i64)> = conn
.query_row(
"SELECT rev, edited_at FROM dsp_profiles WHERE user_id = ?1 AND uid = ?2",
params![user, uid],
|r| Ok((r.get(0)?, r.get(1)?)),
)
.optional()?;
if let Some((rev, at)) = existing
&& at > edited_at
{
return Ok(Saved { rev, stored: false });
}
let rev = next_rev(conn, user)?;
conn.execute(
"INSERT INTO dsp_profiles (user_id, uid, rev, edited_at, doc) VALUES (?1, ?2, ?3, ?4, ?5)
ON CONFLICT (user_id, uid) DO UPDATE
SET rev = excluded.rev, edited_at = excluded.edited_at, doc = excluded.doc",
params![user, uid, rev, edited_at, doc],
)?;
Ok(Saved { rev, stored: true })
}
pub fn live_docs(conn: &Connection, user: i64) -> Result<Vec<(String, String)>, DbError> {
let mut stmt = conn.prepare(
"SELECT uid, doc FROM dsp_profiles WHERE user_id = ?1 AND doc IS NOT NULL ORDER BY uid",
)?;
Ok(stmt
.query_map([user], |r| Ok((r.get(0)?, r.get(1)?)))?
.collect::<Result<Vec<_>, _>>()?)
}
pub fn files(conn: &Connection, user: i64) -> Result<HashMap<String, i64>, DbError> {
let mut stmt = conn.prepare("SELECT sha256, length(data) FROM dsp_files WHERE user_id = ?1")?;
Ok(stmt
.query_map([user], |r| Ok((r.get(0)?, r.get(1)?)))?
.collect::<Result<HashMap<_, _>, _>>()?)
}
pub fn file(conn: &Connection, user: i64, sha256: &str) -> Result<Option<Vec<u8>>, DbError> {
Ok(conn
.query_row(
"SELECT data FROM dsp_files WHERE user_id = ?1 AND sha256 = ?2",
params![user, sha256],
|r| r.get(0),
)
.optional()?)
}
pub fn store_file(conn: &Connection, user: i64, sha256: &str, data: &[u8]) -> Result<(), DbError> {
conn.execute(
"INSERT OR IGNORE INTO dsp_files (user_id, sha256, data) VALUES (?1, ?2, ?3)",
params![user, sha256, data],
)?;
Ok(())
}
pub fn keep_files(conn: &Connection, user: i64, keep: &HashSet<String>) -> Result<usize, DbError> {
let held = files(conn, user)?;
let mut dropped = 0;
for sha in held.keys().filter(|s| !keep.contains(*s)) {
dropped += conn.execute(
"DELETE FROM dsp_files WHERE user_id = ?1 AND sha256 = ?2",
params![user, sha],
)?;
}
Ok(dropped)
}
pub fn dismissed(conn: &Connection, user: i64) -> Result<Vec<String>, DbError> {
let mut stmt =
conn.prepare("SELECT output FROM dsp_dismissed WHERE user_id = ?1 ORDER BY output")?;
Ok(stmt
.query_map([user], |r| r.get(0))?
.collect::<Result<Vec<_>, _>>()?)
}
pub fn count(conn: &Connection, user: i64, uid: &str) -> Result<(i64, bool), DbError> {
Ok(conn.query_row(
"SELECT COUNT(*), COALESCE(SUM(uid = ?2), 0) > 0 FROM dsp_profiles WHERE user_id = ?1",
params![user, uid],
|r| Ok((r.get(0)?, r.get(1)?)),
)?)
}
pub fn dismissed_count(conn: &Connection, user: i64) -> Result<i64, DbError> {
Ok(conn.query_row(
"SELECT COUNT(*) FROM dsp_dismissed WHERE user_id = ?1",
[user],
|r| r.get(0),
)?)
}
pub fn dismiss(conn: &Connection, user: i64, output: &str) -> Result<bool, DbError> {
let rev = next_rev(conn, user)?;
Ok(conn.execute(
"INSERT OR IGNORE INTO dsp_dismissed (user_id, output, rev) VALUES (?1, ?2, ?3)",
params![user, output, rev],
)? > 0)
}
pub fn sync_cursor(conn: &Connection, url: &str) -> Result<i64, DbError> {
Ok(conn
.query_row(
"SELECT cursor FROM dsp_sync_cursor WHERE url = ?1",
[url],
|r| r.get(0),
)
.optional()?
.unwrap_or(0))
}
pub fn set_sync_cursor(conn: &Connection, url: &str, cursor: i64) -> Result<(), DbError> {
conn.execute(
"INSERT INTO dsp_sync_cursor (url, cursor) VALUES (?1, ?2)
ON CONFLICT (url) DO UPDATE SET cursor = excluded.cursor",
params![url, cursor],
)?;
Ok(())
}
pub fn synced(conn: &Connection, url: &str) -> Result<HashMap<String, (i64, String)>, DbError> {
let mut stmt = conn.prepare("SELECT uid, rev, hash FROM dsp_synced WHERE url = ?1")?;
Ok(stmt
.query_map([url], |r| Ok((r.get(0)?, (r.get(1)?, r.get(2)?))))?
.collect::<Result<HashMap<_, _>, _>>()?)
}
pub fn set_synced(
conn: &Connection,
url: &str,
uid: &str,
rev: i64,
hash: &str,
) -> Result<(), DbError> {
conn.execute(
"INSERT INTO dsp_synced (url, uid, rev, hash) VALUES (?1, ?2, ?3, ?4)
ON CONFLICT (url, uid) DO UPDATE SET rev = excluded.rev, hash = excluded.hash",
params![url, uid, rev, hash],
)?;
Ok(())
}
pub fn forget_synced(conn: &Connection, url: &str, uid: &str) -> Result<(), DbError> {
conn.execute(
"DELETE FROM dsp_synced WHERE url = ?1 AND uid = ?2",
params![url, uid],
)?;
Ok(())
}
pub fn forget_server(conn: &Connection, url: &str) -> Result<(), DbError> {
conn.execute("DELETE FROM dsp_synced WHERE url = ?1", [url])?;
conn.execute("DELETE FROM dsp_sync_cursor WHERE url = ?1", [url])?;
Ok(())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LocalEdit {
pub hash: String,
pub edited_at: i64,
pub refused: Option<String>,
pub note: Option<String>,
}
pub fn local_edits(conn: &Connection) -> Result<HashMap<String, LocalEdit>, DbError> {
let mut stmt = conn.prepare("SELECT uid, hash, edited_at, refused, note FROM dsp_local")?;
Ok(stmt
.query_map([], |r| {
Ok((
r.get(0)?,
LocalEdit {
hash: r.get(1)?,
edited_at: r.get(2)?,
refused: r.get(3)?,
note: r.get(4)?,
},
))
})?
.collect::<Result<HashMap<_, _>, _>>()?)
}
pub fn set_local_edit(
conn: &Connection,
uid: &str,
hash: &str,
edited_at: i64,
) -> Result<(), DbError> {
conn.execute(
"INSERT INTO dsp_local (uid, hash, edited_at, refused) VALUES (?1, ?2, ?3, NULL)
ON CONFLICT (uid) DO UPDATE
SET hash = excluded.hash, edited_at = excluded.edited_at, refused = NULL",
params![uid, hash, edited_at],
)?;
Ok(())
}
pub fn set_refused(conn: &Connection, uid: &str, refused: Option<&str>) -> Result<(), DbError> {
conn.execute(
"UPDATE dsp_local SET refused = ?2 WHERE uid = ?1",
params![uid, refused],
)?;
Ok(())
}
pub fn set_note(conn: &Connection, uid: &str, note: &str) -> Result<(), DbError> {
conn.execute(
"INSERT INTO dsp_local (uid, hash, edited_at, note) VALUES (?1, '', 0, ?2)
ON CONFLICT (uid) DO UPDATE SET note = excluded.note",
params![uid, note],
)?;
Ok(())
}
pub fn forget_local(conn: &Connection, uid: &str) -> Result<(), DbError> {
conn.execute("DELETE FROM dsp_local WHERE uid = ?1", [uid])?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn db() -> (rusqlite::Connection, i64) {
let conn = Connection::open_in_memory().unwrap();
conn.pragma_update(None, "foreign_keys", "on").unwrap();
crate::db::schema::create_tables(&conn).unwrap();
conn.execute(
"INSERT INTO users (id, username, password_hash, role) VALUES (1, 'mate', 'x', 'user')",
[],
)
.unwrap();
(conn, 1)
}
#[test]
fn the_last_edit_wins() {
let (conn, user) = db();
let c = &conn;
assert_eq!(
save(c, user, "a", 100, Some("v1")).unwrap(),
Saved {
rev: 1,
stored: true
}
);
assert_eq!(
save(c, user, "a", 50, Some("old")).unwrap(),
Saved {
rev: 1,
stored: false
}
);
assert_eq!(
save(c, user, "a", 200, None).unwrap(),
Saved {
rev: 2,
stored: true
}
);
assert!(dismiss(c, user, "AirPods").unwrap());
assert!(!dismiss(c, user, "AirPods").unwrap());
let (rows, cursor) = changes(c, user, 0).unwrap();
assert_eq!(cursor, 3);
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].doc, None);
assert_eq!(changes(c, user, 3).unwrap(), (vec![], 3));
assert!(live_docs(c, user).unwrap().is_empty());
}
#[test]
fn files_no_profile_names_are_dropped() {
let (conn, user) = db();
let c = &conn;
store_file(c, user, "aa", b"one").unwrap();
store_file(c, user, "bb", b"two").unwrap();
assert_eq!(files(c, user).unwrap().get("aa"), Some(&3));
keep_files(c, user, &HashSet::from(["bb".to_string()])).unwrap();
assert_eq!(file(c, user, "aa").unwrap(), None);
assert_eq!(file(c, user, "bb").unwrap().as_deref(), Some(&b"two"[..]));
}
}