diff --git a/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql b/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql index 0b3076c6..504f89ba 100644 --- a/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql +++ b/sync-server/src/app_state/database/migrations/20241207143519_bootstrap.sql @@ -1,30 +1,34 @@ -CREATE TABLE IF NOT EXISTS documents ( - vault_update_id INTEGER NOT NULL PRIMARY KEY, +CREATE TABLE events ( + event_id INTEGER PRIMARY KEY AUTOINCREMENT CHECK(event_id <= 9007199254740991), -- JavaScript’s Number.MAX_SAFE_INTEGER + request_id TEXT NOT NULL UNIQUE, + request_fingerprint BLOB NOT NULL, + event_json TEXT NOT NULL +); + +CREATE TABLE documents ( + vault_update_id INTEGER PRIMARY KEY REFERENCES events(event_id), document_id TEXT NOT NULL, - relative_path TEXT NOT NULL, - updated_date TIMESTAMP NOT NULL, + updated_date TEXT NOT NULL, content BLOB NOT NULL, - is_deleted BOOLEAN NOT NULL, user_id TEXT NOT NULL, device_id TEXT NOT NULL ); -CREATE VIEW IF NOT EXISTS latest_document_versions AS -SELECT d.* -FROM documents d -INNER JOIN ( - SELECT MAX(vault_update_id) AS max_version_id - FROM documents - GROUP BY document_id -) max_versions -ON d.vault_update_id = max_versions.max_version_id; +CREATE INDEX documents_by_id ON documents(document_id); -CREATE INDEX IF NOT EXISTS idx_documents_vault_id_relative_path -ON documents (relative_path); +CREATE VIEW latest_document_versions AS + SELECT d.* FROM documents d JOIN ( + SELECT document_id, MAX(vault_update_id) AS version FROM documents GROUP BY document_id + ) heads ON d.vault_update_id = heads.version; --- The immutable version and its acknowledgement commit in one transaction. -CREATE TABLE push_acknowledgements ( - request_id TEXT NOT NULL PRIMARY KEY, - request_fingerprint BLOB NOT NULL, - vault_update_id INTEGER NOT NULL REFERENCES documents(vault_update_id) +CREATE TABLE file_manifests ( + file_manifest_id INTEGER PRIMARY KEY REFERENCES events(event_id) +); + +CREATE TABLE file_manifest_entries ( + file_manifest_id INTEGER NOT NULL REFERENCES file_manifests(file_manifest_id), + document_id TEXT NOT NULL, + path TEXT NOT NULL, + PRIMARY KEY (file_manifest_id, document_id), + UNIQUE (file_manifest_id, path) ); diff --git a/sync-server/src/app_state/database/models.rs b/sync-server/src/app_state/database/models.rs index 24c0c370..81cb6343 100644 --- a/sync-server/src/app_state/database/models.rs +++ b/sync-server/src/app_state/database/models.rs @@ -1,91 +1,125 @@ use base64::{Engine as _, engine::general_purpose::STANDARD}; use chrono::{DateTime, Utc}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; use ts_rs::TS; pub type VaultId = String; pub type VaultUpdateId = i64; - pub type DocumentId = uuid::Uuid; pub type UserId = String; pub type DeviceId = String; +pub type FileManifestEntries = BTreeMap; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, sqlx::FromRow)] pub struct StoredDocumentVersion { pub vault_update_id: VaultUpdateId, + #[sqlx(try_from = "uuid::fmt::Hyphenated")] pub document_id: DocumentId, - pub relative_path: String, pub updated_date: DateTime, pub content: Vec, - pub is_deleted: bool, pub user_id: UserId, pub device_id: DeviceId, } -impl PartialEq for StoredDocumentVersion { - fn eq(&self, other: &Self) -> bool { - self.vault_update_id == other.vault_update_id - } -} - -#[derive(TS, Debug, Clone, Serialize)] +#[derive(TS, Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] +#[ts(export)] pub struct DocumentVersionWithoutContent { - #[ts(as = "i32")] + #[ts(as = "f64")] pub vault_update_id: VaultUpdateId, - pub document_id: DocumentId, - pub relative_path: String, pub updated_date: DateTime, - pub is_deleted: bool, pub user_id: UserId, pub device_id: DeviceId, - - #[ts(as = "i32")] - pub content_size: u64, + pub content_size: usize, } -impl From for DocumentVersionWithoutContent { - fn from(value: StoredDocumentVersion) -> Self { +impl From<&StoredDocumentVersion> for DocumentVersionWithoutContent { + fn from(v: &StoredDocumentVersion) -> Self { Self { - vault_update_id: value.vault_update_id, - document_id: value.document_id, - relative_path: value.relative_path, - updated_date: value.updated_date, - is_deleted: value.is_deleted, - user_id: value.user_id, - device_id: value.device_id, - content_size: value.content.len() as u64, + vault_update_id: v.vault_update_id, + document_id: v.document_id, + updated_date: v.updated_date, + user_id: v.user_id.clone(), + device_id: v.device_id.clone(), + content_size: v.content.len(), } } } +impl From for DocumentVersionWithoutContent { + fn from(v: StoredDocumentVersion) -> Self { + Self::from(&v) + } +} + +#[derive(TS, Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[ts(export)] +pub struct DocumentVersion { + #[serde(flatten)] + pub metadata: DocumentVersionWithoutContent, + pub content_base64: String, +} +impl From for DocumentVersion { + fn from(v: StoredDocumentVersion) -> Self { + Self { + metadata: (&v).into(), + content_base64: STANDARD.encode(&v.content), + } + } +} + +#[derive(TS, Debug, Clone, Serialize, Deserialize, Default)] +#[serde(rename_all = "camelCase")] +#[ts(export)] +pub struct FileManifest { + #[ts(as = "f64")] + pub file_manifest_id: VaultUpdateId, + #[ts(type = "Record")] + pub entries: FileManifestEntries, +} + +#[derive(TS, Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "camelCase")] +#[ts(export)] +pub enum VaultEvent { + Content { + document: DocumentVersionWithoutContent, + }, + FileManifest { + #[serde(rename = "fileManifest")] + #[ts(rename = "fileManifest")] + file_manifest: FileManifest, + }, +} + +#[derive(TS, Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +#[ts(export)] +pub struct EventRecord { + #[ts(as = "f64")] + pub event_id: VaultUpdateId, + pub request_id: uuid::Uuid, + #[serde(flatten)] + pub event: VaultEvent, +} #[derive(TS, Debug, Clone, Serialize)] #[serde(rename_all = "camelCase")] -pub struct DocumentVersion { - #[ts(as = "i32")] - pub vault_update_id: VaultUpdateId, - - pub document_id: DocumentId, - pub relative_path: String, - pub updated_date: DateTime, - pub content_base64: String, - pub is_deleted: bool, - pub user_id: UserId, - pub device_id: DeviceId, +#[ts(export)] +pub struct VaultSnapshot { + #[ts(as = "f64")] + pub head_event_id: VaultUpdateId, + pub file_manifest: FileManifest, + pub documents: Vec, } -impl From for DocumentVersion { - fn from(value: StoredDocumentVersion) -> Self { - Self { - vault_update_id: value.vault_update_id, - document_id: value.document_id, - relative_path: value.relative_path, - updated_date: value.updated_date, - content_base64: STANDARD.encode(&value.content), - is_deleted: value.is_deleted, - user_id: value.user_id, - device_id: value.device_id, - } - } +#[derive(TS, Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +#[ts(export)] +pub struct EventBatch { + #[ts(as = "f64")] + pub head_event_id: VaultUpdateId, + pub events: Vec, } diff --git a/sync-server/src/app_state/database/mutations.rs b/sync-server/src/app_state/database/mutations.rs new file mode 100644 index 00000000..59383789 --- /dev/null +++ b/sync-server/src/app_state/database/mutations.rs @@ -0,0 +1,68 @@ +use super::{Database, Transaction, models::*}; +use anyhow::Result; + +impl Database { + pub async fn allocate_event( + tx: &mut Transaction<'_>, + request_id: uuid::Uuid, + fingerprint: &[u8], + ) -> Result { + Ok(sqlx::query( + "INSERT INTO events(request_id, request_fingerprint, event_json) VALUES (?, ?, '')", + ) + .bind(request_id.to_string()) + .bind(fingerprint) + .execute(&mut **tx) + .await? + .last_insert_rowid()) + } + + pub async fn insert_document_version( + tx: &mut Transaction<'_>, + version: &StoredDocumentVersion, + ) -> Result<()> { + sqlx::query("INSERT INTO documents(vault_update_id, document_id, updated_date, content, user_id, device_id) VALUES (?, ?, ?, ?, ?, ?)") + .bind(version.vault_update_id) + .bind(version.document_id.hyphenated()) + .bind(version.updated_date) + .bind(&version.content) + .bind(&version.user_id) + .bind(&version.device_id) + .execute(&mut **tx) + .await?; + + Ok(()) + } + + pub async fn insert_file_manifest( + tx: &mut Transaction<'_>, + file_manifest: &FileManifest, + ) -> Result<()> { + sqlx::query("INSERT INTO file_manifests(file_manifest_id) VALUES (?)") + .bind(file_manifest.file_manifest_id) + .execute(&mut **tx) + .await?; + + for (document_id, path) in &file_manifest.entries { + sqlx::query( + "INSERT INTO file_manifest_entries(file_manifest_id, document_id, path) VALUES (?, ?, ?)", + ) + .bind(file_manifest.file_manifest_id) + .bind(document_id.hyphenated()) + .bind(path) + .execute(&mut **tx) + .await?; + } + + Ok(()) + } + + pub async fn write_event(tx: &mut Transaction<'_>, event: &EventRecord) -> Result<()> { + sqlx::query("UPDATE events SET event_json = ? WHERE event_id = ?") + .bind(serde_json::to_string(event)?) + .bind(event.event_id) + .execute(&mut **tx) + .await?; + Ok(()) + } +} diff --git a/sync-server/src/app_state/database/queries.rs b/sync-server/src/app_state/database/queries.rs new file mode 100644 index 00000000..edc77801 --- /dev/null +++ b/sync-server/src/app_state/database/queries.rs @@ -0,0 +1,140 @@ +use super::{Database, Transaction, models::*}; +use anyhow::{Result, ensure}; + +impl Database { + pub async fn get_request_event( + tx: &mut Transaction<'_>, + request_id: uuid::Uuid, + ) -> Result, String)>> { + Ok(sqlx::query_as( + "SELECT request_fingerprint, event_json FROM events WHERE request_id = ?", + ) + .bind(request_id.to_string()) + .fetch_optional(&mut **tx) + .await?) + } + + pub async fn get_latest_document_version( + &self, + vault: &VaultId, + id: &DocumentId, + transaction: Option<&mut Transaction<'_>>, + ) -> Result> { + let query = sqlx::query_as::<_, StoredDocumentVersion>( + "SELECT * FROM documents WHERE document_id = ? ORDER BY vault_update_id DESC LIMIT 1", + ) + .bind(id.hyphenated()); + + Ok(if let Some(tx) = transaction { + query.fetch_optional(&mut **tx).await? + } else { + query + .fetch_optional(&self.get_connection_pool(vault).await?) + .await? + }) + } + + pub async fn get_document_version( + &self, + vault: &VaultId, + version: VaultUpdateId, + transaction: Option<&mut Transaction<'_>>, + ) -> Result> { + let query = sqlx::query_as::<_, StoredDocumentVersion>( + "SELECT * FROM documents WHERE vault_update_id = ?", + ) + .bind(version); + + Ok(if let Some(tx) = transaction { + query.fetch_optional(&mut **tx).await? + } else { + query + .fetch_optional(&self.get_connection_pool(vault).await?) + .await? + }) + } + + pub async fn current_file_manifest(tx: &mut Transaction<'_>) -> Result { + let Some(file_manifest_id) = sqlx::query_scalar::<_, VaultUpdateId>( + "SELECT file_manifest_id FROM file_manifests ORDER BY file_manifest_id DESC LIMIT 1", + ) + .fetch_optional(&mut **tx) + .await? + else { + return Ok(FileManifest::default()); + }; + + let entries = sqlx::query_as::<_, (uuid::fmt::Hyphenated, String)>( + "SELECT document_id, path FROM file_manifest_entries WHERE file_manifest_id = ?", + ) + .bind(file_manifest_id) + .fetch_all(&mut **tx) + .await? + .into_iter() + .map(|(document_id, path)| (document_id.into_uuid(), path)) + .collect(); + + Ok(FileManifest { + file_manifest_id, + entries, + }) + } + + pub async fn latest_event_id(tx: &mut Transaction<'_>) -> Result { + Ok( + sqlx::query_scalar("SELECT COALESCE(MAX(event_id), 0) FROM events") + .fetch_one(&mut **tx) + .await?, + ) + } + + pub async fn vault_snapshot(&self, vault: &VaultId) -> Result { + let mut tx = self.create_readonly_transaction(vault).await?; + let head_event_id = Self::latest_event_id(&mut tx).await?; + let file_manifest = Self::current_file_manifest(&mut tx).await?; + + let versions = sqlx::query_as::<_, StoredDocumentVersion>( + "SELECT d.* FROM latest_document_versions d + JOIN file_manifest_entries m ON m.document_id = d.document_id + WHERE m.file_manifest_id = ? ORDER BY d.vault_update_id", + ) + .bind(file_manifest.file_manifest_id) + .fetch_all(&mut *tx) + .await?; + + let documents = versions.into_iter().map(Into::into).collect(); + + Ok(VaultSnapshot { + head_event_id, + file_manifest, + documents, + }) + } + + pub async fn events_after(&self, vault: &VaultId, after: VaultUpdateId) -> Result { + let mut tx = self.create_readonly_transaction(vault).await?; + let head_event_id = Self::latest_event_id(&mut tx).await?; + + ensure!( + (0..=head_event_id).contains(&after), + "Event cursor is outside this vault's history" + ); + + let rows: Vec = sqlx::query_scalar( + "SELECT event_json FROM events WHERE event_id > ? ORDER BY event_id", + ) + .bind(after) + .fetch_all(&mut *tx) + .await?; + + let events = rows + .iter() + .map(|json| serde_json::from_str(json)) + .collect::>()?; + + Ok(EventBatch { + head_event_id, + events, + }) + } +}