use super::models::{AlertPref, RetentionPref, SourceRow, StoredClip, StoredDevice};
use super::{Store, StoreError};
use rusqlite::params;
use rusqlite::OptionalExtension;
pub fn insert_clip(store: &Store, c: &StoredClip) -> Result<(), StoreError> {
store.with_conn(|conn| {
conn.execute(
r#"INSERT OR REPLACE INTO clips
(id, source, source_key, content_type, content, media_path, byte_size, created_at, pinned, pinned_at, synced)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)"#,
params![
c.id, c.source, c.source_key, c.content_type, c.content,
c.media_path, c.byte_size, c.created_at,
if c.pinned { 1i64 } else { 0 }, c.pinned_at,
if c.synced { 1i64 } else { 0 },
],
)?;
Ok(())
})
}
pub fn list_clips(
store: &Store,
from: Option<&str>,
limit: Option<i64>,
since_ms: Option<i64>,
pinned_only: bool,
default_limit: i64,
) -> Result<Vec<StoredClip>, StoreError> {
let mut sql = String::from(
"SELECT id, source, source_key, content_type, content, media_path, byte_size, created_at, pinned, pinned_at, synced
FROM clips WHERE 1=1"
);
let mut binds: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(s) = from {
sql.push_str(" AND source = ?");
binds.push(Box::new(s.to_string()));
}
if let Some(t) = since_ms {
sql.push_str(" AND created_at >= ?");
binds.push(Box::new(t));
}
if pinned_only {
sql.push_str(" AND pinned = 1");
}
sql.push_str(" ORDER BY created_at DESC LIMIT ?");
binds.push(Box::new(limit.unwrap_or(default_limit)));
store.with_conn(|conn| {
let mut stmt = conn.prepare(&sql)?;
let rows: Vec<StoredClip> = stmt
.query_map(
rusqlite::params_from_iter(binds.iter().map(|b| &**b as &dyn rusqlite::ToSql)),
|r| {
Ok(StoredClip {
id: r.get(0)?,
source: r.get(1)?,
source_key: r.get(2)?,
content_type: r.get(3)?,
content: r.get(4)?,
media_path: r.get(5)?,
byte_size: r.get(6)?,
created_at: r.get(7)?,
pinned: r.get::<_, i64>(8)? != 0,
pinned_at: r.get(9)?,
synced: r.get::<_, i64>(10)? != 0,
})
},
)?
.filter_map(|r| r.ok())
.collect();
Ok(rows)
})
}
pub fn get_clip(store: &Store, id: &str) -> Result<Option<StoredClip>, StoreError> {
store.with_conn(|conn| {
let mut stmt = conn.prepare(
"SELECT id, source, source_key, content_type, content, media_path, byte_size, created_at, pinned, pinned_at, synced
FROM clips WHERE id = ?1"
)?;
let mut rows = stmt.query_map(params![id], |r| Ok(StoredClip {
id: r.get(0)?, source: r.get(1)?, source_key: r.get(2)?,
content_type: r.get(3)?, content: r.get(4)?, media_path: r.get(5)?,
byte_size: r.get(6)?, created_at: r.get(7)?,
pinned: r.get::<_, i64>(8)? != 0, pinned_at: r.get(9)?,
synced: r.get::<_, i64>(10)? != 0,
}))?;
if let Some(row) = rows.next() { Ok(Some(row?)) } else { Ok(None) }
})
}
pub fn delete_clip(store: &Store, id: &str) -> Result<(), StoreError> {
store.with_conn(|conn| {
conn.execute("DELETE FROM clips WHERE id = ?1", params![id])?;
Ok(())
})
}
pub fn set_pinned(store: &Store, id: &str, pinned: bool, when_ms: i64) -> Result<(), StoreError> {
store.with_conn(|conn| {
conn.execute(
"UPDATE clips SET pinned = ?1, pinned_at = CASE WHEN ?1 = 1 THEN ?2 ELSE NULL END WHERE id = ?3",
params![if pinned { 1i64 } else { 0 }, when_ms, id],
)?;
Ok(())
})
}
pub fn list_sources(store: &Store) -> Result<Vec<SourceRow>, StoreError> {
store.with_conn(|conn| {
let mut stmt = conn.prepare(
"SELECT source, COUNT(*) AS c, MAX(created_at) AS last_seen
FROM clips GROUP BY source ORDER BY last_seen DESC NULLS LAST",
)?;
let rows: Vec<SourceRow> = stmt
.query_map([], |r| {
Ok(SourceRow {
source: r.get(0)?,
clip_count: r.get(1)?,
last_seen: r.get(2)?,
})
})?
.filter_map(|r| r.ok())
.collect();
Ok(rows)
})
}
pub fn upsert_device(store: &Store, d: &StoredDevice) -> Result<(), StoreError> {
store.with_conn(|conn| {
conn.execute(
r#"INSERT OR REPLACE INTO devices
(id, hostname, nickname, source_key, machine_id, public_key,
paired_at, last_push_at, online, refreshed_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)"#,
params![
d.id,
d.hostname,
d.nickname,
d.source_key,
d.machine_id,
d.public_key,
d.paired_at,
d.last_push_at,
if d.online { 1i64 } else { 0 },
d.refreshed_at,
],
)?;
Ok(())
})
}
pub fn list_devices(store: &Store) -> Result<Vec<StoredDevice>, StoreError> {
store.with_conn(|conn| {
let mut stmt = conn.prepare(
"SELECT id, hostname, nickname, source_key, machine_id, public_key,
paired_at, last_push_at, online, refreshed_at
FROM devices ORDER BY last_push_at DESC NULLS LAST",
)?;
let rows: Vec<StoredDevice> = stmt
.query_map([], |r| {
Ok(StoredDevice {
id: r.get(0)?,
hostname: r.get(1)?,
nickname: r.get(2)?,
source_key: r.get(3)?,
machine_id: r.get(4)?,
public_key: r.get(5)?,
paired_at: r.get(6)?,
last_push_at: r.get(7)?,
online: r.get::<_, i64>(8)? != 0,
refreshed_at: r.get(9)?,
})
})?
.filter_map(|r| r.ok())
.collect();
Ok(rows)
})
}
pub fn set_retention(store: &Store, device_id: &str, days: i64) -> Result<(), StoreError> {
store.with_conn(|conn| {
conn.execute(
"INSERT INTO retention_prefs(device_id, days) VALUES(?1, ?2)
ON CONFLICT(device_id) DO UPDATE SET days = excluded.days",
params![device_id, days],
)?;
Ok(())
})
}
pub fn list_retention(store: &Store) -> Result<Vec<RetentionPref>, StoreError> {
store.with_conn(|conn| {
let mut stmt = conn.prepare("SELECT device_id, days FROM retention_prefs")?;
let rows: Vec<RetentionPref> = stmt
.query_map([], |r| {
Ok(RetentionPref {
device_id: r.get(0)?,
days: r.get(1)?,
})
})?
.filter_map(|r| r.ok())
.collect();
Ok(rows)
})
}
pub fn set_alert_pref(store: &Store, source: &str, enabled: bool) -> Result<(), StoreError> {
store.with_conn(|conn| {
conn.execute(
"INSERT INTO alert_prefs(source, enabled) VALUES(?1, ?2)
ON CONFLICT(source) DO UPDATE SET enabled = excluded.enabled",
params![source, if enabled { 1i64 } else { 0 }],
)?;
Ok(())
})
}
pub fn list_alert_prefs(store: &Store) -> Result<Vec<AlertPref>, StoreError> {
store.with_conn(|conn| {
let mut stmt = conn.prepare("SELECT source, enabled FROM alert_prefs")?;
let rows: Vec<AlertPref> = stmt
.query_map([], |r| {
Ok(AlertPref {
source: r.get(0)?,
enabled: r.get::<_, i64>(1)? != 0,
})
})?
.filter_map(|r| r.ok())
.collect();
Ok(rows)
})
}
pub fn watermark(store: &Store) -> Result<Option<String>, StoreError> {
store.with_conn(|conn| {
conn.query_row(
"SELECT value FROM meta WHERE key='last_sync_watermark'",
[],
|r| r.get::<_, String>(0),
)
.map(Some)
.or_else(|e| match e {
rusqlite::Error::QueryReturnedNoRows => Ok(None),
other => Err(other),
})
})
}
pub fn set_watermark(store: &Store, ulid: &str) -> Result<(), StoreError> {
store.with_conn(|conn| {
conn.execute(
"INSERT INTO meta(key, value) VALUES('last_sync_watermark', ?1)
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
params![ulid],
)?;
Ok(())
})
}
pub fn clip_count(store: &Store) -> Result<i64, StoreError> {
store.with_conn(|conn| conn.query_row("SELECT COUNT(*) FROM clips", [], |r| r.get::<_, i64>(0)))
}
pub fn count_clips_before(store: &Store, cutoff_ms: i64) -> Result<i64, StoreError> {
store.with_conn(|conn| {
conn.query_row(
"SELECT COUNT(*) FROM clips WHERE created_at < ?1",
rusqlite::params![cutoff_ms],
|r| r.get::<_, i64>(0),
)
})
}
pub fn clear_all_clips(store: &Store) -> Result<i64, StoreError> {
store.with_conn(|conn| {
let n = conn.execute("DELETE FROM clips", [])?;
Ok(n as i64)
})
}
pub fn search_clips(store: &Store, query: &str, limit: i64) -> Result<Vec<StoredClip>, StoreError> {
store.with_conn(|conn| {
let mut stmt = conn.prepare(
"SELECT c.id, c.source, c.source_key, c.content_type, c.content, c.media_path,
c.byte_size, c.created_at, c.pinned, c.pinned_at, c.synced
FROM clips c JOIN clips_fts f ON f.rowid = c.rowid
WHERE clips_fts MATCH ?1 ORDER BY rank LIMIT ?2",
)?;
let rows: Vec<StoredClip> = stmt
.query_map(params![query, limit], |r| {
Ok(StoredClip {
id: r.get(0)?,
source: r.get(1)?,
source_key: r.get(2)?,
content_type: r.get(3)?,
content: r.get(4)?,
media_path: r.get(5)?,
byte_size: r.get(6)?,
created_at: r.get(7)?,
pinned: r.get::<_, i64>(8)? != 0,
pinned_at: r.get(9)?,
synced: r.get::<_, i64>(10)? != 0,
})
})?
.filter_map(|r| r.ok())
.collect();
Ok(rows)
})
}
pub fn list_unsynced_clips(store: &Store) -> Result<Vec<StoredClip>, StoreError> {
store.with_conn(|conn| {
let mut stmt = conn.prepare(
"SELECT id, source, source_key, content_type, content, media_path, byte_size,
created_at, pinned, pinned_at, synced
FROM clips
WHERE synced = 0
ORDER BY created_at ASC",
)?;
let rows: Vec<StoredClip> = stmt
.query_map([], |r| {
Ok(StoredClip {
id: r.get(0)?,
source: r.get(1)?,
source_key: r.get(2)?,
content_type: r.get(3)?,
content: r.get(4)?,
media_path: r.get(5)?,
byte_size: r.get(6)?,
created_at: r.get(7)?,
pinned: r.get::<_, i64>(8)? != 0,
pinned_at: r.get(9)?,
synced: r.get::<_, i64>(10)? != 0,
})
})?
.collect::<Result<Vec<_>, _>>()?;
Ok(rows)
})
}
pub fn enforce_offline_cap(store: &Store, max: usize) -> Result<usize, StoreError> {
store.with_conn(|conn| {
let count: i64 =
conn.query_row("SELECT COUNT(*) FROM clips WHERE synced = 0", [], |r| {
r.get(0)
})?;
if (count as usize) <= max {
return Ok(0);
}
let excess = count as usize - max;
let dropped = conn.execute(
"DELETE FROM clips
WHERE id IN (
SELECT id FROM clips WHERE synced = 0 ORDER BY created_at ASC LIMIT ?1
)",
params![excess as i64],
)?;
log::warn!(
"offline queue cap: dropped {} oldest unsynced clips (cap={})",
dropped,
max
);
Ok(dropped)
})
}
pub fn replace_id_and_mark_synced(
store: &Store,
old_id: &str,
new_id: &str,
) -> Result<usize, StoreError> {
store.with_conn(|conn| {
let n = conn.execute(
"UPDATE clips SET id = ?1, synced = 1 WHERE id = ?2",
params![new_id, old_id],
)?;
Ok(n)
})
}
const LAST_FLUSH_KEY: &str = "backlog.last_flush_at";
pub fn get_last_flush_at(store: &Store) -> Result<Option<i64>, StoreError> {
store.with_conn(|conn| {
let v: Option<String> = conn
.query_row(
"SELECT value FROM meta WHERE key = ?1",
params![LAST_FLUSH_KEY],
|r| r.get(0),
)
.optional()?;
Ok(v.and_then(|s| s.parse().ok()))
})
}
pub fn set_last_flush_at(store: &Store, ts: i64) -> Result<(), StoreError> {
store.with_conn(|conn| {
conn.execute(
"INSERT INTO meta(key, value) VALUES(?1, ?2)
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
params![LAST_FLUSH_KEY, ts.to_string()],
)?;
Ok(())
})
}
#[cfg(test)]
mod tests {
use super::super::models::StoredClip;
use super::*;
#[test]
fn insert_clip_persists_synced_false() {
let store = Store::open(std::path::Path::new(":memory:")).unwrap();
let clip = StoredClip {
id: "01HXABC".into(),
source: "atlas0".into(),
source_key: None,
content_type: "text".into(),
content: Some(b"hello".to_vec()),
media_path: None,
byte_size: 5,
created_at: 1_700_000_000_000,
pinned: false,
pinned_at: None,
synced: false,
};
insert_clip(&store, &clip).unwrap();
let row = get_clip(&store, &clip.id).unwrap().unwrap();
assert!(
!row.synced,
"synced=false must survive an insert/read round-trip"
);
}
#[test]
fn list_unsynced_clips_returns_oldest_first() {
let store = Store::open(std::path::Path::new(":memory:")).unwrap();
fn make(id: &str, ts: i64, synced: bool) -> StoredClip {
StoredClip {
id: id.into(),
source: "s".into(),
source_key: None,
content_type: "text".into(),
content: Some(b"x".to_vec()),
media_path: None,
byte_size: 1,
created_at: ts,
pinned: false,
pinned_at: None,
synced,
}
}
for c in [
make("a", 30, true),
make("b", 10, false),
make("c", 20, false),
] {
insert_clip(&store, &c).unwrap();
}
let rows = list_unsynced_clips(&store).unwrap();
let ids: Vec<&str> = rows.iter().map(|c| c.id.as_str()).collect();
assert_eq!(ids, vec!["b", "c"]);
}
#[test]
fn enforce_offline_cap_drops_oldest_unsynced() {
let store = Store::open(std::path::Path::new(":memory:")).unwrap();
for (id, ts) in [("a", 10i64), ("b", 20), ("c", 30)] {
let c = StoredClip {
id: id.into(),
source: "s".into(),
source_key: None,
content_type: "text".into(),
content: Some(b"x".to_vec()),
media_path: None,
byte_size: 1,
created_at: ts,
pinned: false,
pinned_at: None,
synced: false,
};
insert_clip(&store, &c).unwrap();
}
let dropped = enforce_offline_cap(&store, 2).unwrap();
assert_eq!(dropped, 1);
let remaining = list_unsynced_clips(&store).unwrap();
let ids: Vec<&str> = remaining.iter().map(|c| c.id.as_str()).collect();
assert_eq!(ids, vec!["b", "c"]);
}
#[test]
fn enforce_offline_cap_is_noop_when_under_cap() {
let store = Store::open(std::path::Path::new(":memory:")).unwrap();
let c = StoredClip {
id: "a".into(),
source: "s".into(),
source_key: None,
content_type: "text".into(),
content: Some(b"x".to_vec()),
media_path: None,
byte_size: 1,
created_at: 0,
pinned: false,
pinned_at: None,
synced: false,
};
insert_clip(&store, &c).unwrap();
let dropped = enforce_offline_cap(&store, 10).unwrap();
assert_eq!(dropped, 0);
}
#[test]
fn replace_id_and_mark_synced_swaps_id_and_flag() {
let store = Store::open(std::path::Path::new(":memory:")).unwrap();
let c = StoredClip {
id: "local-01H".into(),
source: "s".into(),
source_key: None,
content_type: "text".into(),
content: Some(b"x".to_vec()),
media_path: None,
byte_size: 1,
created_at: 0,
pinned: false,
pinned_at: None,
synced: false,
};
insert_clip(&store, &c).unwrap();
let n = replace_id_and_mark_synced(&store, "local-01H", "01HRELAYID").unwrap();
assert_eq!(n, 1);
assert!(get_clip(&store, "local-01H").unwrap().is_none());
let after = get_clip(&store, "01HRELAYID").unwrap().unwrap();
assert!(after.synced);
}
#[test]
fn replace_id_and_mark_synced_is_benign_when_row_missing() {
let store = Store::open(std::path::Path::new(":memory:")).unwrap();
let n = replace_id_and_mark_synced(&store, "local-gone", "01HNEW").unwrap();
assert_eq!(n, 0);
}
#[test]
fn last_flush_at_roundtrips() {
let store = Store::open(std::path::Path::new(":memory:")).unwrap();
assert!(get_last_flush_at(&store).unwrap().is_none());
set_last_flush_at(&store, 1_700_000_000).unwrap();
assert_eq!(get_last_flush_at(&store).unwrap(), Some(1_700_000_000));
set_last_flush_at(&store, 1_700_001_000).unwrap();
assert_eq!(get_last_flush_at(&store).unwrap(), Some(1_700_001_000));
}
}