store: persist Docket state in SQLite
Outside dev mode the database now starts empty instead of serving fixtures, so nobody can sign in until users are configured and JMAP import exists. Assisted-by: Claude Opus 5.5 via Claude Code
diff --git a/.gitignore b/.gitignore
index ea8c4bf..c536345 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1 +1,4 @@
/target
+*.db
+*.db-shm
+*.db-wal
diff --git a/Cargo.lock b/Cargo.lock
index fbd105a..c015441 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -420,7 +420,9 @@ dependencies = [
"maud",
"miette",
"reqwest",
+ "rusqlite",
"serde",
+ "serde_json",
"tempfile",
"thiserror",
"tokio",
@@ -466,6 +468,18 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "fallible-iterator"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
+
+[[package]]
+name = "fallible-streaming-iterator"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
+
[[package]]
name = "fastrand"
version = "2.5.0"
@@ -484,6 +498,12 @@ version = "1.0.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
+[[package]]
+name = "foldhash"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
+
[[package]]
name = "form_urlencoded"
version = "1.2.2"
@@ -599,11 +619,32 @@ dependencies = [
"tracing",
]
+[[package]]
+name = "hashbrown"
+version = "0.16.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
+dependencies = [
+ "foldhash",
+]
+
[[package]]
name = "hashbrown"
version = "0.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
+dependencies = [
+ "foldhash",
+]
+
+[[package]]
+name = "hashlink"
+version = "0.12.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a596f1b20ed2cc5ecac41a164aaebc7258057060f06c0cf7a2ba3991ee7990fb"
+dependencies = [
+ "hashbrown 0.17.1",
+]
[[package]]
name = "heck"
@@ -896,7 +937,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
dependencies = [
"equivalent",
- "hashbrown",
+ "hashbrown 0.17.1",
]
[[package]]
@@ -933,10 +974,12 @@ dependencies = [
"jiff-core",
"jiff-static",
"jiff-tzdb-platform",
+ "js-sys",
"log",
"portable-atomic",
"portable-atomic-util",
"serde_core",
+ "wasm-bindgen",
"windows-link",
]
@@ -1071,6 +1114,17 @@ version = "0.2.189"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
+[[package]]
+name = "libsqlite3-sys"
+version = "0.38.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f1d20bef17f513b9b3004532233187769cd072d790971f4e4da0e346eb6401e8"
+dependencies = [
+ "cc",
+ "pkg-config",
+ "vcpkg",
+]
+
[[package]]
name = "linux-raw-sys"
version = "0.12.1"
@@ -1528,6 +1582,32 @@ dependencies = [
"windows-sys 0.52.0",
]
+[[package]]
+name = "rsqlite-vfs"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c51c9ae4df8a7fba42103df5c621fa3c37eccf3a3c650879e90fc48b11cc192c"
+dependencies = [
+ "hashbrown 0.16.1",
+ "thiserror",
+]
+
+[[package]]
+name = "rusqlite"
+version = "0.40.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "23f2a97da3e3873c73cb2a2e71b35c40ff95e0b1eefa8d72d8499a6928c3b5b3"
+dependencies = [
+ "bitflags 2.13.2",
+ "fallible-iterator",
+ "fallible-streaming-iterator",
+ "hashlink",
+ "jiff",
+ "libsqlite3-sys",
+ "smallvec",
+ "sqlite-wasm-rs",
+]
+
[[package]]
name = "rustc-demangle"
version = "0.1.28"
@@ -1831,6 +1911,18 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "sqlite-wasm-rs"
+version = "0.5.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc3efc0da82635d7e1ced0053bbbfa8c7ab9645d0bf36ceb4f7127bb85315d75"
+dependencies = [
+ "cc",
+ "js-sys",
+ "rsqlite-vfs",
+ "wasm-bindgen",
+]
+
[[package]]
name = "stable_deref_trait"
version = "1.2.1"
@@ -2249,6 +2341,12 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
+[[package]]
+name = "vcpkg"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
+
[[package]]
name = "version_check"
version = "0.9.5"
diff --git a/Cargo.toml b/Cargo.toml
index f217199..cc59094 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -16,7 +16,9 @@ jiff = "*"
kdl = "*"
maud = { version = "*", features = ["axum"] }
miette = { version = "*", features = ["fancy"] }
+rusqlite = { version = "*", features = ["bundled", "jiff"] }
serde = { version = "*", features = ["derive"] }
+serde_json = "*"
thiserror = "*"
tokio = { version = "*", features = ["full"] }
tower = "*"
diff --git a/docket.kdl b/docket.kdl
index 2c939c3..ec141c9 100644
--- a/docket.kdl
+++ b/docket.kdl
@@ -1,5 +1,7 @@
// Docket's settings, read from this file (see `--help` for pointing
// DOCKET_CONFIG elsewhere). This copy is the local-dev one: `dev` serves
-// fixture data without Tailscale, picking a user from the sidebar.
+// fixture data without Tailscale, picking a user from the sidebar, and
+// seeds the database with it the first time. Delete the file to start over.
bind "127.0.0.1:3000"
+database "docket.db"
dev
diff --git a/docs/DESIGN.md b/docs/DESIGN.md
index 8e88c44..b55387a 100644
--- a/docs/DESIGN.md
+++ b/docs/DESIGN.md
@@ -119,7 +119,7 @@ The primary other client is **Mail.app over IMAP**; Fastmail web stays in folder
## Data and sync
- **Mail server (JMAP)**: messages, folders, and `Docket/` state labels — source of truth for mail and filing.
-- **App database**: users, credentials (names matching `docket.kdl`; tokens stay in their files), accounts (credential + `accountId`), per-message state and assignees, per-user read, per-thread comments, history, sync state.
+- **App database**: a cache of threads and messages (filled by JMAP import; fixtures seed it in dev), users, credentials (names matching `docket.kdl`; tokens stay in their files), accounts (credential + `accountId`), per-message state and assignees, per-user read, per-thread comments, history, sync state.
- Records are keyed by account + Message-ID; JMAP ids and `threadId` are cached alongside, since they can change on reimport.
- One JMAP session and push connection (EventSource) per credential; `Email/changes` / `Mailbox/changes` per account, with periodic polling as a fallback.
- Rights are re-read on session refresh.
diff --git a/fly.kdl b/fly.kdl
index 23970e9..429e2b8 100644
--- a/fly.kdl
+++ b/fly.kdl
@@ -3,6 +3,9 @@
// because Fly's edge routes to the machine, not to localhost. `dev` is
// the whole of what makes them demos: fixture data instead of JMAP, and
// a stand-in identity in place of the Tailscale-User-Login header that
-// tailscale serve supplies in front of the real deployment.
+// tailscale serve supplies in front of the real deployment. The database
+// sits on the machine's own disk, which a deploy replaces, so each deploy
+// starts from the fixtures again.
bind "0.0.0.0:3000"
+database "/tmp/docket.db"
dev
diff --git a/migrations/0001_init.sql b/migrations/0001_init.sql
new file mode 100644
index 0000000..eb3adac
--- /dev/null
+++ b/migrations/0001_init.sql
@@ -0,0 +1,100 @@
+-- Docket's own data, plus a cache of the mail it tracks. The mail server
+-- stays the source of truth for messages and folders; JMAP import fills
+-- these tables and fixtures seed them in dev.
+
+CREATE TABLE users (
+ slug TEXT PRIMARY KEY,
+ name TEXT NOT NULL,
+ -- The Tailscale-User-Login value that identifies this user.
+ login TEXT NOT NULL UNIQUE
+) STRICT;
+
+CREATE TABLE accounts (
+ slug TEXT PRIMARY KEY,
+ name TEXT NOT NULL,
+ address TEXT NOT NULL,
+ read_only INTEGER NOT NULL CHECK (read_only IN (0, 1))
+) STRICT;
+
+-- Listed in insertion order, which is the order menus show them.
+CREATE TABLE folders (
+ name TEXT PRIMARY KEY
+) STRICT;
+
+CREATE TABLE threads (
+ id INTEGER PRIMARY KEY,
+ account TEXT NOT NULL REFERENCES accounts (slug),
+ subject TEXT NOT NULL,
+ -- JMAP's threadId, cached: it can change on reimport.
+ jmap_thread_id TEXT
+) STRICT;
+
+-- Received messages carry state and folder; sent ones carry who sent them
+-- and to whom. Each column belongs to one kind or the other.
+CREATE TABLE messages (
+ id INTEGER PRIMARY KEY,
+ account TEXT NOT NULL REFERENCES accounts (slug),
+ -- The RFC 5322 Message-ID, without angle brackets.
+ message_id TEXT NOT NULL,
+ -- JMAP's Email id, cached: it can change on reimport.
+ jmap_id TEXT,
+ thread INTEGER NOT NULL REFERENCES threads (id),
+ at TEXT NOT NULL,
+ -- JSON arrays of display names or addresses.
+ cc TEXT NOT NULL DEFAULT '[]',
+ bcc TEXT NOT NULL DEFAULT '[]',
+ body TEXT NOT NULL,
+ kind TEXT NOT NULL CHECK (kind IN ('received', 'sent')),
+ from_name TEXT,
+ from_addr TEXT,
+ state TEXT CHECK (state IN ('inbox', 'do', 'wait', 'watch', 'done')),
+ folder TEXT REFERENCES folders (name),
+ sent_by TEXT REFERENCES users (slug),
+ sent_to TEXT,
+ UNIQUE (account, message_id),
+ CHECK (
+ (kind = 'received'
+ AND from_name IS NOT NULL AND from_addr IS NOT NULL AND state IS NOT NULL
+ AND sent_by IS NULL AND sent_to IS NULL)
+ OR (kind = 'sent'
+ AND sent_by IS NOT NULL AND sent_to IS NOT NULL
+ AND from_name IS NULL AND from_addr IS NULL AND state IS NULL AND folder IS NULL)
+ )
+) STRICT;
+
+CREATE INDEX messages_thread ON messages (thread);
+CREATE INDEX messages_state ON messages (state);
+
+CREATE TABLE assignees (
+ message INTEGER NOT NULL REFERENCES messages (id),
+ user TEXT NOT NULL REFERENCES users (slug),
+ PRIMARY KEY (message, user)
+) STRICT;
+
+-- Per-person read tracking; $seen is shared and can't say who read what.
+CREATE TABLE reads (
+ user TEXT NOT NULL REFERENCES users (slug),
+ message INTEGER NOT NULL REFERENCES messages (id),
+ PRIMARY KEY (user, message)
+) STRICT;
+
+CREATE TABLE comments (
+ id INTEGER PRIMARY KEY,
+ thread INTEGER NOT NULL REFERENCES threads (id),
+ author TEXT NOT NULL REFERENCES users (slug),
+ at TEXT NOT NULL,
+ text TEXT NOT NULL
+) STRICT;
+
+CREATE INDEX comments_thread ON comments (thread);
+
+-- Every change to a message's values, in the words of its undo toast.
+CREATE TABLE history (
+ id INTEGER PRIMARY KEY,
+ message INTEGER NOT NULL REFERENCES messages (id),
+ user TEXT NOT NULL REFERENCES users (slug),
+ at TEXT NOT NULL,
+ event TEXT NOT NULL
+) STRICT;
+
+CREATE INDEX history_message ON history (message);
diff --git a/src/bin/docket/config.rs b/src/bin/docket/config.rs
index a37d493..efe1aab 100644
--- a/src/bin/docket/config.rs
+++ b/src/bin/docket/config.rs
@@ -14,6 +14,7 @@ use miette::SourceSpan;
use thiserror::Error;
const DEFAULT_BIND: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 3000);
+const DEFAULT_DATABASE: &str = "docket.db";
#[derive(Debug)]
pub struct Config {
@@ -24,6 +25,10 @@ pub struct Config {
/// Serve without Tailscale: requests with no identity header act as a
/// user picked from the sidebar. For local work against fixtures only.
pub dev: bool,
+
+ /// SQLite database file, created if missing. A relative path is
+ /// relative to the working directory.
+ pub database: Utf8PathBuf,
}
impl Config {
@@ -52,6 +57,7 @@ fn parse(source: &str) -> Result<Config, Error> {
let document = KdlDocument::parse(source)?;
let mut bind = None;
let mut dev = None;
+ let mut database = None;
for node in document.nodes() {
match node.name().value() {
@@ -73,6 +79,15 @@ fn parse(source: &str) -> Result<Config, Error> {
}
dev = Some(parse_dev(node)?);
}
+ "database" => {
+ if database.is_some() {
+ return Err(Error::Duplicate {
+ name: "database".to_owned(),
+ span: node.name().span(),
+ });
+ }
+ database = Some(parse_database(node)?);
+ }
name => {
return Err(Error::Unknown {
name: name.to_owned(),
@@ -85,6 +100,7 @@ fn parse(source: &str) -> Result<Config, Error> {
Ok(Config {
bind: bind.unwrap_or(DEFAULT_BIND),
dev: dev.unwrap_or(false),
+ database: database.unwrap_or_else(|| Utf8PathBuf::from(DEFAULT_DATABASE)),
})
}
@@ -98,6 +114,14 @@ fn parse_bind(node: &KdlNode) -> Result<SocketAddr, Error> {
})
}
+fn parse_database(node: &KdlNode) -> Result<Utf8PathBuf, Error> {
+ single_value(node, "database")?
+ .and_then(KdlValue::as_string)
+ .filter(|path| !path.is_empty())
+ .map(Utf8PathBuf::from)
+ .ok_or(Error::DatabaseValue { span: node.span() })
+}
+
fn parse_dev(node: &KdlNode) -> Result<bool, Error> {
match single_value(node, "dev")? {
None => Ok(true),
@@ -131,7 +155,10 @@ enum Error {
Kdl(#[from] KdlError),
#[error("unknown setting `{name}`")]
- #[diagnostic(code(docket::config::unknown), help("known settings: `bind`, `dev`"))]
+ #[diagnostic(
+ code(docket::config::unknown),
+ help("known settings: `bind`, `database`, `dev`")
+ )]
Unknown {
name: String,
#[label]
@@ -171,6 +198,13 @@ enum Error {
span: SourceSpan,
},
+ #[error(r#"database takes a quoted path, like database "docket.db""#)]
+ #[diagnostic(code(docket::config::database_value))]
+ DatabaseValue {
+ #[label]
+ span: SourceSpan,
+ },
+
#[error("dev takes #true or #false, or nothing at all (which means true)")]
#[diagnostic(code(docket::config::dev_value))]
DevValue {
@@ -189,6 +223,7 @@ mod tests {
let config = parse("").unwrap();
assert_eq!(config.bind.to_string(), "127.0.0.1:3000");
assert!(!config.dev);
+ assert_eq!(config.database, "docket.db");
}
#[test]
@@ -223,6 +258,26 @@ mod tests {
assert!(matches!(parse("bind"), Err(Error::BindValue { .. })));
}
+ #[test]
+ fn database_takes_a_path() {
+ assert_eq!(
+ parse(r#"database "/var/lib/docket/docket.db""#)
+ .unwrap()
+ .database,
+ "/var/lib/docket/docket.db"
+ );
+ for source in ["database", "database 1", r#"database """#] {
+ assert!(
+ matches!(parse(source), Err(Error::DatabaseValue { .. })),
+ "{source}"
+ );
+ }
+ assert!(matches!(
+ parse("database \"a.db\"\ndatabase \"b.db\""),
+ Err(Error::Duplicate { .. })
+ ));
+ }
+
#[test]
fn bare_dev_means_true() {
let config = parse("dev").unwrap();
diff --git a/src/bin/docket/main.rs b/src/bin/docket/main.rs
index 47915bd..057e2a8 100644
--- a/src/bin/docket/main.rs
+++ b/src/bin/docket/main.rs
@@ -18,8 +18,14 @@ async fn main() -> miette::Result<()> {
let args = config::Args::parse();
let config = config::Config::load(&args.config)?;
- // Fixtures stand in for the database and JMAP sync until they exist.
- let state = docket::routes::AppState::new(docket::fixtures::store(), config.dev);
+ // The fixtures' clock keeps their ages stable until JMAP sync brings
+ // real mail.
+ let store = docket::store::Store::open(&config.database, docket::fixtures::now())?;
+ if config.dev && store.is_empty()? {
+ docket::fixtures::seed(&store)?;
+ tracing::info!(database = %config.database, "seeded fixtures");
+ }
+ let state = docket::routes::AppState::new(store, config.dev);
let app = docket::routes::router(state);
let listener = TcpListener::bind(config.bind).await.into_diagnostic()?;
diff --git a/src/error.rs b/src/error.rs
index 699bcb1..4ad03cf 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -8,6 +8,10 @@ pub enum Error {
#[diagnostic(code(docket::io))]
Io(#[from] std::io::Error),
+ #[error("database error: {0}")]
+ #[diagnostic(code(docket::db))]
+ Db(#[from] rusqlite::Error),
+
#[error("no user identity on the request")]
#[diagnostic(code(docket::unauthenticated))]
Unauthenticated,
@@ -28,7 +32,7 @@ pub enum Error {
impl IntoResponse for Error {
fn into_response(self) -> Response {
let status = match self {
- Error::Io(_) => StatusCode::INTERNAL_SERVER_ERROR,
+ Error::Io(_) | Error::Db(_) => StatusCode::INTERNAL_SERVER_ERROR,
Error::Unauthenticated => StatusCode::UNAUTHORIZED,
Error::NotFound(_) => StatusCode::NOT_FOUND,
Error::Forbidden(_) => StatusCode::FORBIDDEN,
@@ -46,6 +50,7 @@ mod tests {
fn statuses() {
let cases = [
(Error::Io(std::io::Error::other("disk")), 500),
+ (Error::Db(rusqlite::Error::QueryReturnedNoRows), 500),
(Error::Unauthenticated, 401),
(Error::NotFound("thread"), 404),
(Error::Forbidden("read-only"), 403),
diff --git a/src/fixtures.rs b/src/fixtures.rs
index 7fa283b..3deac30 100644
--- a/src/fixtures.rs
+++ b/src/fixtures.rs
@@ -1,11 +1,12 @@
-//! Sample household mail, standing in for JMAP until sync exists. Names and
-//! messages follow the wireframes: Alex and Sam share the household account,
-//! and Eli's account is monitored read-only.
+//! Sample household mail that seeds a dev database, standing in for JMAP
+//! until import exists. Names and messages follow the wireframes: Alex and
+//! Sam share the household account, and Eli's account is monitored read-only.
use std::collections::BTreeSet;
use jiff::civil::{DateTime, date};
+use crate::Error;
use crate::model::{Account, Comment, Kind, Message, MessageId, State, Thread, User, Values};
use crate::store::Store;
@@ -29,13 +30,16 @@ fn strings(xs: &[&str]) -> Vec<String> {
xs.iter().map(|&x| x.to_owned()).collect()
}
+#[derive(Default)]
struct Builder {
- store: Store,
+ threads: Vec<Thread>,
+ messages: Vec<Message>,
+ comments: Vec<Comment>,
}
impl Builder {
fn thread(&mut self, id: u32, account: &str, subject: &str) {
- self.store.threads.push(Thread {
+ self.threads.push(Thread {
id,
account: account.to_owned(),
subject: subject.to_owned(),
@@ -53,7 +57,7 @@ impl Builder {
(state, folder, assignees): (State, Option<&str>, &[&str]),
cc: &[&str],
) {
- self.store.messages.push(Message {
+ self.messages.push(Message {
id,
message_id: message_id(id),
thread,
@@ -85,7 +89,7 @@ impl Builder {
body: &str,
(cc, bcc): (&[&str], &[&str]),
) {
- self.store.messages.push(Message {
+ self.messages.push(Message {
id,
message_id: message_id(id),
thread,
@@ -101,10 +105,10 @@ impl Builder {
}
fn comment(&mut self, thread: u32, author: &str, at: DateTime, text: &str) {
- let id = u32::try_from(self.store.comments.len())
+ let id = u32::try_from(self.comments.len())
.unwrap_or(u32::MAX)
.saturating_add(1);
- self.store.comments.push(Comment {
+ self.comments.push(Comment {
id,
thread,
author: author.to_owned(),
@@ -114,12 +118,16 @@ impl Builder {
}
}
-pub fn store() -> Store {
- let mut b = Builder {
- store: Store::default(),
- };
- b.store.now = now();
- b.store.users = vec![
+/// A fresh in-memory store holding the fixtures.
+pub fn store() -> Result<Store, Error> {
+ let store = Store::open_in_memory(now())?;
+ seed(&store)?;
+ Ok(store)
+}
+
+/// Writes the fixtures into an empty database.
+pub fn seed(store: &Store) -> Result<(), Error> {
+ let users = [
User {
slug: "alex".into(),
name: "Alex".into(),
@@ -131,7 +139,7 @@ pub fn store() -> Store {
login: "sam@example.com".into(),
},
];
- b.store.accounts = vec![
+ let accounts = [
Account {
slug: "household".into(),
name: "Household".into(),
@@ -145,7 +153,8 @@ pub fn store() -> Store {
read_only: true,
},
];
- b.store.folders = strings(&["Medical", "School", "House", "Finance"]);
+ let folders = ["Medical", "School", "House", "Finance"];
+ let mut b = Builder::default();
use State::{Do, Done, Inbox, Wait, Watch};
let roofer = ("Northwind Roofing", "office@northwind.co");
@@ -325,19 +334,36 @@ pub fn store() -> Store {
// Everything before yesterday has been read by both of us.
let cutoff = at(9, 30, 0, 0);
- let read: Vec<_> = b
- .store
+ let mut reads: Vec<_> = b
.messages
.iter()
.filter(|m| m.at < cutoff)
- .map(|m| m.id)
+ .flat_map(|m| [("alex", m.id), ("sam", m.id)])
.collect();
- for id in read {
- for user in ["alex", "sam"] {
- b.store.mark_read(user, id);
- }
- }
- b.store.mark_read("sam", 5);
+ reads.push(("sam", 5));
- b.store
+ store.import(|tx| {
+ for user in &users {
+ tx.user(user)?;
+ }
+ for account in &accounts {
+ tx.account(account)?;
+ }
+ for folder in folders {
+ tx.folder(folder)?;
+ }
+ for thread in &b.threads {
+ tx.thread(thread)?;
+ }
+ for message in &b.messages {
+ tx.message(message)?;
+ }
+ for comment in &b.comments {
+ tx.comment(comment)?;
+ }
+ for (user, id) in reads {
+ tx.read(user, id)?;
+ }
+ Ok(())
+ })
}
diff --git a/src/lists.rs b/src/lists.rs
index b140c29..66cbb92 100644
--- a/src/lists.rs
+++ b/src/lists.rs
@@ -2,8 +2,9 @@
//! individual messages, grouped by thread; a group holds only the thread's
//! messages that belong in that section.
-use crate::model::{Kind, Message, State, Thread, User};
-use crate::store::Store;
+use crate::Error;
+use crate::model::{Message, State, Thread, User};
+use crate::store::{Filter, Store};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum View {
@@ -29,38 +30,30 @@ enum Order {
}
#[derive(Debug)]
-pub struct Section<'a> {
+pub struct Section {
pub head: String,
/// Compact rows drop the snippet (Watch).
pub compact: bool,
- pub groups: Vec<Group<'a>>,
+ pub groups: Vec<Group>,
}
#[derive(Debug)]
-pub struct Group<'a> {
- pub thread: &'a Thread,
- pub rows: Vec<&'a Message>,
+pub struct Group {
+ pub thread: Thread,
+ pub rows: Vec<Message>,
}
-pub fn sections<'a>(store: &'a Store, me: &User, view: &View) -> Vec<Section<'a>> {
- let received = || store.messages.iter().filter(|m| m.values().is_some());
- let in_state = move |state| received().filter(move |m| m.state() == Some(state));
-
+pub fn sections(store: &Store, me: &User, view: &View) -> Result<Vec<Section>, Error> {
let mut out = Vec::new();
match view {
View::ForMe => {
for state in State::LANES {
- let msgs = in_state(state)
- .filter(|m| {
- m.is_assigned_to(&me.slug) || (state == State::Inbox && m.is_unassigned())
- })
- .collect();
- out.push(section(
- store,
- state.name().to_uppercase(),
- msgs,
- Order::NewestFirst,
- ));
+ let msgs = store.messages(Filter::ForMe {
+ user: &me.slug,
+ state,
+ })?;
+ let head = state.name().to_uppercase();
+ out.push(section(store, head, msgs, Order::NewestFirst)?);
}
}
View::Lane(state) => {
@@ -69,63 +62,46 @@ pub fn sections<'a>(store: &'a Store, me: &User, view: &View) -> Vec<Section<'a>
} else {
Order::OldestFirst
};
- let mut s = section(store, String::new(), in_state(*state).collect(), order);
+ let msgs = store.messages(Filter::State(*state))?;
+ let mut s = section(store, String::new(), msgs, order)?;
s.compact = *state == State::Watch;
out.push(s);
}
View::Search(query) => {
- let query = query.trim().to_lowercase();
+ let query = query.trim();
if !query.is_empty() {
- let hits = store
- .messages
- .iter()
- .filter(|m| matches(store, m, &query))
- .collect();
- out.push(section(
- store,
- "RESULTS".to_owned(),
- hits,
- Order::NewestFirst,
- ));
+ let hits = store.messages(Filter::Search(query))?;
+ let head = "RESULTS".to_owned();
+ out.push(section(store, head, hits, Order::NewestFirst)?);
}
}
}
out.retain(|s| !s.groups.is_empty());
- out
+ Ok(out)
}
/// How many messages a view lists, for the sidebar.
-pub fn count(store: &Store, me: &User, view: &View) -> usize {
- sections(store, me, view)
+pub fn count(store: &Store, me: &User, view: &View) -> Result<usize, Error> {
+ let n = sections(store, me, view)?
.iter()
.flat_map(|s| &s.groups)
.map(|g| g.rows.len())
- .sum()
-}
-
-fn matches(store: &Store, m: &Message, query: &str) -> bool {
- let subject = store.thread(m.thread).map_or("", |t| t.subject.as_str());
- let from = match &m.kind {
- Kind::Received { from, .. } => from.as_str(),
- Kind::Sent { by, .. } => store.user_name(by),
- };
- [subject, from, m.body.as_str()]
- .iter()
- .any(|s| s.to_lowercase().contains(query))
+ .sum();
+ Ok(n)
}
-fn section<'a>(
- store: &'a Store,
+/// Takes messages oldest first, as the store returns them.
+fn section(
+ store: &Store,
head: String,
- mut msgs: Vec<&'a Message>,
+ msgs: Vec<Message>,
order: Order,
-) -> Section<'a> {
- msgs.sort_by_key(|m| (m.at, m.id));
- let mut groups: Vec<Group<'a>> = Vec::new();
+) -> Result<Section, Error> {
+ let mut groups: Vec<Group> = Vec::new();
for m in msgs {
if let Some(g) = groups.iter_mut().find(|g| g.thread.id == m.thread) {
g.rows.push(m);
- } else if let Some(thread) = store.thread(m.thread) {
+ } else if let Some(thread) = store.thread(m.thread)? {
groups.push(Group {
thread,
rows: vec![m],
@@ -138,11 +114,11 @@ fn section<'a>(
groups.sort_by_key(|g| std::cmp::Reverse(g.rows.last().map(|m| m.at)));
}
}
- Section {
+ Ok(Section {
head,
compact: false,
groups,
- }
+ })
}
#[cfg(test)]
@@ -151,8 +127,9 @@ mod tests {
use crate::fixtures;
fn summary(store: &Store, user: &str, view: &View) -> Vec<(String, Vec<Vec<u32>>)> {
- let me = store.user(user).unwrap();
- sections(store, me, view)
+ let me = store.user(user).unwrap().unwrap();
+ sections(store, &me, view)
+ .unwrap()
.into_iter()
.map(|s| {
(
@@ -175,7 +152,7 @@ mod tests {
#[test]
fn for_me_groups_by_state_then_thread() {
- let store = fixtures::store();
+ let store = fixtures::store().unwrap();
let sam = summary(&store, "sam", &View::ForMe);
assert_eq!(heads(&store, "sam", &View::ForMe), ["INBOX", "DO", "WATCH"]);
let (_, groups) = &sam[0];
@@ -194,14 +171,14 @@ mod tests {
#[test]
fn sidebar_count() {
- let store = fixtures::store();
- let sam = store.user("sam").unwrap();
- assert_eq!(count(&store, sam, &View::ForMe), 7);
+ let store = fixtures::store().unwrap();
+ let sam = store.user("sam").unwrap().unwrap();
+ assert_eq!(count(&store, &sam, &View::ForMe).unwrap(), 7);
}
#[test]
fn lanes() {
- let store = fixtures::store();
+ let store = fixtures::store().unwrap();
// Every assignee shares one unheaded section: Inbox newest first, Do
// oldest first.
let inbox = summary(&store, "sam", &View::Lane(State::Inbox));
@@ -213,16 +190,21 @@ mod tests {
let wait = summary(&store, "sam", &View::Lane(State::Wait));
assert_eq!(wait[0].1, vec![vec![11]]);
- let me = store.user("sam").unwrap();
- let watch = sections(&store, me, &View::Lane(State::Watch));
+ let me = store.user("sam").unwrap().unwrap();
+ let watch = sections(&store, &me, &View::Lane(State::Watch)).unwrap();
assert!(watch[0].compact);
assert_eq!(watch[0].groups[0].rows[0].id, 14);
- assert!(sections(&store, me, &View::Lane(State::Done)).len() == 1);
+ assert!(
+ sections(&store, &me, &View::Lane(State::Done))
+ .unwrap()
+ .len()
+ == 1
+ );
}
#[test]
fn search_covers_every_message() {
- let store = fixtures::store();
+ let store = fixtures::store().unwrap();
let hits = summary(&store, "sam", &View::Search("downspout".into()));
assert_eq!(hits[0].1, vec![vec![2, 4]]);
let by_sender = summary(&store, "sam", &View::Search("ALEX".into()));
diff --git a/src/model.rs b/src/model.rs
index fa9670e..37bc54e 100644
--- a/src/model.rs
+++ b/src/model.rs
@@ -127,18 +127,6 @@ impl Message {
Kind::Sent { .. } => None,
}
}
-
- pub fn state(&self) -> Option<State> {
- self.values().map(|v| v.state)
- }
-
- pub fn is_assigned_to(&self, user: &str) -> bool {
- self.values().is_some_and(|v| v.assignees.contains(user))
- }
-
- pub fn is_unassigned(&self) -> bool {
- self.values().is_some_and(|v| v.assignees.is_empty())
- }
}
#[derive(Debug, Clone)]
@@ -150,6 +138,15 @@ pub struct Comment {
pub text: String,
}
+/// A change someone made to a message's values, as its toast put it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Event {
+ pub message: MessageId,
+ pub user: String,
+ pub at: DateTime,
+ pub text: String,
+}
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/src/routes.rs b/src/routes.rs
index 16678fd..4a8e212 100644
--- a/src/routes.rs
+++ b/src/routes.rs
@@ -9,21 +9,20 @@ use axum::routing::{get, post};
use axum::{Form, Router};
use maud::Markup;
use serde::Deserialize;
-use tokio::sync::RwLock;
use tower_http::trace::TraceLayer;
use crate::Error;
-use crate::lists::View;
+use crate::lists::{self, View};
use crate::model::{MessageId, State, ThreadId, User};
use crate::store::{Change, Store};
-use crate::views::{self, Page};
+use crate::views::{self, Page, ThreadView};
const IDENTITY_HEADER: &str = "Tailscale-User-Login";
const DEV_COOKIE: &str = "docket_dev_user";
#[derive(Clone)]
pub struct AppState {
- pub store: Arc<RwLock<Store>>,
+ pub store: Arc<Store>,
/// Without Tailscale in front, fall back to a cookie-chosen user.
pub dev: bool,
}
@@ -31,7 +30,7 @@ pub struct AppState {
impl AppState {
pub fn new(store: Store, dev: bool) -> Self {
Self {
- store: Arc::new(RwLock::new(store)),
+ store: Arc::new(store),
dev,
}
}
@@ -73,23 +72,25 @@ impl FromRequestParts<AppState> for Me {
if parts.method == Method::POST && !same_origin(&parts.headers) {
return Err(Error::Forbidden("cross-site request"));
}
- let store = state.store.read().await;
+ let store = &state.store;
if let Some(login) = header(&parts.headers, IDENTITY_HEADER) {
return store
- .user_by_login(login)
- .cloned()
+ .user_by_login(login)?
.map(Me)
.ok_or(Error::Unauthenticated);
}
if !state.dev {
return Err(Error::Unauthenticated);
}
- let chosen = cookie(&parts.headers, DEV_COOKIE).and_then(|slug| store.user(slug));
- chosen
- .or_else(|| store.users.first())
- .cloned()
- .map(Me)
- .ok_or(Error::Unauthenticated)
+ let chosen = match cookie(&parts.headers, DEV_COOKIE) {
+ Some(slug) => store.user(slug)?,
+ None => None,
+ };
+ let user = match chosen {
+ Some(user) => Some(user),
+ None => store.users()?.into_iter().next(),
+ };
+ user.map(Me).ok_or(Error::Unauthenticated)
}
}
@@ -125,7 +126,7 @@ async fn for_me(
OriginalUri(uri): OriginalUri,
Query(sel): Query<Selection>,
) -> Result<Markup, Error> {
- render(&state, &me, &View::ForMe, sel.m, &uri.to_string()).await
+ render(&state, &me, &View::ForMe, sel.m, &uri.to_string())
}
async fn lane(
@@ -138,7 +139,7 @@ async fn lane(
let lane = State::from_slug(&lane)
.filter(|s| State::LANES.contains(s))
.ok_or(Error::NotFound("lane"))?;
- render(&state, &me, &View::Lane(lane), sel.m, &uri.to_string()).await
+ render(&state, &me, &View::Lane(lane), sel.m, &uri.to_string())
}
async fn search(
@@ -148,40 +149,59 @@ async fn search(
Query(sel): Query<Selection>,
) -> Result<Markup, Error> {
let view = View::Search(sel.q.unwrap_or_default());
- render(&state, &me, &view, sel.m, &uri.to_string()).await
+ render(&state, &me, &view, sel.m, &uri.to_string())
}
-async fn render(
+fn render(
state: &AppState,
me: &User,
view: &View,
selected: Option<MessageId>,
here: &str,
) -> Result<Markup, Error> {
- let mut store = state.store.write().await;
- if let Some(id) = selected {
- let msg = store.message(id).ok_or(Error::NotFound("message"))?;
- // Opening a thread reads the selected message and the latest one,
- // the two that open expanded.
- let latest = store.thread_messages(msg.thread).last().map(|m| m.id);
- store.mark_read(&me.slug, id);
- if let Some(latest) = latest {
- store.mark_read(&me.slug, latest);
- }
- }
- let flash = store.take_flash(&me.slug);
- let store = &*store;
+ let store = &state.store;
+ let thread = selected.map(|id| open_thread(store, me, id)).transpose()?;
+ let nav = std::iter::once(View::ForMe)
+ .chain(State::LANES.map(View::Lane))
+ .map(|v| lists::count(store, me, &v).map(|n| (v, n)))
+ .collect::<Result<_, _>>()?;
Ok(views::page(&Page {
- store,
me,
view,
- selected: selected.and_then(|id| store.message(id)),
- flash,
+ now: store.now,
+ users: store.users()?,
+ folders: store.folders()?,
+ nav,
+ sections: lists::sections(store, me, view)?,
+ unread: store.unread(&me.slug)?,
+ selected,
+ thread,
+ flash: store.take_flash(&me.slug),
here,
dev: state.dev,
}))
}
+/// Opening a thread reads the selected message and the latest one, the two
+/// that open expanded.
+fn open_thread(store: &Store, me: &User, id: MessageId) -> Result<ThreadView, Error> {
+ let msg = store.message(id)?.ok_or(Error::NotFound("message"))?;
+ let latest = store
+ .thread_messages(msg.thread)?
+ .last()
+ .map_or(id, |m| m.id);
+ store.mark_read(&me.slug, id)?;
+ store.mark_read(&me.slug, latest)?;
+ Ok(ThreadView {
+ thread: store.thread(msg.thread)?.ok_or(Error::NotFound("thread"))?,
+ read_only: store
+ .thread_account(msg.thread)?
+ .is_some_and(|a| a.read_only),
+ timeline: store.timeline(msg.thread)?,
+ latest,
+ })
+}
+
/// Only same-site paths, so a form can't bounce the browser elsewhere.
fn safe_back(back: &str) -> &str {
if back.starts_with('/') && !back.starts_with("//") && !back.contains('\\') {
@@ -252,7 +272,7 @@ async fn edit(
change: Change,
back: &str,
) -> Result<Redirect, Error> {
- state.store.write().await.edit(&me.slug, id, change)?;
+ state.store.edit(&me.slug, id, change)?;
Ok(Redirect::to(safe_back(back)))
}
@@ -268,11 +288,7 @@ async fn add_comment(
Path(thread): Path<ThreadId>,
Form(form): Form<CommentForm>,
) -> Result<Redirect, Error> {
- state
- .store
- .write()
- .await
- .add_comment(&me.slug, thread, &form.text)?;
+ state.store.add_comment(&me.slug, thread, &form.text)?;
Ok(Redirect::to(safe_back(&form.back)))
}
@@ -286,7 +302,7 @@ async fn undo(
Me(me): Me,
Form(form): Form<BackForm>,
) -> Result<Redirect, Error> {
- state.store.write().await.undo(&me.slug)?;
+ state.store.undo(&me.slug)?;
Ok(Redirect::to(safe_back(&form.back)))
}
@@ -298,7 +314,7 @@ async fn dev_user(
if !same_origin(&headers) {
return Err(Error::Forbidden("cross-site request"));
}
- if state.store.read().await.user(&form.user).is_none() {
+ if state.store.user(&form.user)?.is_none() {
return Err(Error::NotFound("user"));
}
let cookie = format!("{DEV_COOKIE}={}; Path=/; SameSite=Lax", form.user);
diff --git a/src/store.rs b/src/store.rs
index f1b6a6d..2a9412f 100644
--- a/src/store.rs
+++ b/src/store.rs
@@ -1,15 +1,26 @@
-//! An in-memory store seeded from fixtures. It stands in for the database
-//! and JMAP sync until those exist; edits last until the process exits.
+//! Docket's data in SQLite: a cache of the mail it tracks (from fixtures in
+//! dev until JMAP import exists), and the state, assignees, reads, comments,
+//! and history only Docket keeps. Every read is a query returning owned
+//! values. Undo and toasts belong to the browsing session and stay in
+//! memory.
use std::collections::{BTreeMap, BTreeSet};
+use std::sync::{Mutex, MutexGuard, PoisonError};
+use camino::Utf8Path;
use jiff::civil::DateTime;
+use rusqlite::types::{FromSql, FromSqlError, FromSqlResult, ToSqlOutput, ValueRef};
+use rusqlite::{Connection, OptionalExtension as _, Params, Row, ToSql, Transaction, params};
use crate::Error;
use crate::model::{
- Account, Comment, Kind, Message, MessageId, State, Thread, ThreadId, User, Values,
+ Account, Comment, CommentId, Event, Kind, Message, MessageId, State, Thread, ThreadId, User,
+ Values,
};
+/// Applied in order; `PRAGMA user_version` counts how many have run.
+const MIGRATIONS: &[&str] = &[include_str!("../migrations/0001_init.sql")];
+
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Change {
State(State),
@@ -25,102 +36,263 @@ pub struct Flash {
}
/// An entry in a thread's chain, in time order.
+#[derive(Debug, Clone)]
+pub enum Item {
+ Message(Message),
+ Comment(Comment),
+}
+
+/// Which messages a list asks for.
#[derive(Debug, Clone, Copy)]
-pub enum Item<'a> {
- Message(&'a Message),
- Comment(&'a Comment),
+pub enum Filter<'a> {
+ /// Received messages in a state.
+ State(State),
+ /// Received messages in a state that are the user's: assigned to them,
+ /// or unassigned in Inbox.
+ ForMe { user: &'a str, state: State },
+ /// Messages whose subject, sender, or body contains the text. Case is
+ /// ignored for ASCII letters only, as SQLite's LIKE does.
+ Search(&'a str),
}
-#[derive(Debug, Default)]
+/// One connection behind a mutex: two people's clicks never contend long
+/// enough to need a pool, and no lock is held across an await.
+#[derive(Debug)]
pub struct Store {
/// The clock fixtures are written against, so ages stay stable.
pub now: DateTime,
- pub users: Vec<User>,
- pub accounts: Vec<Account>,
- pub folders: Vec<String>,
- pub threads: Vec<Thread>,
- pub messages: Vec<Message>,
- pub comments: Vec<Comment>,
- /// (user slug, message) pairs that user has read.
- pub read: BTreeSet<(String, MessageId)>,
+ inner: Mutex<Inner>,
+}
+
+#[derive(Debug)]
+struct Inner {
+ conn: Connection,
undo: BTreeMap<String, (MessageId, Values)>,
flash: BTreeMap<String, Flash>,
}
impl Store {
- pub fn user(&self, slug: &str) -> Option<&User> {
- self.users.iter().find(|u| u.slug == slug)
+ /// Opens (creating if needed) the database file and brings its schema
+ /// up to date.
+ pub fn open(path: &Utf8Path, now: DateTime) -> Result<Self, Error> {
+ let conn = Connection::open(path)?;
+ conn.pragma_update(None, "journal_mode", "WAL")?;
+ Self::init(conn, now)
}
- pub fn user_by_login(&self, login: &str) -> Option<&User> {
- self.users.iter().find(|u| u.login == login)
+ /// A database that lasts as long as the store, for tests.
+ pub fn open_in_memory(now: DateTime) -> Result<Self, Error> {
+ Self::init(Connection::open_in_memory()?, now)
}
- pub fn user_name<'a>(&'a self, slug: &'a str) -> &'a str {
- self.user(slug).map_or(slug, |u| u.name.as_str())
+ fn init(mut conn: Connection, now: DateTime) -> Result<Self, Error> {
+ conn.pragma_update(None, "foreign_keys", true)?;
+ migrate(&mut conn)?;
+ Ok(Self {
+ now,
+ inner: Mutex::new(Inner {
+ conn,
+ undo: BTreeMap::new(),
+ flash: BTreeMap::new(),
+ }),
+ })
}
- pub fn account(&self, slug: &str) -> Option<&Account> {
- self.accounts.iter().find(|a| a.slug == slug)
+ /// True before anyone has been added: a fresh database.
+ pub fn is_empty(&self) -> Result<bool, Error> {
+ let inner = self.lock();
+ let empty = inner
+ .conn
+ .query_row("SELECT NOT EXISTS (SELECT 1 FROM users)", [], |r| r.get(0))?;
+ Ok(empty)
}
- pub fn thread(&self, id: ThreadId) -> Option<&Thread> {
- self.threads.iter().find(|t| t.id == id)
+ /// Writes mail and people in one transaction: fixtures now, JMAP later.
+ pub fn import(&self, f: impl FnOnce(&Import<'_>) -> Result<(), Error>) -> Result<(), Error> {
+ let mut inner = self.lock();
+ let import = Import {
+ tx: inner.conn.transaction()?,
+ };
+ f(&import)?;
+ import.tx.commit()?;
+ Ok(())
}
- pub fn message(&self, id: MessageId) -> Option<&Message> {
- self.messages.iter().find(|m| m.id == id)
+ pub fn users(&self) -> Result<Vec<User>, Error> {
+ let inner = self.lock();
+ let mut stmt = inner
+ .conn
+ .prepare_cached("SELECT slug, name, login FROM users ORDER BY rowid")?;
+ let users = stmt
+ .query_map([], user_from_row)?
+ .collect::<Result<_, _>>()?;
+ Ok(users)
}
- pub fn thread_account(&self, thread: ThreadId) -> Option<&Account> {
- self.thread(thread).and_then(|t| self.account(&t.account))
+ pub fn user(&self, slug: &str) -> Result<Option<User>, Error> {
+ load_user(&self.lock().conn, slug)
}
- pub fn thread_messages(&self, thread: ThreadId) -> Vec<&Message> {
- let mut msgs: Vec<_> = self
- .messages
- .iter()
- .filter(|m| m.thread == thread)
- .collect();
- msgs.sort_by_key(|m| (m.at, m.id));
- msgs
+ pub fn user_by_login(&self, login: &str) -> Result<Option<User>, Error> {
+ let user = self
+ .lock()
+ .conn
+ .query_row(
+ "SELECT slug, name, login FROM users WHERE login = ?1",
+ [login],
+ user_from_row,
+ )
+ .optional()?;
+ Ok(user)
+ }
+
+ pub fn folders(&self) -> Result<Vec<String>, Error> {
+ let inner = self.lock();
+ let mut stmt = inner
+ .conn
+ .prepare_cached("SELECT name FROM folders ORDER BY rowid")?;
+ let folders = stmt
+ .query_map([], |r| r.get(0))?
+ .collect::<Result<_, _>>()?;
+ Ok(folders)
+ }
+
+ pub fn thread(&self, id: ThreadId) -> Result<Option<Thread>, Error> {
+ let thread = self
+ .lock()
+ .conn
+ .query_row(
+ "SELECT id, account, subject FROM threads WHERE id = ?1",
+ [id],
+ |r| {
+ Ok(Thread {
+ id: r.get(0)?,
+ account: r.get(1)?,
+ subject: r.get(2)?,
+ })
+ },
+ )
+ .optional()?;
+ Ok(thread)
+ }
+
+ pub fn thread_account(&self, thread: ThreadId) -> Result<Option<Account>, Error> {
+ load_thread_account(&self.lock().conn, thread)
+ }
+
+ pub fn message(&self, id: MessageId) -> Result<Option<Message>, Error> {
+ load_message(&self.lock().conn, id)
+ }
+
+ /// The thread's messages, oldest first.
+ pub fn thread_messages(&self, thread: ThreadId) -> Result<Vec<Message>, Error> {
+ load_messages(&self.lock().conn, "m.thread = ?1", [thread])
}
- pub fn timeline(&self, thread: ThreadId) -> Vec<Item<'_>> {
- let mut items: Vec<_> = self
- .thread_messages(thread)
+ /// Messages matching the filter, oldest first.
+ pub fn messages(&self, filter: Filter<'_>) -> Result<Vec<Message>, Error> {
+ let inner = self.lock();
+ let conn = &inner.conn;
+ match filter {
+ Filter::State(state) => {
+ load_messages(conn, "m.kind = 'received' AND m.state = ?1", [state])
+ }
+ Filter::ForMe { user, state } => load_messages(
+ conn,
+ "m.kind = 'received' AND m.state = ?1 AND (
+ EXISTS (SELECT 1 FROM assignees a WHERE a.message = m.id AND a.user = ?2)
+ OR (?1 = 'inbox' AND NOT EXISTS (SELECT 1 FROM assignees a WHERE a.message = m.id))
+ )",
+ params![state, user],
+ ),
+ Filter::Search(text) => {
+ let pattern = format!("%{}%", escape_like(text));
+ load_messages(
+ conn,
+ r"t.subject LIKE ?1 ESCAPE '\' OR m.from_name LIKE ?1 ESCAPE '\'
+ OR u.name LIKE ?1 ESCAPE '\' OR m.body LIKE ?1 ESCAPE '\'",
+ [pattern],
+ )
+ }
+ }
+ }
+
+ /// The thread's messages and comments in time order. A message sorts
+ /// before a comment made at the same moment.
+ pub fn timeline(&self, thread: ThreadId) -> Result<Vec<Item>, Error> {
+ let inner = self.lock();
+ let messages = load_messages(&inner.conn, "m.thread = ?1", [thread])?;
+ let mut stmt = inner.conn.prepare_cached(
+ "SELECT id, thread, author, at, text FROM comments WHERE thread = ?1 ORDER BY at, id",
+ )?;
+ let comments = stmt
+ .query_map([thread], |r| {
+ Ok(Comment {
+ id: r.get(0)?,
+ thread: r.get(1)?,
+ author: r.get(2)?,
+ at: r.get(3)?,
+ text: r.get(4)?,
+ })
+ })?
+ .collect::<Result<Vec<_>, _>>()?;
+ let mut items: Vec<_> = messages
.into_iter()
.map(Item::Message)
- .chain(
- self.comments
- .iter()
- .filter(|c| c.thread == thread)
- .map(Item::Comment),
- )
+ .chain(comments.into_iter().map(Item::Comment))
.collect();
- // Stable: a message sorts before a comment made at the same moment.
items.sort_by_key(|item| match item {
Item::Message(m) => m.at,
Item::Comment(c) => c.at,
});
- items
+ Ok(items)
}
- pub fn is_unread(&self, user: &str, msg: &Message) -> bool {
- matches!(msg.kind, Kind::Received { .. }) && !self.read.contains(&(user.to_owned(), msg.id))
+ /// Received messages the user hasn't opened.
+ pub fn unread(&self, user: &str) -> Result<BTreeSet<MessageId>, Error> {
+ let inner = self.lock();
+ let mut stmt = inner.conn.prepare_cached(
+ "SELECT id FROM messages WHERE kind = 'received'
+ AND id NOT IN (SELECT message FROM reads WHERE user = ?1)",
+ )?;
+ let ids = stmt
+ .query_map([user], |r| r.get(0))?
+ .collect::<Result<_, _>>()?;
+ Ok(ids)
}
- pub fn mark_read(&mut self, user: &str, msg: MessageId) {
- self.read.insert((user.to_owned(), msg));
+ pub fn mark_read(&self, user: &str, msg: MessageId) -> Result<(), Error> {
+ insert_read(&self.lock().conn, user, msg)
}
- pub fn edit(&mut self, user: &str, id: MessageId, change: Change) -> Result<(), Error> {
- let msg = self.message(id).ok_or(Error::NotFound("message"))?;
- let read_only = self.thread_account(msg.thread).is_some_and(|a| a.read_only);
- let Kind::Received { values, .. } = &msg.kind else {
+ /// Every change made to the message's values, oldest first.
+ pub fn history(&self, message: MessageId) -> Result<Vec<Event>, Error> {
+ let inner = self.lock();
+ let mut stmt = inner.conn.prepare_cached(
+ "SELECT message, user, at, event FROM history WHERE message = ?1 ORDER BY id",
+ )?;
+ let events = stmt
+ .query_map([message], |r| {
+ Ok(Event {
+ message: r.get(0)?,
+ user: r.get(1)?,
+ at: r.get(2)?,
+ text: r.get(3)?,
+ })
+ })?
+ .collect::<Result<_, _>>()?;
+ Ok(events)
+ }
+
+ pub fn edit(&self, user: &str, id: MessageId, change: Change) -> Result<(), Error> {
+ let mut inner = self.lock();
+ let tx = inner.conn.transaction()?;
+ let msg = load_message(&tx, id)?.ok_or(Error::NotFound("message"))?;
+ let read_only = load_thread_account(&tx, msg.thread)?.is_some_and(|a| a.read_only);
+ let Kind::Received { values, .. } = msg.kind else {
return Err(Error::BadRequest("sent messages have no values"));
};
- let prev = values.clone();
+ let prev = values;
let mut next = prev.clone();
let text = match change {
@@ -133,7 +305,7 @@ impl Store {
return Err(Error::Forbidden("this account is read-only"));
}
match folder {
- Some(f) if !self.folders.contains(&f) => {
+ Some(f) if !folder_exists(&tx, &f)? => {
return Err(Error::NotFound("folder"));
}
Some(f) => {
@@ -148,11 +320,7 @@ impl Store {
}
}
Change::ToggleAssignee(slug) => {
- let name = self
- .user(&slug)
- .ok_or(Error::NotFound("user"))?
- .name
- .clone();
+ let name = load_user(&tx, &slug)?.ok_or(Error::NotFound("user"))?.name;
if next.assignees.remove(&slug) {
format!("Unassigned {name}")
} else {
@@ -162,9 +330,11 @@ impl Store {
}
};
- self.set_values(id, next);
- self.undo.insert(user.to_owned(), (id, prev));
- self.flash.insert(
+ write_values(&tx, id, &next)?;
+ insert_event(&tx, id, user, self.now, &text)?;
+ tx.commit()?;
+ inner.undo.insert(user.to_owned(), (id, prev));
+ inner.flash.insert(
user.to_owned(),
Flash {
text,
@@ -174,13 +344,19 @@ impl Store {
Ok(())
}
- pub fn undo(&mut self, user: &str) -> Result<(), Error> {
- let (id, prev) = self
+ pub fn undo(&self, user: &str) -> Result<(), Error> {
+ let mut inner = self.lock();
+ let (id, prev) = inner
.undo
- .remove(user)
+ .get(user)
+ .cloned()
.ok_or(Error::BadRequest("nothing to undo"))?;
- self.set_values(id, prev);
- self.flash.insert(
+ let tx = inner.conn.transaction()?;
+ write_values(&tx, id, &prev)?;
+ insert_event(&tx, id, user, self.now, "Undone")?;
+ tx.commit()?;
+ inner.undo.remove(user);
+ inner.flash.insert(
user.to_owned(),
Flash {
text: "Undone".to_owned(),
@@ -190,43 +366,336 @@ impl Store {
Ok(())
}
- pub fn take_flash(&mut self, user: &str) -> Option<Flash> {
- self.flash.remove(user)
+ pub fn take_flash(&self, user: &str) -> Option<Flash> {
+ self.lock().flash.remove(user)
}
- pub fn add_comment(&mut self, user: &str, thread: ThreadId, text: &str) -> Result<(), Error> {
- self.thread(thread).ok_or(Error::NotFound("thread"))?;
+ pub fn add_comment(&self, user: &str, thread: ThreadId, text: &str) -> Result<(), Error> {
let text = text.trim();
if text.is_empty() {
return Err(Error::BadRequest("comment is empty"));
}
- let id = self
- .comments
- .iter()
- .map(|c| c.id)
- .max()
- .unwrap_or(0)
- .saturating_add(1);
- self.comments.push(Comment {
- id,
- thread,
- author: user.to_owned(),
- at: self.now,
- text: text.to_owned(),
- });
+ let inner = self.lock();
+ let exists: bool = inner.conn.query_row(
+ "SELECT EXISTS (SELECT 1 FROM threads WHERE id = ?1)",
+ [thread],
+ |r| r.get(0),
+ )?;
+ if !exists {
+ return Err(Error::NotFound("thread"));
+ }
+ insert_comment(&inner.conn, None, thread, user, self.now, text)?;
+ Ok(())
+ }
+
+ /// A panic mid-request can't leave SQLite half-written (transactions
+ /// roll back on drop), so a poisoned lock is safe to keep using.
+ fn lock(&self) -> MutexGuard<'_, Inner> {
+ self.inner.lock().unwrap_or_else(PoisonError::into_inner)
+ }
+}
+
+/// Writes inside one transaction, from [`Store::import`].
+pub struct Import<'a> {
+ tx: Transaction<'a>,
+}
+
+impl Import<'_> {
+ pub fn user(&self, user: &User) -> Result<(), Error> {
+ self.tx.execute(
+ "INSERT INTO users (slug, name, login) VALUES (?1, ?2, ?3)",
+ params![user.slug, user.name, user.login],
+ )?;
+ Ok(())
+ }
+
+ pub fn account(&self, account: &Account) -> Result<(), Error> {
+ self.tx.execute(
+ "INSERT INTO accounts (slug, name, address, read_only) VALUES (?1, ?2, ?3, ?4)",
+ params![
+ account.slug,
+ account.name,
+ account.address,
+ account.read_only
+ ],
+ )?;
Ok(())
}
- fn set_values(&mut self, id: MessageId, next: Values) {
- let slot = self.messages.iter_mut().find(|m| m.id == id);
- if let Some(Message {
- kind: Kind::Received { values, .. },
- ..
- }) = slot
- {
- *values = next;
+ pub fn folder(&self, name: &str) -> Result<(), Error> {
+ self.tx
+ .execute("INSERT INTO folders (name) VALUES (?1)", [name])?;
+ Ok(())
+ }
+
+ pub fn thread(&self, thread: &Thread) -> Result<(), Error> {
+ self.tx.execute(
+ "INSERT INTO threads (id, account, subject) VALUES (?1, ?2, ?3)",
+ params![thread.id, thread.account, thread.subject],
+ )?;
+ Ok(())
+ }
+
+ /// The message lands in its thread's account.
+ pub fn message(&self, m: &Message) -> Result<(), Error> {
+ let (kind, from_name, from_addr, state, folder, sent_by, sent_to) = match &m.kind {
+ Kind::Received { from, addr, values } => (
+ "received",
+ Some(from),
+ Some(addr),
+ Some(values.state),
+ values.folder.as_ref(),
+ None,
+ None,
+ ),
+ Kind::Sent { by, to } => ("sent", None, None, None, None, Some(by), Some(json(to)?)),
+ };
+ let inserted = self.tx.execute(
+ "INSERT INTO messages (id, account, message_id, thread, at, cc, bcc, body, kind,
+ from_name, from_addr, state, folder, sent_by, sent_to)
+ SELECT ?1, account, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14
+ FROM threads WHERE id = ?3",
+ params![
+ m.id,
+ m.message_id,
+ m.thread,
+ m.at,
+ json(&m.cc)?,
+ json(&m.bcc)?,
+ m.body,
+ kind,
+ from_name,
+ from_addr,
+ state,
+ folder,
+ sent_by,
+ sent_to,
+ ],
+ )?;
+ if inserted == 0 {
+ return Err(Error::NotFound("thread"));
+ }
+ if let Some(values) = m.values() {
+ write_values(&self.tx, m.id, values)?;
+ }
+ Ok(())
+ }
+
+ pub fn comment(&self, c: &Comment) -> Result<(), Error> {
+ insert_comment(&self.tx, Some(c.id), c.thread, &c.author, c.at, &c.text)
+ }
+
+ pub fn read(&self, user: &str, msg: MessageId) -> Result<(), Error> {
+ insert_read(&self.tx, user, msg)
+ }
+}
+
+impl ToSql for State {
+ fn to_sql(&self) -> rusqlite::Result<ToSqlOutput<'_>> {
+ Ok(self.slug().into())
+ }
+}
+
+impl FromSql for State {
+ fn column_result(value: ValueRef<'_>) -> FromSqlResult<Self> {
+ let slug = value.as_str()?;
+ State::from_slug(slug).ok_or_else(|| FromSqlError::Other(format!("no state {slug}").into()))
+ }
+}
+
+fn migrate(conn: &mut Connection) -> Result<(), Error> {
+ let version: i64 = conn.pragma_query_value(None, "user_version", |r| r.get(0))?;
+ for (n, sql) in (1_i64..).zip(MIGRATIONS) {
+ if n <= version {
+ continue;
}
+ let tx = conn.transaction()?;
+ tx.execute_batch(sql)?;
+ tx.pragma_update(None, "user_version", n)?;
+ tx.commit()?;
}
+ Ok(())
+}
+
+fn user_from_row(r: &Row<'_>) -> rusqlite::Result<User> {
+ Ok(User {
+ slug: r.get(0)?,
+ name: r.get(1)?,
+ login: r.get(2)?,
+ })
+}
+
+fn load_user(conn: &Connection, slug: &str) -> Result<Option<User>, Error> {
+ let user = conn
+ .query_row(
+ "SELECT slug, name, login FROM users WHERE slug = ?1",
+ [slug],
+ user_from_row,
+ )
+ .optional()?;
+ Ok(user)
+}
+
+fn load_thread_account(conn: &Connection, thread: ThreadId) -> Result<Option<Account>, Error> {
+ let account = conn
+ .query_row(
+ "SELECT a.slug, a.name, a.address, a.read_only
+ FROM accounts a JOIN threads t ON t.account = a.slug WHERE t.id = ?1",
+ [thread],
+ |r| {
+ Ok(Account {
+ slug: r.get(0)?,
+ name: r.get(1)?,
+ address: r.get(2)?,
+ read_only: r.get(3)?,
+ })
+ },
+ )
+ .optional()?;
+ Ok(account)
+}
+
+fn folder_exists(conn: &Connection, name: &str) -> Result<bool, Error> {
+ let exists = conn.query_row(
+ "SELECT EXISTS (SELECT 1 FROM folders WHERE name = ?1)",
+ [name],
+ |r| r.get(0),
+ )?;
+ Ok(exists)
+}
+
+fn load_message(conn: &Connection, id: MessageId) -> Result<Option<Message>, Error> {
+ Ok(load_messages(conn, "m.id = ?1", [id])?.into_iter().next())
+}
+
+/// Messages matching a WHERE clause over `m` (messages), `t` (their
+/// thread), and `u` (the user who sent them, if one of us did).
+fn load_messages(
+ conn: &Connection,
+ filter: &str,
+ params: impl Params,
+) -> Result<Vec<Message>, Error> {
+ let sql = format!(
+ "SELECT m.id, m.message_id, m.thread, m.at, m.cc, m.bcc, m.body, m.kind,
+ m.from_name, m.from_addr, m.state, m.folder, m.sent_by, m.sent_to,
+ (SELECT json_group_array(user)
+ FROM (SELECT user FROM assignees WHERE message = m.id ORDER BY user))
+ FROM messages m
+ JOIN threads t ON t.id = m.thread
+ LEFT JOIN users u ON u.slug = m.sent_by
+ WHERE {filter}
+ ORDER BY m.at, m.id"
+ );
+ let mut stmt = conn.prepare_cached(&sql)?;
+ let messages = stmt
+ .query_map(params, message_from_row)?
+ .collect::<Result<_, _>>()?;
+ Ok(messages)
+}
+
+fn message_from_row(r: &Row<'_>) -> rusqlite::Result<Message> {
+ let kind = if r.get::<_, String>(7)? == "sent" {
+ Kind::Sent {
+ by: r.get(12)?,
+ to: parse_json(r, 13)?,
+ }
+ } else {
+ Kind::Received {
+ from: r.get(8)?,
+ addr: r.get(9)?,
+ values: Values {
+ state: r.get(10)?,
+ folder: r.get(11)?,
+ assignees: parse_json(r, 14)?,
+ },
+ }
+ };
+ Ok(Message {
+ id: r.get(0)?,
+ message_id: r.get(1)?,
+ thread: r.get(2)?,
+ at: r.get(3)?,
+ cc: parse_json(r, 4)?,
+ bcc: parse_json(r, 5)?,
+ body: r.get(6)?,
+ kind,
+ })
+}
+
+fn write_values(conn: &Connection, id: MessageId, values: &Values) -> Result<(), Error> {
+ conn.execute(
+ "UPDATE messages SET state = ?2, folder = ?3 WHERE id = ?1",
+ params![id, values.state, values.folder],
+ )?;
+ conn.execute("DELETE FROM assignees WHERE message = ?1", [id])?;
+ for user in &values.assignees {
+ conn.execute(
+ "INSERT INTO assignees (message, user) VALUES (?1, ?2)",
+ params![id, user],
+ )?;
+ }
+ Ok(())
+}
+
+fn insert_event(
+ conn: &Connection,
+ message: MessageId,
+ user: &str,
+ at: DateTime,
+ text: &str,
+) -> Result<(), Error> {
+ conn.execute(
+ "INSERT INTO history (message, user, at, event) VALUES (?1, ?2, ?3, ?4)",
+ params![message, user, at, text],
+ )?;
+ Ok(())
+}
+
+fn insert_comment(
+ conn: &Connection,
+ id: Option<CommentId>,
+ thread: ThreadId,
+ author: &str,
+ at: DateTime,
+ text: &str,
+) -> Result<(), Error> {
+ conn.execute(
+ "INSERT INTO comments (id, thread, author, at, text) VALUES (?1, ?2, ?3, ?4, ?5)",
+ params![id, thread, author, at, text],
+ )?;
+ Ok(())
+}
+
+fn insert_read(conn: &Connection, user: &str, msg: MessageId) -> Result<(), Error> {
+ conn.execute(
+ "INSERT OR IGNORE INTO reads (user, message) VALUES (?1, ?2)",
+ params![user, msg],
+ )?;
+ Ok(())
+}
+
+/// Lists of names and addresses are stored as JSON arrays.
+fn json<T: serde::Serialize>(value: &T) -> rusqlite::Result<String> {
+ serde_json::to_string(value).map_err(|e| rusqlite::Error::ToSqlConversionFailure(e.into()))
+}
+
+fn parse_json<T: serde::de::DeserializeOwned>(r: &Row<'_>, idx: usize) -> rusqlite::Result<T> {
+ let text: String = r.get(idx)?;
+ serde_json::from_str(&text).map_err(|e| {
+ rusqlite::Error::FromSqlConversionFailure(idx, rusqlite::types::Type::Text, e.into())
+ })
+}
+
+/// Makes `%` and `_` in search text match themselves.
+fn escape_like(text: &str) -> String {
+ let mut out = String::with_capacity(text.len());
+ for c in text.chars() {
+ if matches!(c, '%' | '_' | '\\') {
+ out.push('\\');
+ }
+ out.push(c);
+ }
+ out
}
#[cfg(test)]
@@ -235,12 +704,18 @@ mod tests {
use crate::fixtures;
fn values(store: &Store, id: MessageId) -> Values {
- store.message(id).unwrap().values().unwrap().clone()
+ store
+ .message(id)
+ .unwrap()
+ .unwrap()
+ .values()
+ .unwrap()
+ .clone()
}
#[test]
- fn edits_record_undo_and_flash() {
- let mut store = fixtures::store();
+ fn edits_record_undo_flash_and_history() {
+ let store = fixtures::store().unwrap();
store.edit("sam", 4, Change::State(State::Do)).unwrap();
assert_eq!(values(&store, 4).state, State::Do);
assert_eq!(
@@ -256,11 +731,25 @@ mod tests {
assert_eq!(values(&store, 4).state, State::Inbox);
assert_eq!(store.take_flash("sam").unwrap().text, "Undone");
assert!(matches!(store.undo("sam"), Err(Error::BadRequest(_))));
+
+ let history: Vec<_> = store
+ .history(4)
+ .unwrap()
+ .into_iter()
+ .map(|e| (e.user, e.at, e.text))
+ .collect();
+ assert_eq!(
+ history,
+ [
+ ("sam".into(), store.now, "Moved to Do".into()),
+ ("sam".into(), store.now, "Undone".into()),
+ ]
+ );
}
#[test]
fn folders() {
- let mut store = fixtures::store();
+ let store = fixtures::store().unwrap();
store
.edit("sam", 4, Change::Folder(Some("Finance".into())))
.unwrap();
@@ -272,11 +761,15 @@ mod tests {
store.edit("sam", 4, Change::Folder(Some("Nope".into()))),
Err(Error::NotFound("folder"))
));
+ assert_eq!(
+ store.folders().unwrap(),
+ ["Medical", "School", "House", "Finance"]
+ );
}
#[test]
fn read_only_accounts_cannot_file() {
- let mut store = fixtures::store();
+ let store = fixtures::store().unwrap();
let eli = fixtures::ELI_PRACTICE;
assert!(matches!(
store.edit("sam", eli, Change::Folder(None)),
@@ -287,7 +780,7 @@ mod tests {
#[test]
fn assignees_toggle() {
- let mut store = fixtures::store();
+ let store = fixtures::store().unwrap();
store
.edit("alex", 4, Change::ToggleAssignee("alex".into()))
.unwrap();
@@ -306,7 +799,7 @@ mod tests {
#[test]
fn bad_edits() {
- let mut store = fixtures::store();
+ let store = fixtures::store().unwrap();
assert!(matches!(
store.edit("sam", 999, Change::State(State::Do)),
Err(Error::NotFound("message"))
@@ -319,10 +812,10 @@ mod tests {
#[test]
fn comments() {
- let mut store = fixtures::store();
- let before = store.timeline(1).len();
+ let store = fixtures::store().unwrap();
+ let before = store.timeline(1).unwrap().len();
store.add_comment("sam", 1, " Called them. ").unwrap();
- let timeline = store.timeline(1);
+ let timeline = store.timeline(1).unwrap();
assert_eq!(timeline.len(), before + 1);
assert!(matches!(timeline.last(), Some(Item::Comment(c)) if c.text == "Called them."));
assert!(matches!(
@@ -335,27 +828,228 @@ mod tests {
));
}
+ #[test]
+ fn timeline_interleaves_comments_by_time() {
+ let store = fixtures::store().unwrap();
+ let order: Vec<_> = store
+ .timeline(1)
+ .unwrap()
+ .into_iter()
+ .map(|item| match item {
+ Item::Message(m) => format!("m{}", m.id),
+ Item::Comment(c) => format!("c{}", c.id),
+ })
+ .collect();
+ assert_eq!(order, ["m1", "c1", "m2", "m3", "m4"]);
+ }
+
#[test]
fn read_tracking() {
- let mut store = fixtures::store();
- let msg = store.message(fixtures::WATER).unwrap().clone();
- assert!(store.is_unread("alex", &msg));
- store.mark_read("alex", msg.id);
- assert!(!store.is_unread("alex", &msg));
- assert!(store.is_unread("sam", &msg));
- let sent = store.message(2).unwrap().clone();
- assert!(!store.is_unread("sam", &sent));
+ let store = fixtures::store().unwrap();
+ assert!(store.unread("alex").unwrap().contains(&fixtures::WATER));
+ store.mark_read("alex", fixtures::WATER).unwrap();
+ store.mark_read("alex", fixtures::WATER).unwrap();
+ assert!(!store.unread("alex").unwrap().contains(&fixtures::WATER));
+ assert!(store.unread("sam").unwrap().contains(&fixtures::WATER));
+ // Sent messages are never unread.
+ assert!(!store.unread("sam").unwrap().contains(&2));
+ }
+
+ #[test]
+ fn messages_round_trip() {
+ let store = fixtures::store().unwrap();
+ let sent = store.message(2).unwrap().unwrap();
+ assert_eq!(sent.message_id, "2@fixtures.docket.invalid");
+ assert_eq!(sent.cc, ["Alex"]);
+ assert_eq!(sent.bcc, ["Pat Lee"]);
+ assert!(
+ matches!(&sent.kind, Kind::Sent { by, to } if by == "sam" && to == &["Northwind Roofing"])
+ );
+ let got = store.message(5).unwrap().unwrap();
+ assert_eq!(got.at, jiff::civil::date(2026, 9, 30).at(15, 30, 0, 0));
+ assert_eq!(
+ got.values(),
+ Some(&Values {
+ state: State::Do,
+ folder: Some("School".into()),
+ assignees: BTreeSet::from(["sam".to_owned()]),
+ })
+ );
+ assert!(store.message(999).unwrap().is_none());
+ }
+
+ #[test]
+ fn search_treats_wildcards_literally() {
+ let store = fixtures::store().unwrap();
+ assert!(store.messages(Filter::Search("%")).unwrap().is_empty());
+ assert!(store.messages(Filter::Search("_")).unwrap().is_empty());
+ assert_eq!(store.messages(Filter::Search("$1,840")).unwrap().len(), 1);
+ assert_eq!(escape_like(r"a%b_c\d"), r"a\%b\_c\\d");
}
#[test]
fn lookups() {
- let store = fixtures::store();
- assert_eq!(store.user_name("sam"), "Sam");
- assert_eq!(store.user_name("pat"), "pat");
+ let store = fixtures::store().unwrap();
+ let slugs: Vec<_> = store.users().unwrap().into_iter().map(|u| u.slug).collect();
+ assert_eq!(slugs, ["alex", "sam"]);
+ assert_eq!(store.user("sam").unwrap().unwrap().name, "Sam");
+ assert!(store.user("pat").unwrap().is_none());
assert_eq!(
- store.user_by_login("alex@example.com").unwrap().slug,
+ store
+ .user_by_login("alex@example.com")
+ .unwrap()
+ .unwrap()
+ .slug,
"alex"
);
- assert!(store.user_by_login("nobody").is_none());
+ assert!(store.user_by_login("nobody").unwrap().is_none());
+ assert_eq!(
+ store.thread(1).unwrap().unwrap().subject,
+ "Gutter repair estimate"
+ );
+ assert!(store.thread(999).unwrap().is_none());
+ assert!(store.thread_account(11).unwrap().unwrap().read_only);
+ assert!(store.thread_account(999).unwrap().is_none());
+ }
+
+ #[test]
+ fn importing_into_a_missing_thread_fails() {
+ let store = fixtures::store().unwrap();
+ let mut m = store.message(4).unwrap().unwrap();
+ m.id = 100;
+ m.message_id = "new@example.com".into();
+ m.thread = 999;
+ let result = store.import(|tx| tx.message(&m));
+ assert!(matches!(result, Err(Error::NotFound("thread"))));
+ assert!(store.message(100).unwrap().is_none());
+ }
+
+ #[test]
+ fn edits_survive_reopening() {
+ let dir = tempfile::tempdir().unwrap();
+ let path = camino::Utf8PathBuf::try_from(dir.path().join("docket.db")).unwrap();
+
+ let store = Store::open(&path, fixtures::now()).unwrap();
+ assert!(store.is_empty().unwrap());
+ fixtures::seed(&store).unwrap();
+ assert!(!store.is_empty().unwrap());
+ store.edit("sam", 4, Change::State(State::Wait)).unwrap();
+ store
+ .edit("sam", 4, Change::ToggleAssignee("alex".into()))
+ .unwrap();
+ store.add_comment("sam", 1, "Called them.").unwrap();
+ store.mark_read("sam", fixtures::WATER).unwrap();
+ drop(store);
+
+ let store = Store::open(&path, fixtures::now()).unwrap();
+ assert!(!store.is_empty().unwrap());
+ let v = values(&store, 4);
+ assert_eq!(v.state, State::Wait);
+ assert_eq!(v.assignees, BTreeSet::from(["alex".to_owned()]));
+ assert!(matches!(
+ store.timeline(1).unwrap().last(),
+ Some(Item::Comment(c)) if c.text == "Called them."
+ ));
+ assert!(!store.unread("sam").unwrap().contains(&fixtures::WATER));
+ assert_eq!(store.history(4).unwrap().len(), 2);
+ // Undo and toasts belong to the session that made the edit.
+ assert!(store.take_flash("sam").is_none());
+ assert!(matches!(store.undo("sam"), Err(Error::BadRequest(_))));
+ }
+
+ #[test]
+ fn rejects_bad_rows() {
+ let store = fixtures::store().unwrap();
+ // The schema refuses a received message without a state, and a
+ // folder that doesn't exist.
+ let inner = store.lock();
+ assert!(
+ inner
+ .conn
+ .execute("UPDATE messages SET state = NULL WHERE id = 4", [])
+ .is_err()
+ );
+ assert!(
+ inner
+ .conn
+ .execute("UPDATE messages SET folder = 'Nope' WHERE id = 4", [])
+ .is_err()
+ );
+ inner
+ .conn
+ .execute("UPDATE messages SET cc = 'not json' WHERE id = 4", [])
+ .unwrap();
+ drop(inner);
+ assert!(matches!(store.message(4), Err(Error::Db(_))));
+ }
+
+ fn exec(store: &Store, sql: &str) {
+ store.lock().conn.execute_batch(sql).unwrap();
+ }
+
+ #[test]
+ fn failed_writes_roll_back_the_whole_edit() {
+ let store = fixtures::store().unwrap();
+ exec(
+ &store,
+ "CREATE TRIGGER no_history BEFORE INSERT ON history BEGIN SELECT RAISE(ABORT, 'no'); END;
+ CREATE TRIGGER no_assignees BEFORE INSERT ON assignees BEGIN SELECT RAISE(ABORT, 'no'); END;",
+ );
+ // The state update goes in before the history row that fails.
+ assert!(matches!(
+ store.edit("sam", 4, Change::State(State::Do)),
+ Err(Error::Db(_))
+ ));
+ assert!(matches!(
+ store.edit("sam", 4, Change::ToggleAssignee("alex".into())),
+ Err(Error::Db(_))
+ ));
+ assert_eq!(values(&store, 4).state, State::Inbox);
+ assert!(values(&store, 4).assignees.is_empty());
+ assert!(store.history(4).unwrap().is_empty());
+ assert!(store.take_flash("sam").is_none());
+
+ // A failed undo stays undoable.
+ exec(&store, "DROP TRIGGER no_history;");
+ store.edit("sam", 4, Change::State(State::Do)).unwrap();
+ exec(
+ &store,
+ "CREATE TRIGGER no_history BEFORE INSERT ON history BEGIN SELECT RAISE(ABORT, 'no'); END;",
+ );
+ assert!(matches!(store.undo("sam"), Err(Error::Db(_))));
+ assert_eq!(values(&store, 4).state, State::Do);
+ exec(&store, "DROP TRIGGER no_history;");
+ store.undo("sam").unwrap();
+ assert_eq!(values(&store, 4).state, State::Inbox);
+ }
+
+ #[test]
+ fn database_errors_surface() {
+ let store = fixtures::store().unwrap();
+ let msg = store.message(4).unwrap().unwrap();
+ exec(&store, "PRAGMA query_only = ON;");
+ let db_err = |r: Result<(), Error>| assert!(matches!(r, Err(Error::Db(_))), "{r:?}");
+ db_err(store.edit("sam", 4, Change::State(State::Do)));
+ db_err(store.mark_read("sam", 4));
+ db_err(store.add_comment("sam", 1, "hi"));
+ db_err(store.import(|tx| tx.folder("Travel")));
+ db_err(store.import(|tx| tx.message(&msg)));
+ exec(&store, "PRAGMA query_only = OFF;");
+
+ // Each table the store reads, gone in turn.
+ exec(&store, "ALTER TABLE folders RENAME TO gone_folders;");
+ db_err(store.edit("sam", 4, Change::Folder(Some("House".into()))));
+ assert!(store.folders().is_err());
+ exec(&store, "ALTER TABLE comments RENAME TO gone_comments;");
+ assert!(store.timeline(1).is_err());
+ exec(&store, "ALTER TABLE reads RENAME TO gone_reads;");
+ assert!(store.unread("sam").is_err());
+ exec(&store, "ALTER TABLE history RENAME TO gone_history;");
+ assert!(store.history(4).is_err());
+ exec(&store, "ALTER TABLE threads RENAME TO gone_threads;");
+ db_err(store.add_comment("sam", 1, "hi"));
+ exec(&store, "ALTER TABLE users RENAME TO gone_users;");
+ assert!(store.users().is_err());
+ assert!(store.is_empty().is_err());
}
}
diff --git a/src/views.rs b/src/views.rs
index 34ad207..e2d5751 100644
--- a/src/views.rs
+++ b/src/views.rs
@@ -2,18 +2,31 @@
//! thread. Alpine handles the menus and collapsing; every change is a plain
//! form post.
+use std::collections::BTreeSet;
+
+use jiff::civil::DateTime;
use maud::{DOCTYPE, Markup, PreEscaped, html};
use crate::dates;
-use crate::lists::{self, Group, Section, View};
-use crate::model::{Comment, Kind, Message, State, User, Values};
-use crate::store::{Flash, Item, Store};
+use crate::lists::{Group, Section, View};
+use crate::model::{Comment, Kind, Message, MessageId, State, Thread, User, Values};
+use crate::store::{Flash, Item};
+/// Everything one page shows, loaded before rendering starts.
pub struct Page<'a> {
- pub store: &'a Store,
pub me: &'a User,
pub view: &'a View,
- pub selected: Option<&'a Message>,
+ pub now: DateTime,
+ pub users: Vec<User>,
+ pub folders: Vec<String>,
+ /// The sidebar's views, each with how many messages it lists.
+ pub nav: Vec<(View, usize)>,
+ pub sections: Vec<Section>,
+ /// Received messages `me` hasn't opened.
+ pub unread: BTreeSet<MessageId>,
+ pub selected: Option<MessageId>,
+ /// The selected message's thread.
+ pub thread: Option<ThreadView>,
pub flash: Option<Flash>,
/// This page's URL, for forms to return to.
pub here: &'a str,
@@ -21,6 +34,14 @@ pub struct Page<'a> {
pub dev: bool,
}
+pub struct ThreadView {
+ pub thread: Thread,
+ pub read_only: bool,
+ pub timeline: Vec<Item>,
+ /// Opens expanded, along with the selected message.
+ pub latest: MessageId,
+}
+
pub fn view_path(view: &View) -> String {
match view {
View::ForMe => "/".to_owned(),
@@ -71,12 +92,20 @@ fn badge(state: State) -> Markup {
html! { span.badge.(tone(state)) { span.mark {} (state.name()) } }
}
-fn assignee_label(store: &Store, values: &Values) -> Option<String> {
+/// The user's display name, or the slug when nobody has it.
+fn user_name<'a>(users: &'a [User], slug: &'a str) -> &'a str {
+ users
+ .iter()
+ .find(|u| u.slug == slug)
+ .map_or(slug, |u| u.name.as_str())
+}
+
+fn assignee_label(users: &[User], values: &Values) -> Option<String> {
(!values.assignees.is_empty()).then(|| {
let names: Vec<_> = values
.assignees
.iter()
- .map(|a| store.user_name(a).to_uppercase())
+ .map(|a| user_name(users, a).to_uppercase())
.collect();
format!("→ {}", names.join(", "))
})
@@ -119,7 +148,6 @@ fn back(here: &str) -> Markup {
}
fn sidebar(p: &Page<'_>) -> Markup {
- let views = std::iter::once(View::ForMe).chain(State::LANES.map(View::Lane));
html! {
nav.side {
div.brand-row {
@@ -127,10 +155,10 @@ fn sidebar(p: &Page<'_>) -> Markup {
button.search-btn type="button" aria-label="Search mail" { (PreEscaped(SEARCH_ICON)) }
}
div.nav {
- @for view in views {
- a.nav-item.on[&view == p.view] href=(view_path(&view)) {
+ @for (view, count) in &p.nav {
+ a.nav-item.on[view == p.view] href=(view_path(view)) {
span { (view.title()) }
- span.type-figure.count { (lists::count(p.store, p.me, &view)) }
+ span.type-figure.count { (count) }
}
}
}
@@ -151,7 +179,7 @@ fn whoami(p: &Page<'_>) -> Markup {
button.who-btn type="button" "@click"="open = !open" { (who) }
form.menu.up method="post" action="/dev/user" x-show="open" x-cloak {
(back(p.here))
- @for user in &p.store.users {
+ @for user in &p.users {
@let on = user.slug == p.me.slug;
button.opt.cur[on] type="submit" name="user" value=(user.slug) {
span { (user.name) }
@@ -167,23 +195,22 @@ fn whoami(p: &Page<'_>) -> Markup {
}
fn list(p: &Page<'_>) -> Markup {
- let sections = lists::sections(p.store, p.me, p.view);
html! {
section.list aria-label="Messages" {
header.list-head {
h1 { (p.view.title()) }
}
- @if sections.is_empty() {
+ @if p.sections.is_empty() {
p.empty { "Nothing here." }
}
- @for section in §ions {
+ @for section in &p.sections {
(list_section(p, section))
}
}
}
}
-fn list_section(p: &Page<'_>, section: &Section<'_>) -> Markup {
+fn list_section(p: &Page<'_>, section: &Section) -> Markup {
html! {
div {
@if !section.head.is_empty() {
@@ -202,8 +229,8 @@ fn list_section(p: &Page<'_>, section: &Section<'_>) -> Markup {
}
/// A thread with several messages listed: the subject heads a row per message.
-fn thread_group(p: &Page<'_>, group: &Group<'_>, rows: &[&Message], compact: bool) -> Markup {
- let unread = rows.iter().any(|m| p.store.is_unread(&p.me.slug, m));
+fn thread_group(p: &Page<'_>, group: &Group, rows: &[Message], compact: bool) -> Markup {
+ let unread = rows.iter().any(|m| p.unread.contains(&m.id));
html! {
div.group.unread[unread] {
span.subj { (group.thread.subject) }
@@ -218,15 +245,15 @@ fn thread_group(p: &Page<'_>, group: &Group<'_>, rows: &[&Message], compact: boo
/// A thread with one message listed collapses into a single row: subject and
/// time, then sender and snippet.
-fn solo(p: &Page<'_>, group: &Group<'_>, m: &Message, compact: bool) -> Markup {
- let selected = p.selected.is_some_and(|s| s.id == m.id);
- let unread = p.store.is_unread(&p.me.slug, m);
+fn solo(p: &Page<'_>, group: &Group, m: &Message, compact: bool) -> Markup {
+ let selected = p.selected == Some(m.id);
+ let unread = p.unread.contains(&m.id);
html! {
a.row.solo.sel[selected].unread[unread] href=(select_path(p.view, m.id)) {
span.subj { (group.thread.subject) }
- span.type-figure.age { (dates::short(p.store.now, m.at)) }
+ span.type-figure.age { (dates::short(p.now, m.at)) }
span.line {
- span.from { (sender(p.store, m)) }
+ span.from { (sender(&p.users, m)) }
@if !compact {
span.snip { " — " (m.body) }
}
@@ -235,20 +262,20 @@ fn solo(p: &Page<'_>, group: &Group<'_>, m: &Message, compact: bool) -> Markup {
}
}
-fn sender(store: &Store, m: &Message) -> String {
+fn sender(users: &[User], m: &Message) -> String {
match &m.kind {
Kind::Received { from, .. } => from.clone(),
- Kind::Sent { by, to } => format!("{} → {}", store.user_name(by), to.join(", ")),
+ Kind::Sent { by, to } => format!("{} → {}", user_name(users, by), to.join(", ")),
}
}
fn row(p: &Page<'_>, m: &Message, compact: bool) -> Markup {
- let selected = p.selected.is_some_and(|s| s.id == m.id);
- let unread = p.store.is_unread(&p.me.slug, m);
+ let selected = p.selected == Some(m.id);
+ let unread = p.unread.contains(&m.id);
html! {
a.row.sel[selected].unread[unread] href=(select_path(p.view, m.id)) {
- span.from { (sender(p.store, m)) }
- span.type-figure.age { (dates::short(p.store.now, m.at)) }
+ span.from { (sender(&p.users, m)) }
+ span.type-figure.age { (dates::short(p.now, m.at)) }
@if !compact {
span.snip { (m.body) }
}
@@ -259,8 +286,8 @@ fn row(p: &Page<'_>, m: &Message, compact: bool) -> Markup {
fn reader(p: &Page<'_>) -> Markup {
html! {
main.reader {
- @if let Some(selected) = p.selected {
- (thread(p, selected))
+ @if let (Some(selected), Some(t)) = (p.selected, &p.thread) {
+ (thread(p, t, selected))
} @else {
p.empty { "No thread open." }
}
@@ -268,35 +295,27 @@ fn reader(p: &Page<'_>) -> Markup {
}
}
-fn thread(p: &Page<'_>, selected: &Message) -> Markup {
- let store = p.store;
- let subject = store
- .thread(selected.thread)
- .map_or("", |t| t.subject.as_str());
- let read_only = store
- .thread_account(selected.thread)
- .is_some_and(|a| a.read_only);
- let latest = store.thread_messages(selected.thread).last().map(|m| m.id);
+fn thread(p: &Page<'_>, t: &ThreadView, selected: MessageId) -> Markup {
html! {
article.thread {
header.thread-head {
- h2 { (subject) }
- @if read_only {
+ h2 { (t.thread.subject) }
+ @if t.read_only {
span.type-label { "Read-only" }
}
}
div.chain {
- @for item in store.timeline(selected.thread) {
+ @for item in &t.timeline {
@match item {
- Item::Comment(c) => (comment(store, c)),
+ Item::Comment(c) => (comment(p, c)),
Item::Message(m) => {
- @let open = m.id == selected.id || Some(m.id) == latest;
- (message(p, m, open, m.id == selected.id, read_only))
+ @let open = m.id == selected || m.id == t.latest;
+ (message(p, m, open, m.id == selected, t.read_only))
}
}
}
}
- form.comment-box method="post" action=(format!("/threads/{}/comments", selected.thread)) {
+ form.comment-box method="post" action=(format!("/threads/{}/comments", t.thread.id)) {
(back(p.here))
input.field type="text" name="text" required placeholder="Add an internal comment to this thread"
aria-label="Internal comment" autocomplete="off";
@@ -306,24 +325,23 @@ fn thread(p: &Page<'_>, selected: &Message) -> Markup {
}
}
-fn comment(store: &Store, c: &Comment) -> Markup {
+fn comment(p: &Page<'_>, c: &Comment) -> Markup {
html! {
div.item.comment {
div.body {
- span.author { b { (store.user_name(&c.author)) } }
+ span.author { b { (user_name(&p.users, &c.author)) } }
span.text { (c.text) }
}
- div.gutter { span.type-figure.date { (dates::short(store.now, c.at)) } }
+ div.gutter { span.type-figure.date { (dates::short(p.now, c.at)) } }
}
}
}
fn message(p: &Page<'_>, m: &Message, open: bool, selected: bool, read_only: bool) -> Markup {
- let store = p.store;
let (who, addr) = match &m.kind {
Kind::Received { from, addr, .. } => (from.clone(), format!("<{addr}>")),
Kind::Sent { by, to } => (
- store.user_name(by).to_owned(),
+ user_name(&p.users, by).to_owned(),
format!("→ {}", to.join(", ")),
),
};
@@ -353,7 +371,7 @@ fn message(p: &Page<'_>, m: &Message, open: bool, selected: bool, read_only: boo
}
div.gutter {
span.when {
- span.type-figure.date { (dates::short(store.now, m.at)) }
+ span.type-figure.date { (dates::short(p.now, m.at)) }
a.icon-btn href=(fastmail_url(m)) target="_blank" rel="noopener"
title="Open in Fastmail" aria-label="Open in Fastmail" { (PreEscaped(EXTERNAL_ICON)) }
}
@@ -399,16 +417,16 @@ fn values_controls(p: &Page<'_>, id: u32, values: &Values, read_only: bool) -> M
button.opt.cur[values.folder.is_none()] type="submit" name="folder" value="" {
span { "No folder" } (check(values.folder.is_none()))
}
- @for folder in &p.store.folders {
+ @for folder in &p.folders {
@let on = values.folder.as_ref() == Some(folder);
button.opt.cur[on] type="submit" name="folder" value=(folder) {
span { (folder) } (check(on))
}
}
};
- let assigned = assignee_label(p.store, values);
+ let assigned = assignee_label(&p.users, values);
let people = html! {
- @for user in &p.store.users {
+ @for user in &p.users {
@let on = values.assignees.contains(&user.slug);
button.opt.cur[on] type="submit" name="user" value=(user.slug) {
span { (user.name) } (check(on))
@@ -461,10 +479,10 @@ mod tests {
#[test]
fn fastmail_links_search_by_message_id() {
- let store = crate::fixtures::store();
- let m = store.message(4).unwrap();
+ let store = crate::fixtures::store().unwrap();
+ let m = store.message(4).unwrap().unwrap();
assert_eq!(
- fastmail_url(m),
+ fastmail_url(&m),
"https://app.fastmail.com/mail/search:msgid%3A%3C4%40fixtures.docket.invalid%3E"
);
}
diff --git a/tests/api.rs b/tests/api.rs
index e5ab06e..b76b080 100644
--- a/tests/api.rs
+++ b/tests/api.rs
@@ -8,7 +8,7 @@ const SAM: &str = "sam@example.com";
const ALEX: &str = "alex@example.com";
async fn spawn_with(dev: bool) -> SocketAddr {
- let app = router(AppState::new(docket::fixtures::store(), dev));
+ let app = router(AppState::new(docket::fixtures::store().unwrap(), dev));
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
@@ -368,7 +368,7 @@ async fn dev_mode_picks_a_user_from_a_cookie() {
#[tokio::test]
async fn dev_mode_with_no_users_is_unauthenticated() {
- let store = docket::store::Store::default();
+ let store = docket::store::Store::open_in_memory(docket::fixtures::now()).unwrap();
let app = router(AppState::new(store, true));
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();