andras/reconcile
1
0
Fork 0
Code Issues Pull requests 5 Projects Releases Packages Wiki Activity Actions

Go from per document to per vault version ids to allow "since" queries

Browse source
This commit is contained in:
Andras Schmelczer 2024-12-14 17:02:44 +00:00
parent fe8d236948
commit 70471f5142
No known key found for this signature in database
GPG key ID: FC8F2C3D3D1A718C
6 changed files with 127 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

76
backend/sync_server/src/database.rs
View file

@ -1,7 +1,7 @@
use std::str::FromStr;
use anyhow::{Context, Result};
use models::{DocumentVersionId, DocumentVersionWithoutContent, StoredDocumentVersion, VaultId};
use models::{DocumentVersionWithoutContent, StoredDocumentVersion, VaultId, VaultUpdateId};
use sqlx::{sqlite::SqliteConnectOptions, types::chrono::Utc};
pub mod models;
use sqlx::{sqlite::SqlitePoolOptions, Pool, Sqlite};
@ -64,8 +64,8 @@ impl Database {
r#"
select
vault_id,
vault_update_id,
relative_path,
version_id,
created_date as "created_date: chrono::DateTime<Utc>",
updated_date as "updated_date: chrono::DateTime<Utc>",
is_deleted
@ -83,6 +83,62 @@ impl Database {
.context("Cannot fetch latest documents")
}
pub async fn get_max_update_id_in_vault(
&self,
vault: &VaultId,
transaction: Option<&mut Transaction<'_>>,
) -> Result<i64> {
let query = sqlx::query!(
r#"
select coalesce(max(vault_update_id), 0) as max_vault_update_id
from documents
where vault_id = ?
"#,
vault
);
if let Some(transaction) = transaction {
query.fetch_one(&mut **transaction).await
} else {
query.fetch_one(&self.connection_pool).await
}
.map(|row| row.max_vault_update_id)
.context("Cannot fetch max update id in vault")
}
pub async fn get_latest_documents_since(
&self,
vault: &VaultId,
vault_update_id: VaultUpdateId,
transaction: Option<&mut Transaction<'_>>,
) -> Result<Vec<DocumentVersionWithoutContent>> {
let query = sqlx::query_as!(
DocumentVersionWithoutContent,
r#"
select
vault_id,
vault_update_id,
relative_path,
created_date as "created_date: chrono::DateTime<Utc>",
updated_date as "updated_date: chrono::DateTime<Utc>",
is_deleted
from latest_document_versions
where vault_id = ? and vault_update_id > ?
"#,
vault,
vault_update_id
);
if let Some(transaction) = transaction {
query.fetch_all(&mut **transaction).await
} else {
query.fetch_all(&self.connection_pool).await
}
.with_context(|| {
format!("Cannot fetch latest documents since vault_update_id {vault_update_id}")
})
}
pub async fn get_latest_document(
&self,
vault: &VaultId,
@ -94,8 +150,8 @@ impl Database {
r#"
select
vault_id,
vault_update_id,
relative_path,
version_id,
created_date as "created_date: chrono::DateTime<Utc>",
updated_date as "updated_date: chrono::DateTime<Utc>",
content,
@ -118,8 +174,7 @@ impl Database {
pub async fn get_document_version(
&self,
vault: &VaultId,
relative_path: &str,
version: &DocumentVersionId,
vault_update_id: VaultUpdateId,
transaction: Option<&mut Transaction<'_>>,
) -> Result<Option<StoredDocumentVersion>> {
let query = sqlx::query_as!(
@ -127,17 +182,16 @@ impl Database {
r#"
select
vault_id,
vault_update_id,
relative_path,
version_id,
created_date as "created_date: chrono::DateTime<Utc>",
updated_date as "updated_date: chrono::DateTime<Utc>",
content,
is_deleted
from documents
where vault_id = ? and relative_path = ? and version_id = ?"#,
where vault_id = ? and vault_update_id = ?"#,
vault,
relative_path,
version
vault_update_id
);
if let Some(transaction) = transaction {
@ -157,8 +211,8 @@ impl Database {
r#"
insert into documents (
vault_id,
vault_update_id,
relative_path,
version_id,
created_date,
updated_date,
content,
@ -167,8 +221,8 @@ impl Database {
values (?, ?, ?, ?, ?, ?, ?)
"#,
version.vault_id,
version.vault_update_id,
version.relative_path,
version.version_id,
version.created_date,
version.updated_date,
version.content,

9
backend/sync_server/src/database/migrations/20241207143519_bootstrap.sql
View file

@ -1,25 +1,24 @@
CREATE TABLE IF NOT EXISTS documents (
vault_id TEXT NOT NULL,
vault_update_id INTEGER NOT NULL,
relative_path TEXT NOT NULL,
version_id INTEGER NOT NULL,
created_date TIMESTAMP NOT NULL,
updated_date TIMESTAMP NOT NULL,
content BLOB NOT NULL,
is_deleted BOOLEAN NOT NULL,
PRIMARY KEY (vault_id, relative_path, version_id)
PRIMARY KEY (vault_id, vault_update_id)
);
CREATE VIEW IF NOT EXISTS latest_document_versions AS
SELECT d.*
FROM documents d
INNER JOIN (
SELECT vault_id, relative_path, MAX(version_id) AS max_version_id
SELECT vault_id, MAX(vault_update_id) AS max_version_id
FROM documents
GROUP BY vault_id, relative_path
) max_versions
ON d.vault_id = max_versions.vault_id
AND d.relative_path = max_versions.relative_path
AND d.version_id = max_versions.max_version_id;
AND d.vault_update_id = max_versions.max_version_id;
CREATE INDEX IF NOT EXISTS idx_documents_vault_doc
ON documents (vault_id, relative_path);

19
backend/sync_server/src/database/models.rs
View file

@ -4,13 +4,13 @@ use serde::Serialize;
use sync_lib::bytes_to_base64;
pub type VaultId = String;
pub type DocumentVersionId = i64;
pub type VaultUpdateId = i64;
#[derive(Debug, Clone)]
pub struct StoredDocumentVersion {
pub vault_id: VaultId,
pub vault_update_id: VaultUpdateId,
pub relative_path: String,
pub version_id: DocumentVersionId,
pub created_date: DateTime<Utc>,
pub updated_date: DateTime<Utc>,
pub content: Vec<u8>,
@ -21,8 +21,8 @@ pub struct StoredDocumentVersion {
#[serde(rename_all = "camelCase")]
pub struct DocumentVersionWithoutContent {
pub vault_id: VaultId,
pub vault_update_id: VaultUpdateId,
pub relative_path: String,
pub version_id: DocumentVersionId,
pub created_date: DateTime<Utc>,
pub updated_date: DateTime<Utc>,
pub is_deleted: bool,
@ -32,8 +32,8 @@ impl From<StoredDocumentVersion> for DocumentVersionWithoutContent {
fn from(value: StoredDocumentVersion) -> Self {
Self {
vault_id: value.vault_id,
vault_update_id: value.vault_update_id,
relative_path: value.relative_path,
version_id: value.version_id,
created_date: value.created_date,
updated_date: value.updated_date,
is_deleted: value.is_deleted,
@ -41,19 +41,12 @@ impl From<StoredDocumentVersion> for DocumentVersionWithoutContent {
}
}
#[derive(Debug, Clone, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct PingResponse {
pub server_version: String,
pub is_authenticated: bool,
}
#[derive(Debug, Clone, Serialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct DocumentVersion {
pub vault_id: VaultId,
pub vault_update_id: VaultUpdateId,
pub relative_path: String,
pub version_id: DocumentVersionId,
pub created_date: DateTime<Utc>,
pub updated_date: DateTime<Utc>,
pub content_base64: String,
@ -64,8 +57,8 @@ impl From<StoredDocumentVersion> for DocumentVersion {
fn from(value: StoredDocumentVersion) -> Self {
Self {
vault_id: value.vault_id,
vault_update_id: value.vault_update_id,
relative_path: value.relative_path,
version_id: value.version_id,
created_date: value.created_date,
updated_date: value.updated_date,
content_base64: bytes_to_base64(&value.content),

19
backend/sync_server/src/server/delete_document.rs
View file

@ -1,4 +1,4 @@
use anyhow::{anyhow, Context};
use anyhow::Context;
use axum::{
extract::{Path, State},
Json,
@ -14,7 +14,7 @@ use super::{auth::auth, requests::DeleteDocumentVersion};
use crate::{
app_state::AppState,
database::models::{StoredDocumentVersion, VaultId},
errors::{not_found_error, server_error, SyncServerError},
errors::{server_error, SyncServerError},
};
// This is required for aide to infer the path parameter types and names
@ -42,23 +42,16 @@ pub async fn delete_document(
.await
.map_err(server_error)?;
let latest_version = state
let last_update_id = state
.database
.get_latest_document(&vault_id, &relative_path, Some(&mut transaction))
.get_max_update_id_in_vault(&vault_id, Some(&mut transaction))
.await
.map_err(server_error)?
.map(Ok)
.unwrap_or_else(|| {
Err(not_found_error(anyhow!(
"Latest document version of document `{}` not found",
relative_path
)))
})?;
.map_err(server_error)?;
let new_version = StoredDocumentVersion {
vault_id,
vault_update_id: last_update_id + 1,
relative_path,
version_id: latest_version.version_id + 1,
content: vec![],
created_date: request.created_date,
updated_date: chrono::Utc::now(),

4
backend/sync_server/src/server/requests.rs
View file

@ -2,12 +2,12 @@ use chrono::{DateTime, Utc};
use schemars::JsonSchema;
use serde::{self, Deserialize};
use crate::database::models::DocumentVersionId;
use crate::database::models::VaultUpdateId;
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct UpdateDocumentVersion {
pub parent_version_id: Option<DocumentVersionId>,
pub parent_version_id: Option<VaultUpdateId>,
pub created_date: DateTime<Utc>,
pub content_base64: String,
}

53
backend/sync_server/src/server/update_document.rs
View file

@ -7,6 +7,7 @@ use axum_extra::{
headers::{authorization::Bearer, Authorization},
TypedHeader,
};
use log::info;
use schemars::JsonSchema;
use serde::Deserialize;
use sync_lib::{base64_to_bytes, merge};
@ -36,10 +37,11 @@ pub async fn update_document(
Json(request): Json<UpdateDocumentVersion>,
) -> Result<Json<DocumentVersion>, SyncServerError> {
auth(&state, auth_header.token())?;
let parent_content = if let Some(parent_version_id) = request.parent_version_id {
state
.database
.get_document_version(&vault_id, &relative_path, &parent_version_id, None)
.get_document_version(&vault_id, parent_version_id, None)
.await
.map_err(server_error)?
.map(Ok)
@ -55,26 +57,59 @@ pub async fn update_document(
Ok(Vec::default())
}?;
let content_bytes = base64_to_bytes(&request.content_base64)
.context("Failed to decode base64 content in request")
.map_err(client_error)?;
let mut transaction = state
.database
.create_transaction()
.await
.map_err(server_error)?;
let mut last_update_id = state
.database
.get_max_update_id_in_vault(&vault_id, Some(&mut transaction))
.await
.map_err(server_error)?;
let latest_version = state
.database
.get_latest_document(&vault_id, &relative_path, Some(&mut transaction))
.await
.map_err(server_error)?;
let content_bytes = base64_to_bytes(&request.content_base64)
.context("Failed to decode base64 content in request")
.map_err(client_error)?;
if let Some(latest) = latest_version.as_ref() {
if content_bytes == latest.content && relative_path == latest.relative_path {
info!("Document content is the same as the latest version, skipping update");
transaction
.rollback()
.await
.context("Failed to rollback transaction")
.map_err(server_error)?;
return Ok(Json(latest.clone().into()));
} else if relative_path != latest.relative_path {
let delete_at_previous_path = StoredDocumentVersion {
vault_id: vault_id.clone(),
vault_update_id: last_update_id + 1,
relative_path: latest.relative_path.clone(),
content: vec![],
created_date: request.created_date,
updated_date: chrono::Utc::now(),
is_deleted: true,
};
last_update_id += 1;
state
.database
.insert_document_version(&delete_at_previous_path, Some(&mut transaction))
.await
.map_err(server_error)?;
}
}
let next_version = latest_version
.as_ref()
.map(|v| v.version_id + 1)
.unwrap_or(0);
let latest_version_content = latest_version
.map(|v| v.content)
.unwrap_or_else(Vec::default);
@ -85,8 +120,8 @@ pub async fn update_document(
let new_version = StoredDocumentVersion {
vault_id,
vault_update_id: last_update_id + 1,
relative_path,
version_id: next_version,
content: merged_content,
created_date: request.created_date,
updated_date: chrono::Utc::now(),