andras/vault-link
andras/vault-link
Watch
1
0
Fork
You've already forked vault-link
 0
Code Issues Pull requests 18 Projects Releases 55 Packages Wiki Activity Actions 31

Migrate server to CAS only

Browse source
This commit is contained in:
Andras Schmelczer 2026-09-16 10:36:20 +01:00
commit 1619c51ad9
1 changed files with 1163 additions and 1092 deletions
Download patch file Download diff file Expand all files Collapse all files

324
sync-server/src/app_state/database.rs
View file

@ -3,21 +3,24 @@ use std::{collections::HashMap, sync::Arc};
use anyhow::{Context as _, Result};
use log::info;
use models::{
DocumentId, DocumentVersionWithoutContent, StoredDocumentVersion, VaultId, VaultUpdateId,
};
use sqlx::{ConnectOptions, sqlite::SqliteConnectOptions, types::chrono::Utc};
pub mod models;
use models::VaultId;
use sqlx::{ConnectOptions, sqlite::SqliteConnectOptions};
use sqlx::{Pool, Sqlite, sqlite::SqlitePoolOptions};
use tokio::sync::Mutex;
use tokio::time::Instant;
use uuid::fmt::Hyphenated;
use super::websocket::{
broadcasts::Broadcasts,
models::{WebSocketServerMessage, WebSocketServerMessageWithOrigin, WebSocketVaultUpdate},
};
pub mod models;
mod get_document_version;
mod get_latest_document;
mod get_latest_document_by_path;
mod get_latest_documents;
mod get_latest_documents_since;
mod get_max_update_id_in_vault;
mod get_push_acknowledgement;
mod insert_document_version;
use super::websocket::broadcasts::Broadcasts;
use crate::config::database_config::DatabaseConfig;
#[derive(Clone)]
@ -176,305 +179,6 @@ impl Database {
Ok(transaction)
}
/// Return the latest state of all documents in the vault
pub async fn get_latest_documents(
&self,
vault: &VaultId,
transaction: Option<&mut Transaction<'_>>,
) -> Result<Vec<DocumentVersionWithoutContent>> {
let query = sqlx::query!(
r#"
select
vault_update_id,
document_id as "document_id: Hyphenated",
relative_path,
updated_date as "updated_date: chrono::DateTime<Utc>",
is_deleted,
user_id,
device_id,
length(content) as "content_size: u64"
from latest_document_versions
order by vault_update_id
"#,
);
if let Some(transaction) = transaction {
query.fetch_all(&mut **transaction).await
} else {
query
.fetch_all(&self.get_connection_pool(vault).await?)
.await
}
.context("Cannot fetch latest documents")
.map(|rows| {
rows.into_iter()
.map(|row| DocumentVersionWithoutContent {
vault_update_id: row.vault_update_id,
document_id: row.document_id.into(),
relative_path: row.relative_path,
updated_date: row.updated_date,
is_deleted: row.is_deleted,
user_id: row.user_id,
device_id: row.device_id,
content_size: row
.content_size
.expect("Content size can't be null but sqlx can't infer it"),
})
.collect()
})
}
/// Return the latest state of all documents (including deleted) in the
/// vault which have changed since the given update id
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!(
r#"
select
vault_update_id,
document_id as "document_id: Hyphenated",
relative_path,
updated_date as "updated_date: chrono::DateTime<Utc>",
is_deleted,
user_id,
device_id,
length(content) as "content_size: u64"
from latest_document_versions
where vault_update_id > ?
order by vault_update_id
"#,
vault_update_id
);
if let Some(transaction) = transaction {
query.fetch_all(&mut **transaction).await
} else {
query
.fetch_all(&self.get_connection_pool(vault).await?)
.await
}
.with_context(|| {
format!("Cannot fetch latest documents since vault_update_id `{vault_update_id}`")
})
.map(|rows| {
rows.into_iter()
.map(|row| DocumentVersionWithoutContent {
vault_update_id: row.vault_update_id,
document_id: row.document_id.into(),
relative_path: row.relative_path,
updated_date: row.updated_date,
is_deleted: row.is_deleted,
user_id: row.user_id,
device_id: row.device_id,
content_size: row
.content_size
.expect("Content size can't be null but sqlx can't infer it"),
})
.collect()
})
}
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
"#,
);
if let Some(transaction) = transaction {
query.fetch_one(&mut **transaction).await
} else {
query
.fetch_one(&self.get_connection_pool(vault).await?)
.await
}
.map(|row| row.max_vault_update_id)
.context("Cannot fetch max update id in vault")
}
pub async fn get_latest_document_by_path(
&self,
vault: &VaultId,
relative_path: &str,
transaction: Option<&mut Transaction<'_>>,
) -> Result<Option<StoredDocumentVersion>> {
let query = sqlx::query_as!(
StoredDocumentVersion,
r#"
select
vault_update_id,
document_id as "document_id: Hyphenated",
relative_path,
updated_date as "updated_date: chrono::DateTime<Utc>",
content,
is_deleted,
user_id,
device_id,
has_been_merged
from latest_document_versions
where relative_path = ? and is_deleted = false
order by vault_update_id desc -- `latest_document_versions` only contains a single latest version of each document, however,
-- multiple documents can have the same `relative_path`, if they have been deleted. That's
-- why we only care about the latest version of the document with the given relative path.
limit 1
"#,
relative_path
);
if let Some(transaction) = transaction {
query.fetch_optional(&mut **transaction).await
} else {
query
.fetch_optional(&self.get_connection_pool(vault).await?)
.await
}
.context("Cannot fetch latest document version")
}
pub async fn get_latest_document(
&self,
vault: &VaultId,
document_id: &DocumentId,
transaction: Option<&mut Transaction<'_>>,
) -> Result<Option<StoredDocumentVersion>> {
let document_id = document_id.as_hyphenated();
let query = sqlx::query_as!(
StoredDocumentVersion,
r#"
select
vault_update_id,
document_id as "document_id: Hyphenated",
relative_path,
updated_date as "updated_date: chrono::DateTime<Utc>",
content,
is_deleted,
user_id,
device_id,
has_been_merged
from latest_document_versions
where document_id = ?
"#,
document_id
);
if let Some(transaction) = transaction {
query.fetch_optional(&mut **transaction).await
} else {
query
.fetch_optional(&self.get_connection_pool(vault).await?)
.await
}
.context("Cannot fetch latest document version")
}
pub async fn get_document_version(
&self,
vault: &VaultId,
vault_update_id: VaultUpdateId,
transaction: Option<&mut Transaction<'_>>,
) -> Result<Option<StoredDocumentVersion>> {
let query = sqlx::query_as!(
StoredDocumentVersion,
r#"
select
vault_update_id,
document_id as "document_id: Hyphenated",
relative_path,
updated_date as "updated_date: chrono::DateTime<Utc>",
content,
is_deleted,
user_id,
device_id,
has_been_merged
from documents
where vault_update_id = ?"#,
vault_update_id
);
if let Some(transaction) = transaction {
query.fetch_optional(&mut **transaction).await
} else {
query
.fetch_optional(&self.get_connection_pool(vault).await?)
.await
}
.context("Cannot fetch document version")
}
// inserting the document must be the last step of the transaction if there's one
pub async fn insert_document_version(
&self,
vault_id: &VaultId,
version: &StoredDocumentVersion,
transaction: Option<Transaction<'_>>,
) -> Result<()> {
let document_id = version.document_id.as_hyphenated();
let query = sqlx::query!(
r#"
insert into documents (
vault_update_id,
document_id,
relative_path,
updated_date,
content,
is_deleted,
user_id,
device_id
)
values (?, ?, ?, ?, ?, ?, ?, ?)
"#,
version.vault_update_id,
document_id,
version.relative_path,
version.updated_date,
version.content,
version.is_deleted,
version.user_id,
version.device_id
);
if let Some(mut transaction) = transaction {
query
.execute(&mut *transaction)
.await
.context("Cannot insert document version")?;
transaction
.commit()
.await
.context("Failed to commit transaction")?;
} else {
query
.execute(&self.get_connection_pool(vault_id).await?)
.await
.context("Cannot insert document version")?;
}
self.broadcasts
.send_document_update(
vault_id.clone(),
WebSocketServerMessageWithOrigin::with_origin(
version.device_id.clone(),
WebSocketServerMessage::VaultUpdate(WebSocketVaultUpdate {
documents: vec![version.clone().into()],
is_initial_sync: false,
}),
),
)
.await;
Ok(())
}
/// Cleanup idle connection pools that haven't been accessed in more than 5 minutes
async fn cleanup_idle_pools(&self) {
let mut pools = self.connection_pools.lock().await;