Add new data model

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)
 );

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<DocumentId, String>;
 
-#[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<Utc>,
     pub content: Vec<u8>,
-    pub is_deleted: bool,
     pub user_id: UserId,
     pub device_id: DeviceId,
 }
 
-impl PartialEq<Self> 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<Utc>,
-    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<StoredDocumentVersion> 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<StoredDocumentVersion> 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<StoredDocumentVersion> 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<string, string>")]
+    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<Utc>,
-    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<DocumentVersionWithoutContent>,
 }
 
-impl From<StoredDocumentVersion> 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<EventRecord>,
 }

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<VaultUpdateId> {
+        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(())
+    }
+}

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<Option<(Vec<u8>, 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<Option<StoredDocumentVersion>> {
+        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<Option<StoredDocumentVersion>> {
+        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<FileManifest> {
+        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<VaultUpdateId> {
+        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<VaultSnapshot> {
+        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<EventBatch> {
+        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<String> = 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::<Result<_, _>>()?;
+
+        Ok(EventBatch {
+            head_event_id,
+            events,
+        })
+    }
+}