andras/reconcile
andras/reconcile
Watch
1
0
Fork
You've already forked reconcile
 0
Code Issues Pull requests 10 Projects Releases Packages Wiki Activity Actions

Delete server

Browse source
This commit is contained in:
Andras Schmelczer 2025-06-15 10:31:31 +01:00
commit 4cf19ca89f
No known key found for this signature in database
GPG key ID: FC8F2C3D3D1A718C
44 changed files with 0 additions and 2970 deletions
Download patch file Download diff file Expand all files Collapse all files

33
backend/Dockerfile
View file

@ -1,33 +0,0 @@
FROM rust:1.87 AS builder
WORKDIR /usr/src/backend
RUN apt update && apt install -y musl-tools
RUN cargo install sqlx-cli
COPY . .
RUN sqlx database create --database-url sqlite://db.sqlite3
RUN sqlx migrate run --source sync_server/src/app_state/database/migrations --database-url sqlite://db.sqlite3
RUN cargo build --package sync_server --release --target x86_64-unknown-linux-musl
# Runtime image
FROM alpine:3.21.3
LABEL org.opencontainers.image.authors="andras@schmelczer.dev"
RUN apk add --no-cache curl
COPY --from=builder /usr/src/backend/target/x86_64-unknown-linux-musl/release/sync_server /app/sync_server
VOLUME /data
EXPOSE 3000/tcp
WORKDIR /data
HEALTHCHECK \
--interval=30s \
--timeout=5s \
CMD curl -f http://localhost:3000/vaults/fake/ping || exit 1
ENTRYPOINT ["/app/sync_server"]

40
backend/sync_server/Cargo.toml
View file

@ -1,40 +0,0 @@
[package]
name = "sync_server"
version.workspace = true
edition.workspace = true
authors.workspace = true
license.workspace = true
repository.workspace = true
[dependencies]
sync_lib = { path = "../sync_lib" }
serde = { workspace = true }
thiserror = { workspace = true }
tokio = { version = "1.44.2", features = ["full"]}
uuid = { version = "1.16.0", features = ["v4", "serde"] }
log = { version = "0.4.27" }
anyhow = { version = "1.0.98", features = ["backtrace"] }
axum = { version = "0.7.4", features = ["ws", "macros", "tracing", "multipart"]}
axum-extra = { version = "0.9.6", features = ["typed-header"] }
axum_typed_multipart = "0.11.0"
tower-http = { version = "0.6.1", features = ["cors", "trace", "limit", "timeout"] }
tracing = "0.1.41"
tracing-subscriber = { version = "0.3.19", features = ["fmt", "env-filter"]}
sqlx = { version = "0.8.6", features = ["sqlite", "runtime-tokio", "uuid", "chrono"] }
chrono = { version = "0.4.41", features = ["serde"] }
rand = "0.9.0"
sanitize-filename = "0.6.0"
regex = "1.11.1"
clap = { version = "4.5.38", features = ["derive"] }
futures = "0.3.31"
serde_yaml = "0.9.34"
serde_json = "1.0.140"
clap-verbosity-flag = "3.0.3"
bimap = "0.6.3"
ts-rs = { version = "10.1", features = ["uuid-impl", "chrono-impl"] }
serde_with = "3.12.0"
[lints]
workspace = true

9
backend/sync_server/README.md
View file

@ -1,9 +0,0 @@
# Sync server
## Creating/resetting the Database for development
```sh
sqlx database create --database-url sqlite://db.sqlite3
sqlx migrate run --source sync_server/src/app_state/database/migrations --database-url sqlite://db.sqlite3
cargo sqlx prepare --workspace
```

5
backend/sync_server/build.rs
View file

@ -1,5 +0,0 @@
// generated by `sqlx migrate build-script`
fn main() {
// trigger recompilation when a new migration is added
println!("cargo:rerun-if-changed=migrations");
}

41
backend/sync_server/src/app_state.rs
View file

@ -1,41 +0,0 @@
pub mod cursors;
pub mod database;
pub mod websocket;
use std::ffi::OsString;
use anyhow::Result;
use cursors::Cursors;
use database::Database;
use websocket::broadcasts::Broadcasts;
use crate::{config::Config, consts::DEFAULT_CONFIG_PATH};
#[derive(Clone, Debug)]
pub struct AppState {
pub config: Config,
pub database: Database,
pub cursors: Cursors,
pub broadcasts: Broadcasts,
}
impl AppState {
pub async fn try_new(config_path: Option<OsString>) -> Result<Self> {
let config_path = config_path.unwrap_or_else(|| OsString::from(DEFAULT_CONFIG_PATH));
let path = std::path::PathBuf::from(config_path);
let config = Config::read_or_create(&path).await?;
let broadcasts = Broadcasts::new(&config.server);
let database = Database::try_new(&config.database, &broadcasts).await?;
let cursors: Cursors = Cursors::new(&config.database, &broadcasts);
Cursors::start_background_task(cursors.clone());
Ok(Self {
config,
database,
cursors,
broadcasts,
})
}
}

128
backend/sync_server/src/app_state/cursors.rs
View file

@ -1,128 +0,0 @@
use core::time::Duration;
use std::{collections::HashMap, sync::Arc};
use tokio::sync::Mutex;
use super::{
database::models::{DeviceId, VaultId},
websocket::{
broadcasts::Broadcasts,
models::{
ClientCursors, CursorPositionFromServer, CursorSpan, WebSocketServerMessage,
WebSocketServerMessageWithOrigin,
},
},
};
use crate::config::database_config::DatabaseConfig;
#[derive(Clone, Debug)]
pub struct Cursors {
config: DatabaseConfig,
broadcasts: Broadcasts,
vault_to_cursors: Arc<Mutex<HashMap<VaultId, Vec<ClientCursorsWithTimeToLive>>>>,
}
impl Cursors {
pub fn new(config: &DatabaseConfig, broadcasts: &Broadcasts) -> Self {
Self {
config: config.clone(),
broadcasts: broadcasts.clone(),
vault_to_cursors: Arc::new(Mutex::new(HashMap::new())),
}
}
pub async fn update_cursors(
&self,
vault_id: VaultId,
user_name: String,
device_id: &DeviceId,
document_to_cursors: HashMap<String, Vec<CursorSpan>>,
) {
let mut vault_to_cursors = self.vault_to_cursors.lock().await;
let all_device_cursors = vault_to_cursors.entry(vault_id).or_insert_with(Vec::new);
all_device_cursors.retain(|c| &c.client_cursors.device_id != device_id);
all_device_cursors.push(ClientCursorsWithTimeToLive::new(ClientCursors {
user_name,
device_id: device_id.to_string(),
cursors: document_to_cursors,
}));
drop(vault_to_cursors); // Explicitly drop the lock before broadcasting to avoid deadlock
self.broadcast_cursors().await;
}
pub async fn get_cursors(&self, vault_id: &VaultId) -> Vec<ClientCursors> {
let vault_to_cursors = self.vault_to_cursors.lock().await;
vault_to_cursors
.get(vault_id)
.map(|cursors| {
cursors
.iter()
.cloned()
.map(|with_ttl| with_ttl.client_cursors)
.collect::<Vec<_>>()
})
.unwrap_or_default()
}
pub fn start_background_task(self) {
tokio::spawn(async move {
loop {
self.remove_expired_cursors().await;
tokio::time::sleep(Duration::from_secs(1)).await;
}
});
}
async fn remove_expired_cursors(&self) {
let mut vault_to_cursors = self.vault_to_cursors.lock().await;
for (_vault_id, cursors) in vault_to_cursors.iter_mut() {
cursors.retain(|cursor| !cursor.is_expired(self.config.cursor_timeout));
}
}
async fn broadcast_cursors(&self) {
let vault_to_cursors = self.vault_to_cursors.lock().await;
for (vault_id, cursors) in vault_to_cursors.iter() {
self.broadcasts
.send_document_update(
vault_id.clone(),
WebSocketServerMessageWithOrigin::new(WebSocketServerMessage::CursorPositions(
CursorPositionFromServer {
clients: cursors.iter().map(|c| c.client_cursors.clone()).collect(),
},
)),
)
.await;
}
}
pub async fn remove_cursors_of_device(&self, vault_id: &str, device_id: &str) {
let mut vault_to_cursors = self.vault_to_cursors.lock().await;
if let Some(cursors) = vault_to_cursors.get_mut(vault_id) {
cursors.retain(|c| c.client_cursors.device_id != device_id);
}
}
}
#[derive(Clone, Debug)]
struct ClientCursorsWithTimeToLive {
client_cursors: ClientCursors,
last_updated: std::time::Instant,
}
impl ClientCursorsWithTimeToLive {
fn new(client_cursors: ClientCursors) -> Self {
Self {
client_cursors,
last_updated: std::time::Instant::now(),
}
}
pub fn is_expired(&self, ttl: Duration) -> bool { self.last_updated.elapsed() > ttl }
}

425
backend/sync_server/src/app_state/database.rs
View file

@ -1,425 +0,0 @@
use core::time::Duration;
use std::{collections::HashMap, sync::Arc};
use anyhow::{Context as _, Result};
use models::{
DocumentId, DocumentVersionWithoutContent, StoredDocumentVersion, VaultId, VaultUpdateId,
};
use sqlx::{sqlite::SqliteConnectOptions, types::chrono::Utc};
pub mod models;
use sqlx::{Pool, Sqlite, sqlite::SqlitePoolOptions};
use tokio::sync::Mutex;
use uuid::fmt::Hyphenated;
use super::websocket::{
broadcasts::Broadcasts,
models::{WebSocketServerMessage, WebSocketServerMessageWithOrigin, WebSocketVaultUpdate},
};
use crate::config::database_config::DatabaseConfig;
#[derive(Clone, Debug)]
pub struct Database {
config: DatabaseConfig,
broadcasts: Broadcasts,
connection_pools: Arc<Mutex<HashMap<VaultId, Pool<Sqlite>>>>,
}
pub type Transaction<'a> = sqlx::Transaction<'a, Sqlite>;
impl Database {
pub async fn try_new(config: &DatabaseConfig, broadcasts: &Broadcasts) -> Result<Self> {
tokio::fs::create_dir_all(&config.databases_directory_path)
.await
.with_context(|| {
format!(
"Failed to create databases directory: {}",
config.databases_directory_path.to_string_lossy()
)
})?;
let mut connection_pools = std::collections::HashMap::new();
let mut entries = tokio::fs::read_dir(&config.databases_directory_path).await?;
while let Some(entry) = entries.next_entry().await? {
if !entry.file_name().to_string_lossy().ends_with(".sqlite") {
continue;
}
let vault: VaultId = entry
.file_name()
.to_string_lossy()
.trim_end_matches(".sqlite")
.to_owned();
connection_pools.insert(
vault.clone(),
Self::create_vault_database(config, &vault).await?,
);
}
Ok(Self {
config: config.clone(),
connection_pools: Arc::new(Mutex::new(connection_pools)),
broadcasts: broadcasts.clone(),
})
}
async fn create_vault_database(
config: &DatabaseConfig,
vault: &VaultId,
) -> Result<Pool<Sqlite>> {
let file_name = config
.databases_directory_path
.join(format!("{vault}.sqlite"));
let connection_options = SqliteConnectOptions::new()
.filename(file_name.clone())
.create_if_missing(true)
.busy_timeout(Duration::from_secs(3600))
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal);
let pool = SqlitePoolOptions::new()
.max_connections(config.max_connections_per_vault)
.test_before_acquire(true)
.connect_with(connection_options)
.await
.with_context(|| format!("Cannot open database at {}", file_name.display()))?;
Self::run_migrations(&pool).await?;
Ok(pool)
}
async fn run_migrations(pool: &Pool<Sqlite>) -> Result<()> {
sqlx::migrate!("src/app_state/database/migrations")
.run(pool)
.await
.context("Cannot check for pending migrations")
}
async fn get_connection_pool(&self, vault: &VaultId) -> Result<Pool<Sqlite>> {
let mut pools = self.connection_pools.lock().await;
if !pools.contains_key(vault) {
let pool = Self::create_vault_database(&self.config, vault).await?;
pools.insert(vault.clone(), pool);
}
let pool = pools
.get(vault)
.expect("Pool was just inserted or already exists");
Ok(pool.clone())
}
/// Attempting to write from this transaction might result in a
/// database locked error. Use this transaction for read-only operations.
pub async fn create_readonly_transaction(
&self,
vault: &VaultId,
) -> Result<Transaction<'static>> {
self.get_connection_pool(vault)
.await?
.begin()
.await
.context("Cannot create transaction")
}
pub async fn create_write_transaction(&self, vault: &VaultId) -> Result<Transaction<'static>> {
let mut transaction = self.create_readonly_transaction(vault).await?;
// sqlx doesn't support immediate transactions for sqlite: https://github.com/launchbadge/sqlx/issues/481
sqlx::query!("END; BEGIN IMMEDIATE;")
.execute(&mut *transaction)
.await?;
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
from latest_document_versions
where relative_path = ?
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
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
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")
}
pub async fn insert_document_version(
&self,
vault_id: &VaultId,
version: &StoredDocumentVersion,
transaction: Option<&mut 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(transaction) = transaction {
query.execute(&mut **transaction).await
} 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(())
}
}

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

@ -1,21 +0,0 @@
CREATE TABLE IF NOT EXISTS documents (
vault_update_id INTEGER NOT NULL PRIMARY KEY,
document_id TEXT NOT NULL,
relative_path TEXT NOT NULL,
updated_date TIMESTAMP NOT NULL,
content BLOB NOT NULL,
is_deleted BOOLEAN 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 IF NOT EXISTS idx_documents_vault_id_relative_path
ON documents (relative_path);

2
backend/sync_server/src/app_state/database/migrations/20250522192949_add_provenance_columns.sql
View file

@ -1,2 +0,0 @@
ALTER TABLE documents ADD COLUMN user_id TEXT NOT NULL DEFAULT "";
ALTER TABLE documents ADD COLUMN device_id TEXT NOT NULL DEFAULT "";

89
backend/sync_server/src/app_state/database/models.rs
View file

@ -1,89 +0,0 @@
use chrono::{DateTime, Utc};
use serde::Serialize;
use sync_lib::bytes_to_base64;
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;
#[derive(Debug, Clone)]
pub struct StoredDocumentVersion {
pub vault_update_id: VaultUpdateId,
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)]
#[serde(rename_all = "camelCase")]
pub struct DocumentVersionWithoutContent {
#[ts(as = "i32")]
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,
}
impl From<StoredDocumentVersion> for DocumentVersionWithoutContent {
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,
is_deleted: value.is_deleted,
user_id: value.user_id,
device_id: value.device_id,
content_size: value.content.len() as u64,
}
}
}
#[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,
}
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: bytes_to_base64(&value.content),
is_deleted: value.is_deleted,
user_id: value.user_id,
device_id: value.device_id,
}
}
}

3
backend/sync_server/src/app_state/websocket.rs
View file

@ -1,3 +0,0 @@
pub mod broadcasts;
pub mod models;
pub mod utils;

63
backend/sync_server/src/app_state/websocket/broadcasts.rs
View file

@ -1,63 +0,0 @@
use std::{collections::HashMap, sync::Arc};
use anyhow::Context;
use tokio::sync::{Mutex, broadcast};
use super::models::WebSocketServerMessageWithOrigin;
use crate::{
app_state::database::models::VaultId, config::server_config::ServerConfig, errors::server_error,
};
#[derive(Debug, Clone)]
pub struct Broadcasts {
max_clients_per_vault: usize,
tx: Arc<Mutex<HashMap<VaultId, broadcast::Sender<WebSocketServerMessageWithOrigin>>>>,
}
impl Broadcasts {
pub fn new(server_config: &ServerConfig) -> Self {
Self {
max_clients_per_vault: server_config.max_clients_per_vault,
tx: Arc::new(Mutex::new(HashMap::new())),
}
}
pub async fn get_receiver(
&self,
vault: VaultId,
) -> broadcast::Receiver<WebSocketServerMessageWithOrigin> {
let tx = self.get_or_create(vault).await;
tx.subscribe()
}
/// Notify all clients (who are subscribed to the vault) about an update.
/// We only log failures.
pub async fn send_document_update(
&self,
vault: VaultId,
document: WebSocketServerMessageWithOrigin,
) {
let tx = self.get_or_create(vault).await;
let result = tx
.send(document)
.context("Cannot broadcast server message to websocket listeners")
.map_err(server_error);
if result.is_err() {
log::debug!("Failed to send message: {result:?}");
}
}
async fn get_or_create(
&self,
vault: VaultId,
) -> broadcast::Sender<WebSocketServerMessageWithOrigin> {
let mut tx = self.tx.lock().await;
tx.entry(vault)
.or_insert_with(|| broadcast::channel(self.max_clients_per_vault).0.clone())
.clone()
}
}

88
backend/sync_server/src/app_state/websocket/models.rs
View file

@ -1,88 +0,0 @@
use std::collections::HashMap;
use serde::{Deserialize, Serialize};
use ts_rs::TS;
use crate::app_state::database::models::{DeviceId, DocumentVersionWithoutContent, VaultUpdateId};
#[derive(TS, Deserialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct WebSocketHandshake {
pub token: String,
pub device_id: DeviceId,
#[ts(as = "Option<i32>")]
pub last_seen_vault_update_id: Option<VaultUpdateId>,
}
#[derive(TS, Serialize, Deserialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct CursorSpan {
pub start: usize,
pub end: usize,
}
#[derive(TS, Deserialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct CursorPositionFromClient {
pub document_to_cursors: HashMap<String, Vec<CursorSpan>>,
}
#[derive(TS, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct ClientCursors {
pub user_name: String,
pub device_id: DeviceId,
pub cursors: HashMap<String, Vec<CursorSpan>>,
}
#[derive(TS, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct CursorPositionFromServer {
pub clients: Vec<ClientCursors>,
}
#[derive(TS, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct WebSocketVaultUpdate {
pub documents: Vec<DocumentVersionWithoutContent>,
pub is_initial_sync: bool,
}
#[derive(TS, Deserialize, Clone, Debug)]
#[serde(rename_all = "camelCase", tag = "type")]
#[ts(export)]
pub enum WebSocketClientMessage {
Handshake(WebSocketHandshake),
CursorPositions(CursorPositionFromClient),
}
#[derive(TS, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase", tag = "type")]
#[ts(export)]
pub enum WebSocketServerMessage {
VaultUpdate(WebSocketVaultUpdate),
CursorPositions(CursorPositionFromServer),
}
#[derive(Clone, Debug)]
pub struct WebSocketServerMessageWithOrigin {
pub origin_device_id: Option<DeviceId>,
pub message: WebSocketServerMessage,
}
impl WebSocketServerMessageWithOrigin {
pub fn new(message: WebSocketServerMessage) -> Self {
Self {
origin_device_id: None,
message,
}
}
pub fn with_origin(origin_device_id: DeviceId, message: WebSocketServerMessage) -> Self {
Self {
origin_device_id: Some(origin_device_id),
message,
}
}
}

80
backend/sync_server/src/app_state/websocket/utils.rs
View file

@ -1,80 +0,0 @@
use anyhow::Context;
use axum::extract::ws::{Message, WebSocket};
use futures::{sink::SinkExt, stream::SplitSink};
use super::models::{WebSocketClientMessage, WebSocketHandshake, WebSocketServerMessage};
use crate::{
app_state::{
AppState,
database::models::{DocumentVersionWithoutContent, VaultId, VaultUpdateId},
},
config::user_config::User,
errors::{SyncServerError, server_error, unauthenticated_error},
server::auth::auth,
};
pub struct AuthenticatedWebSocketHandshake {
pub handshake: WebSocketHandshake,
pub user: User,
}
pub fn get_authenticated_handshake(
state: &AppState,
vault_id: &VaultId,
message: Option<Message>,
) -> Result<AuthenticatedWebSocketHandshake, SyncServerError> {
if let Some(Message::Text(message)) = message {
let message: WebSocketClientMessage = serde_json::from_str(&message)
.context("Failed to parse message")
.map_err(server_error)?;
match message {
WebSocketClientMessage::Handshake(handshake) => {
let user = auth(state, handshake.token.trim(), vault_id)?;
Ok(AuthenticatedWebSocketHandshake { handshake, user })
}
WebSocketClientMessage::CursorPositions(_) => Err(unauthenticated_error(
anyhow::anyhow!("Expected a handshake message"),
)),
}
} else {
Err(unauthenticated_error(anyhow::anyhow!(
"Failed to authenticate due to invalid message"
)))
}
}
pub async fn get_unseen_documents(
state: &AppState,
vault_id: &VaultId,
last_seen_vault_update_id: Option<VaultUpdateId>,
) -> Result<Vec<DocumentVersionWithoutContent>, SyncServerError> {
if let Some(update_id) = last_seen_vault_update_id {
state
.database
.get_latest_documents_since(vault_id, update_id, None)
.await
.map_err(server_error)
} else {
state
.database
.get_latest_documents(vault_id, None)
.await
.map_err(server_error)
}
}
pub async fn send_update_over_websocket(
update: &WebSocketServerMessage,
sender: &mut SplitSink<WebSocket, Message>,
) -> Result<(), SyncServerError> {
let serialized_update = serde_json::to_string(update)
.context("Failed to serialize update")
.map_err(server_error)?;
sender
.send(Message::Text(serialized_update))
.await
.context("Failed to send message over websocket")
.map_err(server_error)
}

2
backend/sync_server/src/cli.rs
View file

@ -1,2 +0,0 @@
pub mod args;
pub mod color_when;

26
backend/sync_server/src/cli/args.rs
View file

@ -1,26 +0,0 @@
use std::ffi::OsString;
use clap::Parser;
use clap_verbosity_flag::{InfoLevel, Verbosity};
use crate::cli::color_when::ColorWhen;
/// Server for backing the `VaultLink` plugin
#[derive(Parser, Debug)]
#[command(version, about, long_about = None)]
pub struct Args {
#[arg(index = 1)]
pub config_path: Option<OsString>,
#[command(flatten)]
pub verbose: Verbosity<InfoLevel>,
#[arg(
long,
value_name = "WHEN",
default_value_t = ColorWhen::Auto,
default_missing_value = "always",
value_enum
)]
pub color: ColorWhen,
}

31
backend/sync_server/src/cli/color_when.rs
View file

@ -1,31 +0,0 @@
use std::io::IsTerminal;
use clap::ValueEnum;
#[derive(ValueEnum, Copy, Clone, Debug, PartialEq, Eq)]
pub enum ColorWhen {
Always,
Auto,
Never,
}
impl ColorWhen {
pub fn use_colors(self) -> bool {
match self {
ColorWhen::Always => true,
ColorWhen::Auto => {
std::env::var_os("NO_COLOR").is_none() && std::io::stderr().is_terminal()
}
ColorWhen::Never => false,
}
}
}
impl std::fmt::Display for ColorWhen {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.to_possible_value()
.expect("no values are skipped")
.get_name()
.fmt(f)
}
}

66
backend/sync_server/src/config.rs
View file

@ -1,66 +0,0 @@
use std::path::Path;
use anyhow::{Context as _, Result};
use database_config::DatabaseConfig;
use log::info;
use serde::{Deserialize, Serialize};
use server_config::ServerConfig;
use tokio::fs;
use user_config::UserConfig;
pub mod database_config;
pub mod server_config;
pub mod user_config;
#[derive(Debug, Deserialize, Serialize, Clone, Default)]
pub struct Config {
#[serde(default)]
pub database: DatabaseConfig,
#[serde(default)]
pub server: ServerConfig,
#[serde(default)]
pub users: UserConfig,
}
impl Config {
pub async fn read_or_create(path: &Path) -> Result<Self> {
let config = if path.exists() {
info!(
"Loading configuration from '{}'",
path.canonicalize().unwrap().display()
);
Self::load_from_file(path).await?
} else {
Self::default()
};
config.write(path).await?;
info!(
"Updated configuration at '{}'",
path.canonicalize().unwrap().display()
);
Ok(config)
}
pub async fn load_from_file(path: &Path) -> Result<Self> {
let contents = fs::read_to_string(path).await.with_context(|| {
format!(
"Cannot load configuration from disk from {}",
path.display()
)
})?;
let config = serde_yaml::from_str(&contents).context("Failed to parse configuration")?;
Ok(config)
}
pub async fn write(&self, path: &Path) -> Result<()> {
let contents = serde_yaml::to_string(&self).context("Failed to serialize configuration")?;
fs::write(path, contents)
.await
.context("Failed to write configuration to disk")
}
}

48
backend/sync_server/src/config/database_config.rs
View file

@ -1,48 +0,0 @@
use std::{path::PathBuf, time::Duration};
use log::debug;
use serde::{Deserialize, Serialize};
use serde_with::serde_as;
use crate::consts::{
DEFAULT_CURSOR_TIMEOUT, DEFAULT_DATABASES_DIRECTORY_PATH, DEFAULT_MAX_CONNECTIONS_PER_VAULT,
};
#[serde_with::serde_as]
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct DatabaseConfig {
#[serde(default = "default_databases_directory_path")]
pub databases_directory_path: PathBuf,
#[serde(default = "default_max_connections_per_vault")]
pub max_connections_per_vault: u32,
#[serde(default = "default_cursor_timeout", rename = "cursor_timeout_seconds")]
#[serde_as(as = "serde_with::DurationSeconds<u64>")]
pub cursor_timeout: Duration,
}
fn default_databases_directory_path() -> PathBuf {
debug!("Using default databases directory path: {DEFAULT_DATABASES_DIRECTORY_PATH:?}");
PathBuf::from(DEFAULT_DATABASES_DIRECTORY_PATH)
}
fn default_max_connections_per_vault() -> u32 {
debug!("Using default max connections: {DEFAULT_MAX_CONNECTIONS_PER_VAULT}");
DEFAULT_MAX_CONNECTIONS_PER_VAULT
}
fn default_cursor_timeout() -> Duration {
debug!("Using default cursor timeout: {DEFAULT_CURSOR_TIMEOUT:?}");
DEFAULT_CURSOR_TIMEOUT
}
impl Default for DatabaseConfig {
fn default() -> Self {
Self {
databases_directory_path: default_databases_directory_path(),
max_connections_per_vault: default_max_connections_per_vault(),
cursor_timeout: default_cursor_timeout(),
}
}
}

50
backend/sync_server/src/config/server_config.rs
View file

@ -1,50 +0,0 @@
use log::debug;
use serde::{Deserialize, Serialize};
use crate::consts::{
DEFAULT_HOST, DEFAULT_MAX_BODY_SIZE_MB, DEFAULT_MAX_CLIENTS_PER_VAULT, DEFAULT_PORT,
DEFAULT_RESPONSE_TIMEOUT_SECONDS,
};
#[derive(Debug, Deserialize, Serialize, Clone, Default)]
pub struct ServerConfig {
#[serde(default = "default_host")]
pub host: String,
#[serde(default = "default_port")]
pub port: u16,
#[serde(default = "default_max_body_size_mb")]
pub max_body_size_mb: usize,
#[serde(default = "default_max_clients_per_vault")]
pub max_clients_per_vault: usize,
#[serde(default = "default_response_timeout_seconds")]
pub response_timeout_seconds: u64,
}
fn default_host() -> String {
debug!("Using default server host: {DEFAULT_HOST}");
DEFAULT_HOST.to_owned()
}
fn default_port() -> u16 {
debug!("Using default server port: {DEFAULT_PORT}");
DEFAULT_PORT
}
fn default_max_body_size_mb() -> usize {
debug!("Using default max body size (MB): {DEFAULT_MAX_BODY_SIZE_MB}");
DEFAULT_MAX_BODY_SIZE_MB
}
fn default_max_clients_per_vault() -> usize {
debug!("Using default max clients per vault: {DEFAULT_MAX_CLIENTS_PER_VAULT}");
DEFAULT_MAX_CLIENTS_PER_VAULT
}
fn default_response_timeout_seconds() -> u64 {
debug!("Using default response timeout (seconds): {DEFAULT_RESPONSE_TIMEOUT_SECONDS}");
DEFAULT_RESPONSE_TIMEOUT_SECONDS
}

Some files were not shown because too many files have changed in this diff Show more