mirror/spacedrive
1
0
Fork 0
mirror of https://github.com/spacedriveapp/spacedrive.git synced 2026-05-18 21:36:56 -04:00
Code Issues Packages Projects Releases Wiki Activity

feat: add cloud volume setup and migration changes

Browse Source
This commit is contained in:
Jamie Pine
2025-10-15 02:03:22 -07:00
parent 3ee1cd1b33
commit 5eafd66bda
22 changed files with 4177 additions and 2187 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
.github/workflows/search-index.yml vendored
View File

@@ -1,24 +0,0 @@
name: Trigger Algolia Crawler
on:
workflow_dispatch:
push:
paths:
- 'docs/**'
branches:
- main
pull_request:
paths:
- '.github/workflows/search-index.yml'
jobs:
trigger_crawler:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Trigger Algolia Crawler
run: |
curl -X POST --user ${{secrets.CRAWLER_USER_ID}}:${{secrets.CRAWLER_API_KEY}} "https://crawler.algolia.com/api/1/crawlers/${{secrets.CRAWLER_ID}}/reindex"

BIN
Cargo.lock generated
View File

Binary file not shown.

238
apps/cli/src/domains/cloud/setup.rs
View File

@@ -47,7 +47,30 @@ pub async fn run_interactive(ctx: &Context) -> Result<()> {
async fn add_cloud_interactive(ctx: &Context) -> Result<()> {
println!("\n=== Add Cloud Storage ===\n");
// 1. Select service type (only S3-compatible for now)
let service_category_idx = select(
"Select cloud storage category",
&[
"S3-compatible (Amazon, R2, B2, MinIO, etc.)".to_string(),
"Google Drive".to_string(),
"Microsoft OneDrive".to_string(),
"Dropbox".to_string(),
"Azure Blob Storage".to_string(),
"Google Cloud Storage".to_string(),
],
)?;
match service_category_idx {
0 => add_s3_interactive(ctx).await,
1 => add_google_drive_interactive(ctx).await,
2 => add_onedrive_interactive(ctx).await,
3 => add_dropbox_interactive(ctx).await,
4 => add_azure_blob_interactive(ctx).await,
5 => add_gcs_interactive(ctx).await,
_ => unreachable!(),
}
}
async fn add_s3_interactive(ctx: &Context) -> Result<()> {
let service_idx = select(
"Select S3-compatible storage provider",
&[
@@ -63,18 +86,16 @@ async fn add_cloud_interactive(ctx: &Context) -> Result<()> {
let (service_type, needs_endpoint, default_region) = match service_idx {
0 => (CloudServiceType::S3, false, None),
1 => (CloudServiceType::S3, true, Some("auto".to_string())), // R2 uses "auto"
1 => (CloudServiceType::S3, true, Some("auto".to_string())),
2 => (CloudServiceType::BackblazeB2, false, None),
3 => (CloudServiceType::Wasabi, false, None),
4 => (CloudServiceType::DigitalOceanSpaces, false, None),
5 => (CloudServiceType::S3, true, Some("us-east-1".to_string())), // MinIO default
5 => (CloudServiceType::S3, true, Some("us-east-1".to_string())),
6 => (CloudServiceType::Other, true, Some("us-east-1".to_string())),
_ => unreachable!(),
};
// 2. Basic configuration
let name = text("Volume name (e.g., 'My S3 Bucket')", false)?.unwrap();
let bucket = text("Bucket name", false)?.unwrap();
let region = if let Some(default) = default_region {
@@ -94,13 +115,11 @@ async fn add_cloud_interactive(ctx: &Context) -> Result<()> {
None
};
// 3. Credentials
println!("\nCredentials will be stored securely in your system keyring\n");
let access_key = password("Access Key ID", false)?.unwrap();
let secret_key = password("Secret Access Key", false)?.unwrap();
// 4. Confirm and save
println!("\nSummary:");
println!(" Provider: {:?}", service_type);
println!(" Name: {}", name);
@@ -113,7 +132,6 @@ async fn add_cloud_interactive(ctx: &Context) -> Result<()> {
confirm_or_abort("Add this cloud volume?", false)?;
// Build and execute action
let input = VolumeAddCloudInput {
service: service_type,
display_name: name.clone(),
@@ -126,6 +144,210 @@ async fn add_cloud_interactive(ctx: &Context) -> Result<()> {
},
};
execute_add_cloud(ctx, input).await
}
async fn add_google_drive_interactive(ctx: &Context) -> Result<()> {
let name = text("Volume name (e.g., 'My Google Drive')", false)?.unwrap();
let root = text("Root folder ID (leave empty for entire drive)", true)?;
println!("\nOAuth Setup:");
println!(" You'll need OAuth credentials from Google Cloud Console");
println!(" Visit: https://console.cloud.google.com/apis/credentials\n");
let client_id = text("OAuth Client ID", false)?.unwrap();
let client_secret = password("OAuth Client Secret", false)?.unwrap();
println!("\nAfter authorizing, you'll receive tokens:");
let access_token = password("Access Token", false)?.unwrap();
let refresh_token = password("Refresh Token", false)?.unwrap();
println!("\nSummary:");
println!(" Provider: Google Drive");
println!(" Name: {}", name);
if let Some(ref r) = root {
println!(" Root: {}", r);
}
println!();
confirm_or_abort("Add this cloud volume?", false)?;
let input = VolumeAddCloudInput {
service: CloudServiceType::GoogleDrive,
display_name: name.clone(),
config: CloudStorageConfig::GoogleDrive {
root,
access_token,
refresh_token,
client_id,
client_secret,
},
};
execute_add_cloud(ctx, input).await
}
async fn add_onedrive_interactive(ctx: &Context) -> Result<()> {
let name = text("Volume name (e.g., 'My OneDrive')", false)?.unwrap();
let root = text("Root folder path (leave empty for entire drive)", true)?;
println!("\nOAuth Setup:");
println!(" You'll need OAuth credentials from Azure Portal");
println!(" Visit: https://portal.azure.com/#blade/Microsoft_AAD_RegisteredApps\n");
let client_id = text("Application (client) ID", false)?.unwrap();
let client_secret = password("Client Secret", false)?.unwrap();
println!("\nAfter authorizing, you'll receive tokens:");
let access_token = password("Access Token", false)?.unwrap();
let refresh_token = password("Refresh Token", false)?.unwrap();
println!("\nSummary:");
println!(" Provider: Microsoft OneDrive");
println!(" Name: {}", name);
if let Some(ref r) = root {
println!(" Root: {}", r);
}
println!();
confirm_or_abort("Add this cloud volume?", false)?;
let input = VolumeAddCloudInput {
service: CloudServiceType::OneDrive,
display_name: name.clone(),
config: CloudStorageConfig::OneDrive {
root,
access_token,
refresh_token,
client_id,
client_secret,
},
};
execute_add_cloud(ctx, input).await
}
async fn add_dropbox_interactive(ctx: &Context) -> Result<()> {
let name = text("Volume name (e.g., 'My Dropbox')", false)?.unwrap();
let root = text("Root folder path (leave empty for entire Dropbox)", true)?;
println!("\nOAuth Setup:");
println!(" You'll need OAuth credentials from Dropbox App Console");
println!(" Visit: https://www.dropbox.com/developers/apps\n");
let client_id = text("App Key (Client ID)", false)?.unwrap();
let client_secret = password("App Secret (Client Secret)", false)?.unwrap();
println!("\nAfter authorizing, you'll receive tokens:");
let access_token = password("Access Token", false)?.unwrap();
let refresh_token = password("Refresh Token", false)?.unwrap();
println!("\nSummary:");
println!(" Provider: Dropbox");
println!(" Name: {}", name);
if let Some(ref r) = root {
println!(" Root: {}", r);
}
println!();
confirm_or_abort("Add this cloud volume?", false)?;
let input = VolumeAddCloudInput {
service: CloudServiceType::Dropbox,
display_name: name.clone(),
config: CloudStorageConfig::Dropbox {
root,
access_token,
refresh_token,
client_id,
client_secret,
},
};
execute_add_cloud(ctx, input).await
}
async fn add_azure_blob_interactive(ctx: &Context) -> Result<()> {
let name = text("Volume name (e.g., 'My Azure Storage')", false)?.unwrap();
let container = text("Container name", false)?.unwrap();
let account_name = text("Storage account name", false)?.unwrap();
let endpoint = text("Custom endpoint (leave empty for default)", true)?;
println!("\nCredentials will be stored securely in your system keyring\n");
let account_key = password("Storage account key", false)?.unwrap();
println!("\nSummary:");
println!(" Provider: Azure Blob Storage");
println!(" Name: {}", name);
println!(" Container: {}", container);
println!(" Account: {}", account_name);
if let Some(ref e) = endpoint {
println!(" Endpoint: {}", e);
}
println!();
confirm_or_abort("Add this cloud volume?", false)?;
let input = VolumeAddCloudInput {
service: CloudServiceType::AzureBlob,
display_name: name.clone(),
config: CloudStorageConfig::AzureBlob {
container,
endpoint,
account_name,
account_key,
},
};
execute_add_cloud(ctx, input).await
}
async fn add_gcs_interactive(ctx: &Context) -> Result<()> {
let name = text("Volume name (e.g., 'My GCS Bucket')", false)?.unwrap();
let bucket = text("Bucket name", false)?.unwrap();
let root = text("Root path (leave empty for entire bucket)", true)?;
let endpoint = text("Custom endpoint (leave empty for default)", true)?;
println!("\nService Account Setup:");
println!(" You'll need a service account JSON key from Google Cloud Console");
println!(" Visit: https://console.cloud.google.com/iam-admin/serviceaccounts\n");
let service_account_path = text("Path to service account JSON file", false)?.unwrap();
let credential = std::fs::read_to_string(&service_account_path)
.map_err(|e| anyhow::anyhow!("Failed to read service account file: {}", e))?;
println!("\nSummary:");
println!(" Provider: Google Cloud Storage");
println!(" Name: {}", name);
println!(" Bucket: {}", bucket);
if let Some(ref r) = root {
println!(" Root: {}", r);
}
if let Some(ref e) = endpoint {
println!(" Endpoint: {}", e);
}
println!();
confirm_or_abort("Add this cloud volume?", false)?;
let input = VolumeAddCloudInput {
service: CloudServiceType::GoogleCloudStorage,
display_name: name.clone(),
config: CloudStorageConfig::GoogleCloudStorage {
bucket,
root,
endpoint,
credential,
},
};
execute_add_cloud(ctx, input).await
}
async fn execute_add_cloud(
ctx: &Context,
input: VolumeAddCloudInput,
) -> Result<()> {
print!("Connecting to cloud storage... ");
std::io::Write::flush(&mut std::io::stdout())?;

172
apps/cli/src/domains/volume/args.rs
View File

@@ -16,25 +16,61 @@ pub struct VolumeAddCloudArgs {
#[arg(long, value_enum)]
pub service: CloudServiceArg,
/// S3 bucket name (for S3 service)
#[arg(long, required_if_eq("service", "s3"))]
/// Bucket name (S3, GCS)
#[arg(long)]
pub bucket: Option<String>,
/// S3 region (for S3 service)
#[arg(long, required_if_eq("service", "s3"))]
/// Region (S3)
#[arg(long)]
pub region: Option<String>,
/// S3 access key ID (for S3 service)
#[arg(long, required_if_eq("service", "s3"))]
/// Access key ID (S3, Azure)
#[arg(long)]
pub access_key_id: Option<String>,
/// S3 secret access key (for S3 service)
#[arg(long, required_if_eq("service", "s3"))]
/// Secret access key (S3, Azure)
#[arg(long)]
pub secret_access_key: Option<String>,
/// Custom S3 endpoint (optional, for S3-compatible services like MinIO, R2, etc.)
/// Custom endpoint (S3, Azure, GCS)
#[arg(long)]
pub endpoint: Option<String>,
/// Root folder path or ID (Google Drive, OneDrive, Dropbox, GCS)
#[arg(long)]
pub root: Option<String>,
/// OAuth access token (Google Drive, OneDrive, Dropbox)
#[arg(long)]
pub access_token: Option<String>,
/// OAuth refresh token (Google Drive, OneDrive, Dropbox)
#[arg(long)]
pub refresh_token: Option<String>,
/// OAuth client ID (Google Drive, OneDrive, Dropbox)
#[arg(long)]
pub client_id: Option<String>,
/// OAuth client secret (Google Drive, OneDrive, Dropbox)
#[arg(long)]
pub client_secret: Option<String>,
/// Container name (Azure Blob)
#[arg(long)]
pub container: Option<String>,
/// Storage account name (Azure Blob)
#[arg(long)]
pub account_name: Option<String>,
/// Storage account key (Azure Blob)
#[arg(long)]
pub account_key: Option<String>,
/// Path to service account JSON file (GCS)
#[arg(long)]
pub service_account: Option<String>,
}
#[derive(clap::ValueEnum, Clone, Debug)]
@@ -71,15 +107,18 @@ impl VolumeAddCloudArgs {
let service = CloudServiceType::from(self.service.clone());
let config = match self.service {
CloudServiceArg::S3 => {
let bucket = self.bucket.ok_or("--bucket is required for S3")?;
let region = self.region.ok_or("--region is required for S3")?;
CloudServiceArg::S3
| CloudServiceArg::BackblazeB2
| CloudServiceArg::Wasabi
| CloudServiceArg::DigitalOceanSpaces => {
let bucket = self.bucket.ok_or("--bucket is required for S3-compatible services")?;
let region = self.region.ok_or("--region is required for S3-compatible services")?;
let access_key_id = self
.access_key_id
.ok_or("--access-key-id is required for S3")?;
.ok_or("--access-key-id is required for S3-compatible services")?;
let secret_access_key = self
.secret_access_key
.ok_or("--secret-access-key is required for S3")?;
.ok_or("--secret-access-key is required for S3-compatible services")?;
CloudStorageConfig::S3 {
bucket,
@@ -89,11 +128,106 @@ impl VolumeAddCloudArgs {
endpoint: self.endpoint,
}
}
_ => {
return Err(format!(
"Service {:?} is not yet supported. Only S3 is currently available.",
self.service
))
CloudServiceArg::GoogleDrive => {
let access_token = self
.access_token
.ok_or("--access-token is required for Google Drive")?;
let refresh_token = self
.refresh_token
.ok_or("--refresh-token is required for Google Drive")?;
let client_id = self
.client_id
.ok_or("--client-id is required for Google Drive")?;
let client_secret = self
.client_secret
.ok_or("--client-secret is required for Google Drive")?;
CloudStorageConfig::GoogleDrive {
root: self.root,
access_token,
refresh_token,
client_id,
client_secret,
}
}
CloudServiceArg::OneDrive => {
let access_token = self
.access_token
.ok_or("--access-token is required for OneDrive")?;
let refresh_token = self
.refresh_token
.ok_or("--refresh-token is required for OneDrive")?;
let client_id = self
.client_id
.ok_or("--client-id is required for OneDrive")?;
let client_secret = self
.client_secret
.ok_or("--client-secret is required for OneDrive")?;
CloudStorageConfig::OneDrive {
root: self.root,
access_token,
refresh_token,
client_id,
client_secret,
}
}
CloudServiceArg::Dropbox => {
let access_token = self
.access_token
.ok_or("--access-token is required for Dropbox")?;
let refresh_token = self
.refresh_token
.ok_or("--refresh-token is required for Dropbox")?;
let client_id = self.client_id.ok_or("--client-id is required for Dropbox")?;
let client_secret = self
.client_secret
.ok_or("--client-secret is required for Dropbox")?;
CloudStorageConfig::Dropbox {
root: self.root,
access_token,
refresh_token,
client_id,
client_secret,
}
}
CloudServiceArg::AzureBlob => {
let container = self
.container
.ok_or("--container is required for Azure Blob")?;
let account_name = self
.account_name
.ok_or("--account-name is required for Azure Blob")?;
let account_key = self
.account_key
.ok_or("--account-key is required for Azure Blob")?;
CloudStorageConfig::AzureBlob {
container,
endpoint: self.endpoint,
account_name,
account_key,
}
}
CloudServiceArg::GoogleCloudStorage => {
let bucket = self
.bucket
.ok_or("--bucket is required for Google Cloud Storage")?;
let service_account_path = self
.service_account
.ok_or("--service-account is required for Google Cloud Storage")?;
let credential = std::fs::read_to_string(&service_account_path).map_err(|e| {
format!("Failed to read service account file '{}': {}", service_account_path, e)
})?;
CloudStorageConfig::GoogleCloudStorage {
bucket,
root: self.root,
endpoint: self.endpoint,
credential,
}
}
};

9
core/Cargo.toml
View File

@@ -73,7 +73,14 @@ notify = "6.1" # File system watching
sha2 = "0.10" # SHA-256 hashing for CAS IDs
# Cloud storage integration
opendal = { version = "0.54", features = ["services-s3"] }
opendal = { version = "0.54", features = [
"services-s3",
"services-gdrive",
"services-onedrive",
"services-dropbox",
"services-azblob",
"services-gcs",
] }
# Logging
tracing = "0.1"

10
core/src/crypto/cloud_credentials.rs
View File

@@ -261,6 +261,16 @@ impl CloudCredential {
}
}
/// Create a new API key credential
pub fn new_api_key(service: crate::volume::CloudServiceType, api_key: String) -> Self {
Self {
service,
data: CredentialData::ApiKey(api_key),
created_at: chrono::Utc::now(),
expires_at: None,
}
}
/// Check if this credential is expired
pub fn is_expired(&self) -> bool {
if let Some(expires_at) = self.expires_at {

1
core/src/domain/volume.rs
View File

@@ -148,6 +148,7 @@ pub struct SpacedriveVolumeId {
pub device_name: Option<String>,
pub volume_name: String,
pub device_id: Uuid,
pub library_id: Uuid, // TODO: Populate this, super helpful when another library comes across this file. Thinking about it now we should probably make this file accept multiple of these entries in case two libraries need to track the same volume.
}
/// Summary information about a volume (for updates and caching)

955
core/src/infra/db/migration/m20240101_000001_initial_schema.rs Normal file
View File

@@ -0,0 +1,955 @@
//! Initial database schema for Spacedrive V2
//!
//! This migration creates all the tables needed for the pure hierarchical
//! virtual location model with closure table support.
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Create libraries table
manager
.create_table(
Table::create()
.table(Libraries::Table)
.if_not_exists()
.col(
ColumnDef::new(Libraries::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(Libraries::Uuid)
.uuid()
.not_null()
.unique_key(),
)
.col(ColumnDef::new(Libraries::Name).string().not_null())
.col(ColumnDef::new(Libraries::DbVersion).integer().not_null())
.col(ColumnDef::new(Libraries::SyncId).uuid())
.col(
ColumnDef::new(Libraries::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(Libraries::UpdatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.to_owned(),
)
.await?;
// Create devices table
manager
.create_table(
Table::create()
.table(Devices::Table)
.if_not_exists()
.col(
ColumnDef::new(Devices::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(ColumnDef::new(Devices::Uuid).uuid().not_null().unique_key())
.col(ColumnDef::new(Devices::Name).string().not_null())
.col(ColumnDef::new(Devices::Os).string().not_null())
.col(ColumnDef::new(Devices::OsVersion).string())
.col(ColumnDef::new(Devices::HardwareModel).string())
.col(ColumnDef::new(Devices::NetworkAddresses).json().not_null())
.col(ColumnDef::new(Devices::IsOnline).boolean().not_null())
.col(
ColumnDef::new(Devices::LastSeenAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(ColumnDef::new(Devices::Capabilities).json().not_null())
.col(
ColumnDef::new(Devices::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(Devices::UpdatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.to_owned(),
)
.await?;
// Create user_metadata table (modern schema for semantic tagging)
manager
.create_table(
Table::create()
.table(UserMetadata::Table)
.if_not_exists()
.col(
ColumnDef::new(UserMetadata::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(UserMetadata::Uuid)
.uuid()
.not_null()
.unique_key(),
)
// Exactly one of these is set - defines the scope
.col(ColumnDef::new(UserMetadata::EntryUuid).uuid()) // File-specific metadata (higher priority)
.col(ColumnDef::new(UserMetadata::ContentIdentityUuid).uuid()) // Content-universal metadata (lower priority)
// All metadata types benefit from scope flexibility
.col(ColumnDef::new(UserMetadata::Notes).text())
.col(
ColumnDef::new(UserMetadata::Favorite)
.boolean()
.default(false),
)
.col(
ColumnDef::new(UserMetadata::Hidden)
.boolean()
.default(false),
)
.col(ColumnDef::new(UserMetadata::CustomData).json().not_null()) // Arbitrary JSON data
.col(
ColumnDef::new(UserMetadata::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(UserMetadata::UpdatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.to_owned(),
)
.await?;
// Create mime_types table (lookup table)
manager
.create_table(
Table::create()
.table(MimeTypes::Table)
.if_not_exists()
.col(
ColumnDef::new(MimeTypes::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(ColumnDef::new(MimeTypes::Uuid).uuid().not_null())
.col(
ColumnDef::new(MimeTypes::MimeType)
.string()
.not_null()
.unique_key(),
)
.col(
ColumnDef::new(MimeTypes::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.to_owned(),
)
.await?;
// Create content_kinds table (lookup table)
manager
.create_table(
Table::create()
.table(ContentKinds::Table)
.if_not_exists()
.col(
ColumnDef::new(ContentKinds::Id)
.integer()
.not_null()
.primary_key(),
)
.col(ColumnDef::new(ContentKinds::Name).string().not_null())
.to_owned(),
)
.await?;
// Create content_identities table
manager
.create_table(
Table::create()
.table(ContentIdentities::Table)
.if_not_exists()
.col(
ColumnDef::new(ContentIdentities::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(ContentIdentities::Uuid)
.uuid()
.not_null()
.unique_key(),
)
.col(ColumnDef::new(ContentIdentities::IntegrityHash).string())
.col(
ColumnDef::new(ContentIdentities::ContentHash)
.string()
.not_null()
.unique_key(),
)
.col(ColumnDef::new(ContentIdentities::MimeTypeId).integer())
.col(
ColumnDef::new(ContentIdentities::KindId)
.integer()
.not_null(),
)
.col(ColumnDef::new(ContentIdentities::MediaData).json())
.col(ColumnDef::new(ContentIdentities::TextContent).text())
.col(
ColumnDef::new(ContentIdentities::TotalSize)
.big_integer()
.not_null(),
)
.col(
ColumnDef::new(ContentIdentities::EntryCount)
.integer()
.not_null()
.default(1),
)
.col(
ColumnDef::new(ContentIdentities::FirstSeenAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(ContentIdentities::LastVerifiedAt)
.timestamp_with_time_zone()
.not_null(),
)
.foreign_key(
ForeignKey::create()
.from(ContentIdentities::Table, ContentIdentities::MimeTypeId)
.to(MimeTypes::Table, MimeTypes::Id)
.on_delete(ForeignKeyAction::SetNull),
)
.foreign_key(
ForeignKey::create()
.from(ContentIdentities::Table, ContentIdentities::KindId)
.to(ContentKinds::Table, ContentKinds::Id)
.on_delete(ForeignKeyAction::Restrict),
)
.to_owned(),
)
.await?;
// Create entries table - This is the core of our hierarchical model
manager
.create_table(
Table::create()
.table(Entries::Table)
.if_not_exists()
.col(
ColumnDef::new(Entries::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(ColumnDef::new(Entries::Uuid).uuid())
.col(ColumnDef::new(Entries::Name).string().not_null())
.col(ColumnDef::new(Entries::Kind).integer().not_null())
.col(ColumnDef::new(Entries::Extension).string())
.col(ColumnDef::new(Entries::MetadataId).integer())
.col(ColumnDef::new(Entries::ContentId).integer())
.col(ColumnDef::new(Entries::Size).big_integer().not_null())
.col(
ColumnDef::new(Entries::AggregateSize)
.big_integer()
.not_null(),
)
.col(ColumnDef::new(Entries::ChildCount).integer().not_null())
.col(ColumnDef::new(Entries::FileCount).integer().not_null())
.col(
ColumnDef::new(Entries::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(Entries::ModifiedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(ColumnDef::new(Entries::AccessedAt).timestamp_with_time_zone())
.col(ColumnDef::new(Entries::Permissions).string())
.col(ColumnDef::new(Entries::Inode).big_integer())
.col(ColumnDef::new(Entries::ParentId).integer())
.foreign_key(
ForeignKey::create()
.from(Entries::Table, Entries::MetadataId)
.to(UserMetadata::Table, UserMetadata::Id)
.on_delete(ForeignKeyAction::SetNull),
)
.foreign_key(
ForeignKey::create()
.from(Entries::Table, Entries::ContentId)
.to(ContentIdentities::Table, ContentIdentities::Id)
.on_delete(ForeignKeyAction::SetNull),
)
.to_owned(),
)
.await?;
// Create entry_closure table for efficient hierarchical queries
manager
.create_table(
Table::create()
.table(EntryClosure::Table)
.if_not_exists()
.col(
ColumnDef::new(EntryClosure::AncestorId)
.integer()
.not_null(),
)
.col(
ColumnDef::new(EntryClosure::DescendantId)
.integer()
.not_null(),
)
.col(ColumnDef::new(EntryClosure::Depth).integer().not_null())
.primary_key(
Index::create()
.col(EntryClosure::AncestorId)
.col(EntryClosure::DescendantId),
)
.foreign_key(
ForeignKey::create()
.from(EntryClosure::Table, EntryClosure::AncestorId)
.to(Entries::Table, Entries::Id)
.on_delete(ForeignKeyAction::Cascade),
)
.foreign_key(
ForeignKey::create()
.from(EntryClosure::Table, EntryClosure::DescendantId)
.to(Entries::Table, Entries::Id)
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create directory_paths table for caching directory paths
manager
.create_table(
Table::create()
.table(DirectoryPaths::Table)
.if_not_exists()
.col(
ColumnDef::new(DirectoryPaths::EntryId)
.integer()
.primary_key(),
)
.col(ColumnDef::new(DirectoryPaths::Path).text().not_null())
.foreign_key(
ForeignKey::create()
.from(DirectoryPaths::Table, DirectoryPaths::EntryId)
.to(Entries::Table, Entries::Id)
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create locations table - Now points to entries instead of storing paths
manager
.create_table(
Table::create()
.table(Locations::Table)
.if_not_exists()
.col(
ColumnDef::new(Locations::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(Locations::Uuid)
.uuid()
.not_null()
.unique_key(),
)
.col(ColumnDef::new(Locations::DeviceId).integer().not_null())
.col(ColumnDef::new(Locations::EntryId).integer().not_null())
.col(ColumnDef::new(Locations::Name).string())
.col(ColumnDef::new(Locations::IndexMode).string().not_null())
.col(ColumnDef::new(Locations::ScanState).string().not_null())
.col(ColumnDef::new(Locations::LastScanAt).timestamp_with_time_zone())
.col(ColumnDef::new(Locations::ErrorMessage).text())
.col(
ColumnDef::new(Locations::TotalFileCount)
.integer()
.not_null(),
)
.col(
ColumnDef::new(Locations::TotalByteSize)
.big_integer()
.not_null(),
)
.col(
ColumnDef::new(Locations::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(Locations::UpdatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.foreign_key(
ForeignKey::create()
.from(Locations::Table, Locations::DeviceId)
.to(Devices::Table, Devices::Id)
.on_delete(ForeignKeyAction::Restrict),
)
.foreign_key(
ForeignKey::create()
.from(Locations::Table, Locations::EntryId)
.to(Entries::Table, Entries::Id)
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create volumes table
manager
.create_table(
Table::create()
.table(Volumes::Table)
.if_not_exists()
.col(
ColumnDef::new(Volumes::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(ColumnDef::new(Volumes::Uuid).uuid().not_null())
.col(ColumnDef::new(Volumes::DeviceId).uuid().not_null())
.col(ColumnDef::new(Volumes::Fingerprint).string().not_null())
.col(ColumnDef::new(Volumes::MountPoint).string())
.col(ColumnDef::new(Volumes::TotalCapacity).big_integer())
.col(ColumnDef::new(Volumes::AvailableCapacity).big_integer())
.col(ColumnDef::new(Volumes::IsRemovable).boolean())
.col(ColumnDef::new(Volumes::IsEjectable).boolean())
.col(ColumnDef::new(Volumes::FileSystem).string())
.col(ColumnDef::new(Volumes::DisplayName).string())
.col(
ColumnDef::new(Volumes::TrackedAt)
.timestamp_with_time_zone()
.not_null()
.default(Expr::current_timestamp()),
)
.col(
ColumnDef::new(Volumes::LastSeenAt)
.timestamp_with_time_zone()
.not_null()
.default(Expr::current_timestamp()),
)
.col(
ColumnDef::new(Volumes::IsOnline)
.boolean()
.not_null()
.default(true),
)
.col(ColumnDef::new(Volumes::ReadSpeedMbps).integer())
.col(ColumnDef::new(Volumes::WriteSpeedMbps).integer())
.col(ColumnDef::new(Volumes::LastSpeedTestAt).timestamp_with_time_zone())
.col(ColumnDef::new(Volumes::IsNetworkDrive).boolean())
.col(ColumnDef::new(Volumes::DeviceModel).string())
.col(ColumnDef::new(Volumes::VolumeType).string())
.col(ColumnDef::new(Volumes::IsUserVisible).boolean())
.col(ColumnDef::new(Volumes::AutoTrackEligible).boolean())
.col(
ColumnDef::new(Volumes::CreatedAt)
.timestamp_with_time_zone()
.not_null()
.default(Expr::current_timestamp()),
)
.col(
ColumnDef::new(Volumes::UpdatedAt)
.timestamp_with_time_zone()
.not_null()
.default(Expr::current_timestamp()),
)
.foreign_key(
ForeignKey::create()
.from(Volumes::Table, Volumes::DeviceId)
.to(Devices::Table, Devices::Uuid)
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create audit_log table
manager
.create_table(
Table::create()
.table(AuditLog::Table)
.if_not_exists()
.col(
ColumnDef::new(AuditLog::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(AuditLog::Uuid)
.string()
.not_null()
.unique_key(),
)
.col(ColumnDef::new(AuditLog::ActionType).string().not_null())
.col(ColumnDef::new(AuditLog::ActorDeviceId).string().not_null())
.col(ColumnDef::new(AuditLog::Targets).string().not_null())
.col(ColumnDef::new(AuditLog::Status).string().not_null())
.col(ColumnDef::new(AuditLog::JobId).string())
.col(
ColumnDef::new(AuditLog::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(ColumnDef::new(AuditLog::CompletedAt).timestamp_with_time_zone())
.col(ColumnDef::new(AuditLog::ErrorMessage).string())
.col(ColumnDef::new(AuditLog::ResultPayload).string())
.to_owned(),
)
.await?;
// Create sync_checkpoints table
manager
.create_table(
Table::create()
.table(SyncCheckpoints::Table)
.if_not_exists()
.col(
ColumnDef::new(SyncCheckpoints::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(SyncCheckpoints::DeviceId)
.integer()
.not_null()
.unique_key(),
)
.col(
ColumnDef::new(SyncCheckpoints::LastSync)
.timestamp_with_time_zone()
.not_null(),
)
.col(ColumnDef::new(SyncCheckpoints::SyncData).json())
.col(
ColumnDef::new(SyncCheckpoints::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(SyncCheckpoints::UpdatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.foreign_key(
ForeignKey::create()
.from(SyncCheckpoints::Table, SyncCheckpoints::DeviceId)
.to(Devices::Table, Devices::Id)
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create indices for better query performance
// Entry indices
manager
.create_index(
Index::create()
.name("idx_entries_uuid")
.table(Entries::Table)
.col(Entries::Uuid)
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_entries_parent_id")
.table(Entries::Table)
.col(Entries::ParentId)
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_entries_kind")
.table(Entries::Table)
.col(Entries::Kind)
.to_owned(),
)
.await?;
// Entry closure indices for efficient queries
manager
.create_index(
Index::create()
.name("idx_entry_closure_descendant")
.table(EntryClosure::Table)
.col(EntryClosure::DescendantId)
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_entry_closure_ancestor_depth")
.table(EntryClosure::Table)
.col(EntryClosure::AncestorId)
.col(EntryClosure::Depth)
.to_owned(),
)
.await?;
// Location indices
manager
.create_index(
Index::create()
.name("idx_locations_entry_id")
.table(Locations::Table)
.col(Locations::EntryId)
.to_owned(),
)
.await?;
// Content identity index
manager
.create_index(
Index::create()
.name("idx_content_identities_content_hash")
.table(ContentIdentities::Table)
.col(ContentIdentities::ContentHash)
.to_owned(),
)
.await?;
// Volume indices
manager
.create_index(
Index::create()
.name("idx_volumes_device_fingerprint")
.table(Volumes::Table)
.col(Volumes::DeviceId)
.col(Volumes::Fingerprint)
.unique()
.to_owned(),
)
.await?;
// Audit log indices
manager
.create_index(
Index::create()
.name("idx_audit_log_action_type")
.table(AuditLog::Table)
.col(AuditLog::ActionType)
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_audit_log_actor_device")
.table(AuditLog::Table)
.col(AuditLog::ActorDeviceId)
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_audit_log_status")
.table(AuditLog::Table)
.col(AuditLog::Status)
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_audit_log_job_id")
.table(AuditLog::Table)
.col(AuditLog::JobId)
.to_owned(),
)
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Drop tables in reverse order of creation
manager
.drop_table(Table::drop().table(SyncCheckpoints::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(AuditLog::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(Volumes::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(Locations::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(DirectoryPaths::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(EntryClosure::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(Entries::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(ContentIdentities::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(ContentKinds::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(MimeTypes::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(UserMetadata::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(Devices::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(Libraries::Table).to_owned())
.await?;
Ok(())
}
}
// Table identifiers
#[derive(DeriveIden)]
enum Libraries {
Table,
Id,
Uuid,
Name,
DbVersion,
SyncId,
CreatedAt,
UpdatedAt,
}
#[derive(DeriveIden)]
enum Devices {
Table,
Id,
Uuid,
Name,
Os,
OsVersion,
HardwareModel,
NetworkAddresses,
IsOnline,
LastSeenAt,
Capabilities,
CreatedAt,
UpdatedAt,
}
#[derive(DeriveIden)]
enum MimeTypes {
Table,
Id,
Uuid,
MimeType,
CreatedAt,
}
#[derive(DeriveIden)]
enum ContentKinds {
Table,
Id,
Name,
}
#[derive(DeriveIden)]
enum UserMetadata {
Table,
Id,
Uuid,
EntryUuid,
ContentIdentityUuid,
Notes,
Favorite,
Hidden,
CustomData,
CreatedAt,
UpdatedAt,
}
#[derive(DeriveIden)]
enum ContentIdentities {
Table,
Id,
Uuid,
IntegrityHash,
ContentHash,
MimeTypeId,
KindId,
MediaData,
TextContent,
TotalSize,
EntryCount,
FirstSeenAt,
LastVerifiedAt,
}
#[derive(DeriveIden)]
enum Entries {
Table,
Id,
Uuid,
Name,
Kind,
Extension,
MetadataId,
ContentId,
Size,
AggregateSize,
ChildCount,
FileCount,
CreatedAt,
ModifiedAt,
AccessedAt,
Permissions,
Inode,
ParentId,
}
#[derive(DeriveIden)]
enum EntryClosure {
Table,
AncestorId,
DescendantId,
Depth,
}
#[derive(DeriveIden)]
enum DirectoryPaths {
Table,
EntryId,
Path,
}
#[derive(DeriveIden)]
enum Locations {
Table,
Id,
Uuid,
DeviceId,
EntryId,
Name,
IndexMode,
ScanState,
LastScanAt,
ErrorMessage,
TotalFileCount,
TotalByteSize,
CreatedAt,
UpdatedAt,
}
#[derive(DeriveIden)]
enum Volumes {
Table,
Id,
Uuid,
DeviceId,
Fingerprint,
DisplayName,
MountPoint,
TotalCapacity,
AvailableCapacity,
IsRemovable,
IsEjectable,
FileSystem,
TrackedAt,
LastSeenAt,
IsOnline,
ReadSpeedMbps,
WriteSpeedMbps,
LastSpeedTestAt,
IsNetworkDrive,
DeviceModel,
VolumeType,
IsUserVisible,
AutoTrackEligible,
CreatedAt,
UpdatedAt,
}
#[derive(DeriveIden)]
enum AuditLog {
Table,
Id,
Uuid,
ActionType,
ActorDeviceId,
Targets,
Status,
JobId,
CreatedAt,
CompletedAt,
ErrorMessage,
ResultPayload,
}
#[derive(DeriveIden)]
enum SyncCheckpoints {
Table,
Id,
DeviceId,
LastSync,
SyncData,
CreatedAt,
UpdatedAt,
}

2064
core/src/infra/db/migration/m20240101_000001_unified_schema.rs
View File

File diff suppressed because it is too large Load Diff

53
core/src/infra/db/migration/m20240102_000001_populate_lookups.rs Normal file
View File

@@ -0,0 +1,53 @@
//! Populate lookup tables with initial data
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Populate content_kinds table
let insert_kinds = Query::insert()
.into_table(ContentKinds::Table)
.columns([ContentKinds::Id, ContentKinds::Name])
.values_panic([0.into(), "unknown".into()])
.values_panic([1.into(), "image".into()])
.values_panic([2.into(), "video".into()])
.values_panic([3.into(), "audio".into()])
.values_panic([4.into(), "document".into()])
.values_panic([5.into(), "archive".into()])
.values_panic([6.into(), "code".into()])
.values_panic([7.into(), "text".into()])
.values_panic([8.into(), "database".into()])
.values_panic([9.into(), "book".into()])
.values_panic([10.into(), "font".into()])
.values_panic([11.into(), "mesh".into()])
.values_panic([12.into(), "config".into()])
.values_panic([13.into(), "encrypted".into()])
.values_panic([14.into(), "key".into()])
.values_panic([15.into(), "executable".into()])
.values_panic([16.into(), "binary".into()])
.to_owned();
manager.exec_stmt(insert_kinds).await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Delete all content kinds
let delete = Query::delete().from_table(ContentKinds::Table).to_owned();
manager.exec_stmt(delete).await?;
Ok(())
}
}
#[derive(DeriveIden)]
enum ContentKinds {
Table,
Id,
Name,
}

156
core/src/infra/db/migration/m20240107_000001_create_collections.rs Normal file
View File

@@ -0,0 +1,156 @@
use sea_orm_migration::prelude::*;
pub struct Migration;
impl MigrationName for Migration {
fn name(&self) -> &str {
"m20240107_000001_create_collections"
}
}
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Create collections table
manager
.create_table(
Table::create()
.table(Collection::Table)
.if_not_exists()
.col(
ColumnDef::new(Collection::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(Collection::Uuid)
.uuid()
.not_null()
.unique_key(),
)
.col(ColumnDef::new(Collection::Name).string().not_null())
.col(ColumnDef::new(Collection::Description).text().null())
.col(
ColumnDef::new(Collection::CreatedAt)
.timestamp()
.not_null()
.default(Expr::current_timestamp()),
)
.col(
ColumnDef::new(Collection::UpdatedAt)
.timestamp()
.not_null()
.default(Expr::current_timestamp()),
)
.to_owned(),
)
.await?;
// Create collection_entries junction table
manager
.create_table(
Table::create()
.table(CollectionEntry::Table)
.if_not_exists()
.col(
ColumnDef::new(CollectionEntry::CollectionId)
.integer()
.not_null(),
)
.col(
ColumnDef::new(CollectionEntry::EntryId)
.integer()
.not_null(),
)
.col(
ColumnDef::new(CollectionEntry::AddedAt)
.timestamp()
.not_null()
.default(Expr::current_timestamp()),
)
.primary_key(
Index::create()
.col(CollectionEntry::CollectionId)
.col(CollectionEntry::EntryId),
)
.foreign_key(
ForeignKey::create()
.name("fk_collection_entry_collection")
.from(CollectionEntry::Table, CollectionEntry::CollectionId)
.to(Collection::Table, Collection::Id)
.on_delete(ForeignKeyAction::Cascade),
)
.foreign_key(
ForeignKey::create()
.name("fk_collection_entry_entry")
.from(CollectionEntry::Table, CollectionEntry::EntryId)
.to(Entry::Table, Entry::Id)
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create indexes
manager
.create_index(
Index::create()
.name("idx_collection_name")
.table(Collection::Table)
.col(Collection::Name)
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_collection_entry_entry_id")
.table(CollectionEntry::Table)
.col(CollectionEntry::EntryId)
.to_owned(),
)
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.drop_table(Table::drop().table(CollectionEntry::Table).to_owned())
.await?;
manager
.drop_table(Table::drop().table(Collection::Table).to_owned())
.await?;
Ok(())
}
}
#[derive(Iden)]
enum Collection {
Table,
Id,
Uuid,
Name,
Description,
CreatedAt,
UpdatedAt,
}
#[derive(Iden)]
enum CollectionEntry {
Table,
CollectionId,
EntryId,
AddedAt,
}
#[derive(Iden)]
enum Entry {
Table,
Id,
}

248
core/src/infra/db/migration/m20250109_000001_create_sidecars.rs Normal file
View File

@@ -0,0 +1,248 @@
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Create sidecars table
manager
.create_table(
Table::create()
.table(Sidecar::Table)
.if_not_exists()
.col(
ColumnDef::new(Sidecar::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(ColumnDef::new(Sidecar::ContentUuid).uuid().not_null())
.col(ColumnDef::new(Sidecar::Kind).string().not_null())
.col(ColumnDef::new(Sidecar::Variant).string().not_null())
.col(ColumnDef::new(Sidecar::Format).string().not_null())
.col(ColumnDef::new(Sidecar::RelPath).string().not_null())
.col(ColumnDef::new(Sidecar::SourceEntryId).integer().null())
.col(ColumnDef::new(Sidecar::Size).big_integer().not_null())
.col(ColumnDef::new(Sidecar::Checksum).string().null())
.col(
ColumnDef::new(Sidecar::Status)
.string()
.not_null()
.default("pending"),
)
.col(ColumnDef::new(Sidecar::Source).string().null())
.col(
ColumnDef::new(Sidecar::Version)
.integer()
.not_null()
.default(1),
)
.col(
ColumnDef::new(Sidecar::CreatedAt)
.timestamp()
.not_null()
.default(Expr::current_timestamp()),
)
.col(
ColumnDef::new(Sidecar::UpdatedAt)
.timestamp()
.not_null()
.default(Expr::current_timestamp()),
)
.foreign_key(
ForeignKey::create()
.name("fk_sidecar_content")
.from(Sidecar::Table, Sidecar::ContentUuid)
.to(ContentIdentities::Table, ContentIdentities::Uuid)
.on_delete(ForeignKeyAction::Cascade)
.on_update(ForeignKeyAction::Cascade),
)
.foreign_key(
ForeignKey::create()
.name("fk_sidecar_source_entry")
.from(Sidecar::Table, Sidecar::SourceEntryId)
.to(Entries::Table, Entries::Id)
.on_delete(ForeignKeyAction::SetNull)
.on_update(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create unique index on (content_uuid, kind, variant)
manager
.create_index(
Index::create()
.if_not_exists()
.name("idx_sidecar_unique")
.table(Sidecar::Table)
.col(Sidecar::ContentUuid)
.col(Sidecar::Kind)
.col(Sidecar::Variant)
.unique()
.to_owned(),
)
.await?;
// Create sidecar_availability table
manager
.create_table(
Table::create()
.table(SidecarAvailability::Table)
.if_not_exists()
.col(
ColumnDef::new(SidecarAvailability::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(SidecarAvailability::ContentUuid)
.uuid()
.not_null(),
)
.col(
ColumnDef::new(SidecarAvailability::Kind)
.string()
.not_null(),
)
.col(
ColumnDef::new(SidecarAvailability::Variant)
.string()
.not_null(),
)
.col(
ColumnDef::new(SidecarAvailability::DeviceUuid)
.uuid()
.not_null(),
)
.col(
ColumnDef::new(SidecarAvailability::Has)
.boolean()
.not_null()
.default(false),
)
.col(
ColumnDef::new(SidecarAvailability::Size)
.big_integer()
.null(),
)
.col(
ColumnDef::new(SidecarAvailability::Checksum)
.string()
.null(),
)
.col(
ColumnDef::new(SidecarAvailability::LastSeenAt)
.timestamp()
.not_null()
.default(Expr::current_timestamp()),
)
.foreign_key(
ForeignKey::create()
.name("fk_sidecar_availability_content")
.from(SidecarAvailability::Table, SidecarAvailability::ContentUuid)
.to(ContentIdentities::Table, ContentIdentities::Uuid)
.on_delete(ForeignKeyAction::Cascade)
.on_update(ForeignKeyAction::Cascade),
)
.foreign_key(
ForeignKey::create()
.name("fk_sidecar_availability_device")
.from(SidecarAvailability::Table, SidecarAvailability::DeviceUuid)
.to(Devices::Table, Devices::Uuid)
.on_delete(ForeignKeyAction::Cascade)
.on_update(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create unique index on (content_uuid, kind, variant, device_uuid)
manager
.create_index(
Index::create()
.if_not_exists()
.name("idx_sidecar_availability_unique")
.table(SidecarAvailability::Table)
.col(SidecarAvailability::ContentUuid)
.col(SidecarAvailability::Kind)
.col(SidecarAvailability::Variant)
.col(SidecarAvailability::DeviceUuid)
.unique()
.to_owned(),
)
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Drop sidecar_availability table
manager
.drop_table(Table::drop().table(SidecarAvailability::Table).to_owned())
.await?;
// Drop sidecars table
manager
.drop_table(Table::drop().table(Sidecar::Table).to_owned())
.await?;
Ok(())
}
}
#[derive(Iden)]
enum Sidecar {
Table,
Id,
ContentUuid,
Kind,
Variant,
Format,
RelPath,
SourceEntryId,
Size,
Checksum,
Status,
Source,
Version,
CreatedAt,
UpdatedAt,
}
#[derive(Iden)]
enum SidecarAvailability {
Table,
Id,
ContentUuid,
Kind,
Variant,
DeviceUuid,
Has,
Size,
Checksum,
LastSeenAt,
}
#[derive(Iden)]
enum ContentIdentities {
Table,
Uuid,
}
#[derive(Iden)]
enum Devices {
Table,
Uuid,
}
#[derive(Iden)]
enum Entries {
Table,
Id,
}

194
core/src/infra/db/migration/m20250110_000001_refactor_volumes_table.rs Normal file
View File

@@ -0,0 +1,194 @@
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// For SQLite, we can't easily alter columns, so we'll just add the UUID column
// if the table exists with the old schema
// Try to add UUID column to existing table
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(
ColumnDef::new(Volumes::Uuid)
.string() // SQLite doesn't have native UUID type
.not_null()
.default(""), // Will be populated later
)
.to_owned(),
)
.await;
// Add other missing columns one by one (SQLite limitation)
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::Fingerprint).string())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::DisplayName).string())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(
ColumnDef::new(Volumes::TrackedAt)
.timestamp()
.not_null()
.default(Expr::current_timestamp()),
)
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::LastSpeedTestAt).timestamp())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::ReadSpeedMbps).integer())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::WriteSpeedMbps).integer())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(
ColumnDef::new(Volumes::IsOnline).boolean().default(true),
)
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::IsNetworkDrive).boolean())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::DeviceModel).string())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::VolumeType).string())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::IsUserVisible).boolean())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(ColumnDef::new(Volumes::AutoTrackEligible).boolean())
.to_owned(),
)
.await;
let _ = manager
.alter_table(
Table::alter()
.table(Volumes::Table)
.add_column_if_not_exists(
ColumnDef::new(Volumes::LastSeenAt)
.timestamp()
.not_null()
.default(Expr::current_timestamp()),
)
.to_owned(),
)
.await;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Remove added columns
// Note: SQLite doesn't support dropping columns easily
Ok(())
}
}
#[derive(DeriveIden)]
enum Volumes {
Table,
Id,
Uuid,
DeviceId,
Fingerprint,
DisplayName,
MountPoint,
TotalCapacity,
AvailableCapacity,
ReadSpeedMbps,
WriteSpeedMbps,
IsRemovable,
IsEjectable,
IsOnline,
IsNetworkDrive,
FileSystemType,
DeviceModel,
VolumeType,
IsUserVisible,
AutoTrackEligible,
TrackedAt,
LastSeenAt,
LastSpeedTestAt,
CreatedAt,
UpdatedAt,
}

63
core/src/infra/db/migration/m20250112_000001_create_indexer_rules.rs Normal file
View File

@@ -0,0 +1,63 @@
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.create_table(
Table::create()
.table(IndexerRules::Table)
.if_not_exists()
.col(
ColumnDef::new(IndexerRules::Id)
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(IndexerRules::Name)
.string()
.not_null()
.unique_key(),
)
.col(ColumnDef::new(IndexerRules::Default).boolean().not_null())
.col(ColumnDef::new(IndexerRules::RulesBlob).binary().not_null())
.col(
ColumnDef::new(IndexerRules::CreatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(IndexerRules::UpdatedAt)
.timestamp_with_time_zone()
.not_null(),
)
.to_owned(),
)
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.drop_table(Table::drop().table(IndexerRules::Table).to_owned())
.await?;
Ok(())
}
}
#[derive(DeriveIden)]
enum IndexerRules {
Table,
Id,
Name,
Default,
RulesBlob,
CreatedAt,
UpdatedAt,
}

583
core/src/infra/db/migration/m20250115_000001_semantic_tags.rs Normal file
View File

@@ -0,0 +1,583 @@
//! Migration: Create semantic tagging system
//!
//! This migration creates the complete semantic tagging infrastructure:
//! - Enhanced tag table with polymorphic naming
//! - Hierarchical relationships with closure table
//! - Context-aware tag applications
//! - Usage pattern tracking for intelligent suggestions
//! - Full-text search across all tag variants
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Create the enhanced tag table
manager
.create_table(
Table::create()
.table(Alias::new("tag"))
.if_not_exists()
.col(
ColumnDef::new(Alias::new("id"))
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(Alias::new("uuid"))
.uuid()
.not_null()
.unique_key(),
)
.col(
ColumnDef::new(Alias::new("canonical_name"))
.string()
.not_null(),
)
.col(ColumnDef::new(Alias::new("display_name")).string())
.col(ColumnDef::new(Alias::new("formal_name")).string())
.col(ColumnDef::new(Alias::new("abbreviation")).string())
.col(ColumnDef::new(Alias::new("aliases")).json())
.col(ColumnDef::new(Alias::new("namespace")).string())
.col(
ColumnDef::new(Alias::new("tag_type"))
.string()
.not_null()
.default("standard"),
)
.col(ColumnDef::new(Alias::new("color")).string())
.col(ColumnDef::new(Alias::new("icon")).string())
.col(ColumnDef::new(Alias::new("description")).text())
.col(
ColumnDef::new(Alias::new("is_organizational_anchor"))
.boolean()
.default(false),
)
.col(
ColumnDef::new(Alias::new("privacy_level"))
.string()
.default("normal"),
)
.col(
ColumnDef::new(Alias::new("search_weight"))
.integer()
.default(100),
)
.col(ColumnDef::new(Alias::new("attributes")).json())
.col(ColumnDef::new(Alias::new("composition_rules")).json())
.col(
ColumnDef::new(Alias::new("created_at"))
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(Alias::new("updated_at"))
.timestamp_with_time_zone()
.not_null(),
)
.col(ColumnDef::new(Alias::new("created_by_device")).uuid())
.to_owned(),
)
.await?;
// Create indexes for the tag table
manager
.create_index(
Index::create()
.name("idx_tag_canonical_name")
.table(Alias::new("tag"))
.col(Alias::new("canonical_name"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_tag_namespace")
.table(Alias::new("tag"))
.col(Alias::new("namespace"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_tag_type")
.table(Alias::new("tag"))
.col(Alias::new("tag_type"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_tag_privacy_level")
.table(Alias::new("tag"))
.col(Alias::new("privacy_level"))
.to_owned(),
)
.await?;
// Create the tag_relationship table
manager
.create_table(
Table::create()
.table(Alias::new("tag_relationship"))
.if_not_exists()
.col(
ColumnDef::new(Alias::new("id"))
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(Alias::new("parent_tag_id"))
.integer()
.not_null(),
)
.col(
ColumnDef::new(Alias::new("child_tag_id"))
.integer()
.not_null(),
)
.col(
ColumnDef::new(Alias::new("relationship_type"))
.string()
.not_null()
.default("parent_child"),
)
.col(ColumnDef::new(Alias::new("strength")).float().default(1.0))
.col(
ColumnDef::new(Alias::new("created_at"))
.timestamp_with_time_zone()
.not_null(),
)
.foreign_key(
&mut ForeignKey::create()
.name("fk_tag_relationship_parent")
.from(Alias::new("tag_relationship"), Alias::new("parent_tag_id"))
.to(Alias::new("tag"), Alias::new("id"))
.on_delete(ForeignKeyAction::Cascade),
)
.foreign_key(
&mut ForeignKey::create()
.name("fk_tag_relationship_child")
.from(Alias::new("tag_relationship"), Alias::new("child_tag_id"))
.to(Alias::new("tag"), Alias::new("id"))
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create indexes for tag_relationship
manager
.create_index(
Index::create()
.name("idx_tag_relationship_parent")
.table(Alias::new("tag_relationship"))
.col(Alias::new("parent_tag_id"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_tag_relationship_child")
.table(Alias::new("tag_relationship"))
.col(Alias::new("child_tag_id"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_tag_relationship_type")
.table(Alias::new("tag_relationship"))
.col(Alias::new("relationship_type"))
.to_owned(),
)
.await?;
// Create the tag_closure table for efficient hierarchical queries
manager
.create_table(
Table::create()
.table(Alias::new("tag_closure"))
.if_not_exists()
.col(
ColumnDef::new(Alias::new("ancestor_id"))
.integer()
.not_null(),
)
.col(
ColumnDef::new(Alias::new("descendant_id"))
.integer()
.not_null(),
)
.col(ColumnDef::new(Alias::new("depth")).integer().not_null())
.col(
ColumnDef::new(Alias::new("path_strength"))
.float()
.not_null(),
)
.primary_key(
Index::create()
.col(Alias::new("ancestor_id"))
.col(Alias::new("descendant_id")),
)
.foreign_key(
&mut ForeignKey::create()
.name("fk_tag_closure_ancestor")
.from(Alias::new("tag_closure"), Alias::new("ancestor_id"))
.to(Alias::new("tag"), Alias::new("id"))
.on_delete(ForeignKeyAction::Cascade),
)
.foreign_key(
&mut ForeignKey::create()
.name("fk_tag_closure_descendant")
.from(Alias::new("tag_closure"), Alias::new("descendant_id"))
.to(Alias::new("tag"), Alias::new("id"))
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create indexes for tag_closure
manager
.create_index(
Index::create()
.name("idx_tag_closure_ancestor")
.table(Alias::new("tag_closure"))
.col(Alias::new("ancestor_id"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_tag_closure_descendant")
.table(Alias::new("tag_closure"))
.col(Alias::new("descendant_id"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_tag_closure_depth")
.table(Alias::new("tag_closure"))
.col(Alias::new("depth"))
.to_owned(),
)
.await?;
// Create the user_metadata_tag table
manager
.create_table(
Table::create()
.table(Alias::new("user_metadata_tag"))
.if_not_exists()
.col(
ColumnDef::new(Alias::new("id"))
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(
ColumnDef::new(Alias::new("user_metadata_id"))
.integer()
.not_null(),
)
.col(ColumnDef::new(Alias::new("tag_id")).integer().not_null())
.col(ColumnDef::new(Alias::new("applied_context")).string())
.col(ColumnDef::new(Alias::new("applied_variant")).string())
.col(
ColumnDef::new(Alias::new("confidence"))
.float()
.default(1.0),
)
.col(
ColumnDef::new(Alias::new("source"))
.string()
.default("user"),
)
.col(ColumnDef::new(Alias::new("instance_attributes")).json())
.col(
ColumnDef::new(Alias::new("created_at"))
.timestamp_with_time_zone()
.not_null(),
)
.col(
ColumnDef::new(Alias::new("updated_at"))
.timestamp_with_time_zone()
.not_null(),
)
.col(ColumnDef::new(Alias::new("device_uuid")).uuid().not_null())
.foreign_key(
&mut ForeignKey::create()
.name("fk_user_metadata_tag_metadata")
.from(
Alias::new("user_metadata_tag"),
Alias::new("user_metadata_id"),
)
.to(Alias::new("user_metadata"), Alias::new("id"))
.on_delete(ForeignKeyAction::Cascade),
)
.foreign_key(
&mut ForeignKey::create()
.name("fk_user_metadata_tag_tag")
.from(Alias::new("user_metadata_tag"), Alias::new("tag_id"))
.to(Alias::new("tag"), Alias::new("id"))
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create indexes for user_metadata_tag
manager
.create_index(
Index::create()
.name("idx_user_metadata_tag_metadata")
.table(Alias::new("user_metadata_tag"))
.col(Alias::new("user_metadata_id"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_user_metadata_tag_tag")
.table(Alias::new("user_metadata_tag"))
.col(Alias::new("tag_id"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_user_metadata_tag_source")
.table(Alias::new("user_metadata_tag"))
.col(Alias::new("source"))
.to_owned(),
)
.await?;
// Create the tag_usage_pattern table
manager
.create_table(
Table::create()
.table(Alias::new("tag_usage_pattern"))
.if_not_exists()
.col(
ColumnDef::new(Alias::new("id"))
.integer()
.not_null()
.auto_increment()
.primary_key(),
)
.col(ColumnDef::new(Alias::new("tag_id")).integer().not_null())
.col(
ColumnDef::new(Alias::new("co_occurrence_tag_id"))
.integer()
.not_null(),
)
.col(
ColumnDef::new(Alias::new("occurrence_count"))
.integer()
.default(1),
)
.col(
ColumnDef::new(Alias::new("last_used_together"))
.timestamp_with_time_zone()
.not_null(),
)
.foreign_key(
&mut ForeignKey::create()
.name("fk_tag_usage_pattern_tag")
.from(Alias::new("tag_usage_pattern"), Alias::new("tag_id"))
.to(Alias::new("tag"), Alias::new("id"))
.on_delete(ForeignKeyAction::Cascade),
)
.foreign_key(
&mut ForeignKey::create()
.name("fk_tag_usage_pattern_co_occurrence")
.from(
Alias::new("tag_usage_pattern"),
Alias::new("co_occurrence_tag_id"),
)
.to(Alias::new("tag"), Alias::new("id"))
.on_delete(ForeignKeyAction::Cascade),
)
.to_owned(),
)
.await?;
// Create indexes for tag_usage_pattern
manager
.create_index(
Index::create()
.name("idx_tag_usage_pattern_tag")
.table(Alias::new("tag_usage_pattern"))
.col(Alias::new("tag_id"))
.to_owned(),
)
.await?;
manager
.create_index(
Index::create()
.name("idx_tag_usage_pattern_co_occurrence")
.table(Alias::new("tag_usage_pattern"))
.col(Alias::new("co_occurrence_tag_id"))
.to_owned(),
)
.await?;
// Create full-text search indexes
manager
.create_index(
Index::create()
.name("idx_tag_fulltext")
.table(Alias::new("tag"))
.col(Alias::new("canonical_name"))
.col(Alias::new("display_name"))
.col(Alias::new("formal_name"))
.col(Alias::new("abbreviation"))
.col(Alias::new("aliases"))
.col(Alias::new("description"))
.to_owned(),
)
.await?;
// TODO: FTS5 virtual table temporarily disabled for debugging
// Create FTS5 virtual table for full-text search
// manager
// .get_connection()
// .execute_unprepared(
// "CREATE VIRTUAL TABLE IF NOT EXISTS tag_search_fts USING fts5(
// tag_id UNINDEXED,
// canonical_name,
// display_name,
// formal_name,
// abbreviation,
// aliases,
// description,
// content='tag',
// content_rowid='id'
// )",
// )
// .await?;
// Create triggers to maintain FTS5 table
// manager
// .get_connection()
// .execute_unprepared(
// "CREATE TRIGGER IF NOT EXISTS tag_ai AFTER INSERT ON tag BEGIN
// INSERT INTO tag_search_fts(
// rowid, tag_id, canonical_name, display_name, formal_name,
// abbreviation, aliases, description
// ) VALUES (
// NEW.id, NEW.id, NEW.canonical_name, NEW.display_name, NEW.formal_name,
// NEW.abbreviation, NEW.aliases, NEW.description
// );
// END",
// )
// .await?;
// manager
// .get_connection()
// .execute_unprepared(
// "CREATE TRIGGER IF NOT EXISTS tag_au AFTER UPDATE ON tag BEGIN
// DELETE FROM tag_search_fts WHERE rowid = OLD.id;
// INSERT INTO tag_search_fts(rowid, tag_id, canonical_name, display_name, formal_name, abbreviation, aliases, description)
// VALUES (NEW.id, NEW.id, NEW.canonical_name, NEW.display_name, NEW.formal_name, NEW.abbreviation, NEW.aliases, NEW.description);
// END",
// )
// .await?;
// manager
// .get_connection()
// .execute_unprepared(
// "CREATE TRIGGER IF NOT EXISTS tag_ad AFTER DELETE ON tag BEGIN
// DELETE FROM tag_search_fts WHERE rowid = OLD.id;
// END",
// )
// .await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Drop FTS5 table and triggers first (temporarily disabled)
// manager
// .get_connection()
// .execute_unprepared("DROP TRIGGER IF EXISTS tag_ad")
// .await?;
// manager
// .get_connection()
// .execute_unprepared("DROP TRIGGER IF EXISTS tag_au")
// .await?;
// manager
// .get_connection()
// .execute_unprepared("DROP TRIGGER IF EXISTS tag_ai")
// .await?;
// manager
// .get_connection()
// .execute_unprepared("DROP TABLE IF EXISTS tag_search_fts")
// .await?;
// Drop tables in reverse order
manager
.drop_table(
Table::drop()
.table(Alias::new("tag_usage_pattern"))
.to_owned(),
)
.await?;
manager
.drop_table(
Table::drop()
.table(Alias::new("user_metadata_tag"))
.to_owned(),
)
.await?;
manager
.drop_table(Table::drop().table(Alias::new("tag_closure")).to_owned())
.await?;
manager
.drop_table(
Table::drop()
.table(Alias::new("tag_relationship"))
.to_owned(),
)
.await?;
manager
.drop_table(Table::drop().table(Alias::new("tag")).to_owned())
.await?;
Ok(())
}
}

160
core/src/infra/db/migration/m20250120_000001_create_fts5_search_index.rs Normal file
View File

@@ -0,0 +1,160 @@
//! FTS5 Search Index Migration
//!
//! Creates FTS5 virtual table for high-performance full-text search
//! and associated triggers for real-time index updates.
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Create FTS5 virtual table for search indexing
manager
.get_connection()
.execute_unprepared(
r#"
CREATE VIRTUAL TABLE search_index USING fts5(
content='entries',
content_rowid='id',
name,
extension,
tokenize="unicode61 remove_diacritics 2 tokenchars '.@-_'",
prefix='2,3'
);
"#,
)
.await?;
// Create trigger for INSERT operations
manager
.get_connection()
.execute_unprepared(
r#"
CREATE TRIGGER IF NOT EXISTS entries_search_insert
AFTER INSERT ON entries WHEN new.kind = 0
BEGIN
INSERT INTO search_index(rowid, name, extension)
VALUES (new.id, new.name, new.extension);
END;
"#,
)
.await?;
// Create trigger for UPDATE operations
manager
.get_connection()
.execute_unprepared(
r#"
CREATE TRIGGER IF NOT EXISTS entries_search_update
AFTER UPDATE ON entries WHEN new.kind = 0
BEGIN
UPDATE search_index SET
name = new.name,
extension = new.extension
WHERE rowid = new.id;
END;
"#,
)
.await?;
// Create trigger for DELETE operations
manager
.get_connection()
.execute_unprepared(
r#"
CREATE TRIGGER IF NOT EXISTS entries_search_delete
AFTER DELETE ON entries WHEN old.kind = 0
BEGIN
DELETE FROM search_index WHERE rowid = old.id;
END;
"#,
)
.await?;
// Populate FTS5 index with existing file entries
manager
.get_connection()
.execute_unprepared(
r#"
INSERT INTO search_index(rowid, name, extension)
SELECT id, name, extension FROM entries WHERE kind = 0;
"#,
)
.await?;
// Create search analytics table for query optimization
manager
.get_connection()
.execute_unprepared(
r#"
CREATE TABLE search_analytics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query_text TEXT NOT NULL,
query_hash TEXT NOT NULL,
search_mode TEXT NOT NULL,
execution_time_ms INTEGER NOT NULL,
result_count INTEGER NOT NULL,
fts5_used BOOLEAN DEFAULT TRUE,
semantic_used BOOLEAN DEFAULT FALSE,
user_clicked_result BOOLEAN DEFAULT FALSE,
clicked_result_position INTEGER,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
"#,
)
.await?;
// Create index on query_hash for performance analytics
manager
.get_connection()
.execute_unprepared(
r#"
CREATE INDEX idx_search_analytics_query_hash
ON search_analytics(query_hash);
"#,
)
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Drop analytics table and index
manager
.get_connection()
.execute_unprepared("DROP INDEX IF EXISTS idx_search_analytics_query_hash;")
.await?;
manager
.get_connection()
.execute_unprepared("DROP TABLE IF EXISTS search_analytics;")
.await?;
// Drop triggers
manager
.get_connection()
.execute_unprepared("DROP TRIGGER IF EXISTS entries_search_delete;")
.await?;
manager
.get_connection()
.execute_unprepared("DROP TRIGGER IF EXISTS entries_search_update;")
.await?;
manager
.get_connection()
.execute_unprepared("DROP TRIGGER IF EXISTS entries_search_insert;")
.await?;
// Drop FTS5 virtual table
manager
.get_connection()
.execute_unprepared("DROP TABLE IF EXISTS search_index;")
.await?;
Ok(())
}
}

111
core/src/infra/db/migration/m20251009_000001_add_sync_to_devices.rs Normal file
View File

@@ -0,0 +1,111 @@
//! Migration to add sync fields to devices table
//!
//! Extends the devices table with sync coordination fields.
//! This eliminates the need for a separate sync_partners table - if a device
//! is registered in a library, it's a sync partner.
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// Add sync_enabled column (defaults to true - all registered devices sync by default)
manager
.alter_table(
Table::alter()
.table(Devices::Table)
.add_column(
ColumnDef::new(Devices::SyncEnabled)
.boolean()
.not_null()
.default(true),
)
.to_owned(),
)
.await?;
// Add last_sync_at column to track last successful sync
manager
.alter_table(
Table::alter()
.table(Devices::Table)
.add_column(ColumnDef::new(Devices::LastSyncAt).timestamp_with_time_zone())
.to_owned(),
)
.await?;
// Add last_state_watermark column to track last device state watermark
manager
.alter_table(
Table::alter()
.table(Devices::Table)
.add_column(ColumnDef::new(Devices::LastStateWatermark).timestamp_with_time_zone())
.to_owned(),
)
.await?;
// Add last_shared_watermark column to track last shared resource watermark (HLC)
manager
.alter_table(
Table::alter()
.table(Devices::Table)
.add_column(ColumnDef::new(Devices::LastSharedWatermark).string())
.to_owned(),
)
.await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager
.alter_table(
Table::alter()
.table(Devices::Table)
.drop_column(Devices::SyncEnabled)
.to_owned(),
)
.await?;
manager
.alter_table(
Table::alter()
.table(Devices::Table)
.drop_column(Devices::LastSyncAt)
.to_owned(),
)
.await?;
manager
.alter_table(
Table::alter()
.table(Devices::Table)
.drop_column(Devices::LastStateWatermark)
.to_owned(),
)
.await?;
manager
.alter_table(
Table::alter()
.table(Devices::Table)
.drop_column(Devices::LastSharedWatermark)
.to_owned(),
)
.await?;
Ok(())
}
}
#[derive(DeriveIden)]
enum Devices {
Table,
SyncEnabled,
LastSyncAt,
LastStateWatermark,
LastSharedWatermark,
}

22
core/src/infra/db/migration/mod.rs
View File

@@ -2,13 +2,31 @@
use sea_orm_migration::prelude::*;
mod m20240101_000001_unified_schema;
mod m20240101_000001_initial_schema;
mod m20240102_000001_populate_lookups;
mod m20240107_000001_create_collections;
mod m20250109_000001_create_sidecars;
mod m20250110_000001_refactor_volumes_table;
mod m20250112_000001_create_indexer_rules;
mod m20250115_000001_semantic_tags;
mod m20250120_000001_create_fts5_search_index;
mod m20251009_000001_add_sync_to_devices;
pub struct Migrator;
#[async_trait::async_trait]
impl MigratorTrait for Migrator {
fn migrations() -> Vec<Box<dyn MigrationTrait>> {
vec![Box::new(m20240101_000001_unified_schema::Migration)]
vec![
Box::new(m20240101_000001_initial_schema::Migration),
Box::new(m20240102_000001_populate_lookups::Migration),
Box::new(m20240107_000001_create_collections::Migration),
Box::new(m20250109_000001_create_sidecars::Migration),
Box::new(m20250110_000001_refactor_volumes_table::Migration),
Box::new(m20250112_000001_create_indexer_rules::Migration),
Box::new(m20250115_000001_semantic_tags::Migration),
Box::new(m20250120_000001_create_fts5_search_index::Migration),
Box::new(m20251009_000001_add_sync_to_devices::Migration),
]
}
}

186
core/src/ops/volumes/add_cloud/action.rs
View File

@@ -32,6 +32,39 @@ pub enum CloudStorageConfig {
secret_access_key: String,
endpoint: Option<String>,
},
GoogleDrive {
root: Option<String>,
access_token: String,
refresh_token: String,
client_id: String,
client_secret: String,
},
OneDrive {
root: Option<String>,
access_token: String,
refresh_token: String,
client_id: String,
client_secret: String,
},
Dropbox {
root: Option<String>,
access_token: String,
refresh_token: String,
client_id: String,
client_secret: String,
},
AzureBlob {
container: String,
endpoint: Option<String>,
account_name: String,
account_key: String,
},
GoogleCloudStorage {
bucket: String,
root: Option<String>,
endpoint: Option<String>,
credential: String, // Service account JSON
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -93,6 +126,159 @@ impl LibraryAction for VolumeAddCloudAction {
let mount_point = PathBuf::from(format!("cloud://s3/{}", bucket));
(backend, credential, mount_point)
}
CloudStorageConfig::GoogleDrive {
root,
access_token,
refresh_token,
client_id,
client_secret,
} => {
let backend = CloudBackend::new_google_drive(
access_token,
refresh_token,
client_id,
client_secret,
root.clone(),
)
.await
.map_err(|e| {
ActionError::InvalidInput(format!("Failed to create Google Drive backend: {}", e))
})?;
let credential = CloudCredential::new_oauth(
CloudServiceType::GoogleDrive,
access_token.clone(),
refresh_token.clone(),
None, // Google Drive tokens typically don't have a fixed expiry in the refresh flow
);
let mount_point = PathBuf::from(format!(
"cloud://gdrive/{}",
root.as_deref().unwrap_or("root")
));
(backend, credential, mount_point)
}
CloudStorageConfig::OneDrive {
root,
access_token,
refresh_token,
client_id,
client_secret,
} => {
let backend = CloudBackend::new_onedrive(
access_token,
refresh_token,
client_id,
client_secret,
root.clone(),
)
.await
.map_err(|e| {
ActionError::InvalidInput(format!("Failed to create OneDrive backend: {}", e))
})?;
let credential = CloudCredential::new_oauth(
CloudServiceType::OneDrive,
access_token.clone(),
refresh_token.clone(),
None,
);
let mount_point = PathBuf::from(format!(
"cloud://onedrive/{}",
root.as_deref().unwrap_or("root")
));
(backend, credential, mount_point)
}
CloudStorageConfig::Dropbox {
root,
access_token,
refresh_token,
client_id,
client_secret,
} => {
let backend = CloudBackend::new_dropbox(
access_token,
refresh_token,
client_id,
client_secret,
root.clone(),
)
.await
.map_err(|e| {
ActionError::InvalidInput(format!("Failed to create Dropbox backend: {}", e))
})?;
let credential = CloudCredential::new_oauth(
CloudServiceType::Dropbox,
access_token.clone(),
refresh_token.clone(),
None,
);
let mount_point = PathBuf::from(format!(
"cloud://dropbox/{}",
root.as_deref().unwrap_or("root")
));
(backend, credential, mount_point)
}
CloudStorageConfig::AzureBlob {
container,
endpoint,
account_name,
account_key,
} => {
let backend = CloudBackend::new_azure_blob(
container,
account_name,
account_key,
endpoint.clone(),
)
.await
.map_err(|e| {
ActionError::InvalidInput(format!("Failed to create Azure Blob backend: {}", e))
})?;
let credential = CloudCredential::new_access_key(
CloudServiceType::AzureBlob,
account_name.clone(),
account_key.clone(),
None,
);
let mount_point = PathBuf::from(format!("cloud://azblob/{}", container));
(backend, credential, mount_point)
}
CloudStorageConfig::GoogleCloudStorage {
bucket,
root,
endpoint,
credential: service_account_json,
} => {
let backend = CloudBackend::new_google_cloud_storage(
bucket,
service_account_json,
root.clone(),
endpoint.clone(),
)
.await
.map_err(|e| {
ActionError::InvalidInput(format!("Failed to create GCS backend: {}", e))
})?;
let credential = CloudCredential::new_api_key(
CloudServiceType::GoogleCloudStorage,
service_account_json.clone(),
);
let mount_point = PathBuf::from(format!("cloud://gcs/{}", bucket));
(backend, credential, mount_point)
}
};

154
core/src/volume/backend/cloud.rs
View File

@@ -73,6 +73,160 @@ impl CloudBackend {
})
}
/// Create a new cloud backend for Google Drive
pub async fn new_google_drive(
access_token: impl AsRef<str>,
refresh_token: impl AsRef<str>,
client_id: impl AsRef<str>,
client_secret: impl AsRef<str>,
root: Option<String>,
) -> Result<Self, VolumeError> {
let mut builder = opendal::services::Gdrive::default()
.access_token(access_token.as_ref())
.refresh_token(refresh_token.as_ref())
.client_id(client_id.as_ref())
.client_secret(client_secret.as_ref());
if let Some(r) = &root {
builder = builder.root(r);
}
let operator = opendal::Operator::new(builder)
.map_err(|e| {
VolumeError::Platform(format!("Failed to create Google Drive operator: {}", e))
})?
.finish();
Ok(Self {
operator,
service_type: CloudServiceType::GoogleDrive,
root: PathBuf::from(root.unwrap_or_else(|| "/".to_string())),
})
}
/// Create a new cloud backend for OneDrive
pub async fn new_onedrive(
access_token: impl AsRef<str>,
refresh_token: impl AsRef<str>,
client_id: impl AsRef<str>,
client_secret: impl AsRef<str>,
root: Option<String>,
) -> Result<Self, VolumeError> {
let mut builder = opendal::services::Onedrive::default()
.access_token(access_token.as_ref())
.refresh_token(refresh_token.as_ref())
.client_id(client_id.as_ref())
.client_secret(client_secret.as_ref());
if let Some(r) = &root {
builder = builder.root(r);
}
let operator = opendal::Operator::new(builder)
.map_err(|e| {
VolumeError::Platform(format!("Failed to create OneDrive operator: {}", e))
})?
.finish();
Ok(Self {
operator,
service_type: CloudServiceType::OneDrive,
root: PathBuf::from(root.unwrap_or_else(|| "/".to_string())),
})
}
/// Create a new cloud backend for Dropbox
pub async fn new_dropbox(
access_token: impl AsRef<str>,
refresh_token: impl AsRef<str>,
client_id: impl AsRef<str>,
client_secret: impl AsRef<str>,
root: Option<String>,
) -> Result<Self, VolumeError> {
let mut builder = opendal::services::Dropbox::default()
.access_token(access_token.as_ref())
.refresh_token(refresh_token.as_ref())
.client_id(client_id.as_ref())
.client_secret(client_secret.as_ref());
if let Some(r) = &root {
builder = builder.root(r);
}
let operator = opendal::Operator::new(builder)
.map_err(|e| {
VolumeError::Platform(format!("Failed to create Dropbox operator: {}", e))
})?
.finish();
Ok(Self {
operator,
service_type: CloudServiceType::Dropbox,
root: PathBuf::from(root.unwrap_or_else(|| "/".to_string())),
})
}
/// Create a new cloud backend for Azure Blob Storage
pub async fn new_azure_blob(
container: impl AsRef<str>,
account_name: impl AsRef<str>,
account_key: impl AsRef<str>,
endpoint: Option<String>,
) -> Result<Self, VolumeError> {
let mut builder = opendal::services::Azblob::default()
.container(container.as_ref())
.account_name(account_name.as_ref())
.account_key(account_key.as_ref());
if let Some(ep) = endpoint {
builder = builder.endpoint(&ep);
}
let operator = opendal::Operator::new(builder)
.map_err(|e| {
VolumeError::Platform(format!("Failed to create Azure Blob operator: {}", e))
})?
.finish();
Ok(Self {
operator,
service_type: CloudServiceType::AzureBlob,
root: PathBuf::from("/"),
})
}
/// Create a new cloud backend for Google Cloud Storage
pub async fn new_google_cloud_storage(
bucket: impl AsRef<str>,
credential: impl AsRef<str>,
root: Option<String>,
endpoint: Option<String>,
) -> Result<Self, VolumeError> {
let mut builder = opendal::services::Gcs::default()
.bucket(bucket.as_ref())
.credential(credential.as_ref());
if let Some(r) = &root {
builder = builder.root(r);
}
if let Some(ep) = endpoint {
builder = builder.endpoint(&ep);
}
let operator = opendal::Operator::new(builder)
.map_err(|e| {
VolumeError::Platform(format!("Failed to create GCS operator: {}", e))
})?
.finish();
Ok(Self {
operator,
service_type: CloudServiceType::GoogleCloudStorage,
root: PathBuf::from(root.unwrap_or_else(|| "/".to_string())),
})
}
/// Create a cloud backend from a pre-configured OpenDAL operator
pub fn from_operator(operator: opendal::Operator, service_type: CloudServiceType) -> Self {
Self {

243
core/src/volume/manager.rs
View File

@@ -165,8 +165,17 @@ impl VolumeManager {
match credential_manager.get_credential(library.id(), &db_volume.fingerprint) {
Ok(credential) => {
// Get mount point and service name from database
let mount_point_str = match &db_volume.mount_point {
Some(mp) => mp,
None => {
warn!("No mount point for cloud volume {}", fingerprint.0);
continue;
}
};
// Recreate the cloud backend from stored credentials
match credential.service {
let backend_result = match credential.service {
crate::volume::CloudServiceType::S3 => {
if let crate::crypto::cloud_credentials::CredentialData::AccessKey {
access_key_id,
@@ -174,82 +183,183 @@ impl VolumeManager {
..
} = &credential.data
{
// Parse mount point to extract bucket info
let mount_point_str = db_volume.mount_point.as_ref()
.ok_or_else(|| VolumeError::InvalidData("No mount point for cloud volume".to_string()))?;
// Extract bucket name from mount_point like "cloud://s3/bucket-name"
let bucket = mount_point_str
.strip_prefix("cloud://s3/")
.ok_or_else(|| VolumeError::InvalidData(format!("Invalid S3 mount point: {}", mount_point_str)))?;
.unwrap_or("unknown");
// Try to recreate the backend (we don't have region/endpoint stored, use defaults)
match crate::volume::CloudBackend::new_s3(
crate::volume::CloudBackend::new_s3(
bucket,
"us-east-1", // Default region - TODO: store this in DB
"us-east-1", // Default region
access_key_id,
secret_access_key,
None,
).await {
Ok(backend) => {
let now = chrono::Utc::now();
// Reconstruct the Volume struct
let volume = Volume {
id: db_volume.uuid, // Use UUID from database
fingerprint: fingerprint.clone(),
device_id: db_volume.device_id,
name: db_volume.display_name.clone().unwrap_or_else(|| bucket.to_string()),
library_id: None,
is_tracked: true,
mount_point: std::path::PathBuf::from(mount_point_str),
mount_points: vec![std::path::PathBuf::from(mount_point_str)],
volume_type: crate::volume::types::VolumeType::Network,
mount_type: crate::volume::types::MountType::Network,
disk_type: crate::volume::types::DiskType::Unknown,
file_system: crate::volume::types::FileSystem::Other("S3".to_string()),
total_capacity: db_volume.total_capacity.unwrap_or(0) as u64,
available_space: db_volume.available_capacity.unwrap_or(0) as u64,
is_read_only: false,
is_mounted: true,
hardware_id: None,
backend: Some(Arc::new(backend)),
apfs_container: None,
container_volume_id: None,
path_mappings: Vec::new(),
is_user_visible: db_volume.is_user_visible.unwrap_or(true),
auto_track_eligible: db_volume.auto_track_eligible.unwrap_or(false),
read_speed_mbps: db_volume.read_speed_mbps.map(|s| s as u64),
write_speed_mbps: db_volume.write_speed_mbps.map(|s| s as u64),
created_at: db_volume.tracked_at,
updated_at: now,
last_seen_at: db_volume.last_seen_at,
total_files: None,
total_directories: None,
last_stats_update: None,
display_name: db_volume.display_name.clone(),
is_favorite: false,
color: None,
icon: None,
error_message: None,
};
).await
} else {
warn!("Invalid credential type for S3 volume {}", fingerprint.0);
continue;
}
}
crate::volume::CloudServiceType::GoogleDrive => {
if let crate::crypto::cloud_credentials::CredentialData::OAuth {
access_token,
refresh_token,
} = &credential.data
{
// Extract root from mount point if available
let root = mount_point_str
.strip_prefix("cloud://gdrive/")
.map(|s| s.to_string());
// Register in memory
let mut volumes = self.volumes.write().await;
volumes.insert(fingerprint.clone(), volume);
loaded_count += 1;
info!("Loaded cloud volume {} from database", db_volume.display_name.as_ref().unwrap_or(&bucket.to_string()));
}
Err(e) => {
warn!("Failed to recreate cloud backend for volume {}: {}", fingerprint.0, e);
}
}
crate::volume::CloudBackend::new_google_drive(
access_token,
refresh_token,
"", // client_id not stored yet
"", // client_secret not stored yet
root,
).await
} else {
warn!("Invalid credential type for Google Drive volume {}", fingerprint.0);
continue;
}
}
crate::volume::CloudServiceType::OneDrive => {
if let crate::crypto::cloud_credentials::CredentialData::OAuth {
access_token,
refresh_token,
} = &credential.data
{
let root = mount_point_str
.strip_prefix("cloud://onedrive/")
.map(|s| s.to_string());
crate::volume::CloudBackend::new_onedrive(
access_token,
refresh_token,
"",
"",
root,
).await
} else {
warn!("Invalid credential type for OneDrive volume {}", fingerprint.0);
continue;
}
}
crate::volume::CloudServiceType::Dropbox => {
if let crate::crypto::cloud_credentials::CredentialData::OAuth {
access_token,
refresh_token,
} = &credential.data
{
let root = mount_point_str
.strip_prefix("cloud://dropbox/")
.map(|s| s.to_string());
crate::volume::CloudBackend::new_dropbox(
access_token,
refresh_token,
"",
"",
root,
).await
} else {
warn!("Invalid credential type for Dropbox volume {}", fingerprint.0);
continue;
}
}
crate::volume::CloudServiceType::AzureBlob => {
if let crate::crypto::cloud_credentials::CredentialData::AccessKey {
access_key_id,
secret_access_key,
..
} = &credential.data
{
let container = mount_point_str
.strip_prefix("cloud://azblob/")
.unwrap_or("unknown");
crate::volume::CloudBackend::new_azure_blob(
container,
access_key_id,
secret_access_key,
None,
).await
} else {
warn!("Invalid credential type for Azure Blob volume {}", fingerprint.0);
continue;
}
}
crate::volume::CloudServiceType::GoogleCloudStorage => {
if let crate::crypto::cloud_credentials::CredentialData::ApiKey(service_account_json) = &credential.data {
let bucket = mount_point_str
.strip_prefix("cloud://gcs/")
.unwrap_or("unknown");
crate::volume::CloudBackend::new_google_cloud_storage(
bucket,
service_account_json,
None,
None,
).await
} else {
warn!("Invalid credential type for GCS volume {}", fingerprint.0);
continue;
}
}
_ => {
warn!(
"Unsupported cloud service type for volume {}",
fingerprint.0
);
warn!("Unsupported cloud service type {:?} for volume {}", credential.service, fingerprint.0);
continue;
}
};
match backend_result {
Ok(backend) => {
let now = chrono::Utc::now();
let volume = Volume {
id: db_volume.uuid,
fingerprint: fingerprint.clone(),
device_id: db_volume.device_id,
name: db_volume.display_name.clone().unwrap_or_else(|| "Cloud Volume".to_string()),
library_id: None,
is_tracked: true,
mount_point: std::path::PathBuf::from(mount_point_str),
mount_points: vec![std::path::PathBuf::from(mount_point_str)],
volume_type: crate::volume::types::VolumeType::Network,
mount_type: crate::volume::types::MountType::Network,
disk_type: crate::volume::types::DiskType::Unknown,
file_system: crate::volume::types::FileSystem::Other(format!("{:?}", credential.service)),
total_capacity: db_volume.total_capacity.unwrap_or(0) as u64,
available_space: db_volume.available_capacity.unwrap_or(0) as u64,
is_read_only: false,
is_mounted: true,
hardware_id: None,
backend: Some(Arc::new(backend)),
apfs_container: None,
container_volume_id: None,
path_mappings: Vec::new(),
is_user_visible: db_volume.is_user_visible.unwrap_or(true),
auto_track_eligible: db_volume.auto_track_eligible.unwrap_or(false),
read_speed_mbps: db_volume.read_speed_mbps.map(|s| s as u64),
write_speed_mbps: db_volume.write_speed_mbps.map(|s| s as u64),
created_at: db_volume.tracked_at,
updated_at: now,
last_seen_at: db_volume.last_seen_at,
total_files: None,
total_directories: None,
last_stats_update: None,
display_name: db_volume.display_name.clone(),
is_favorite: false,
color: None,
icon: None,
error_message: None,
};
let mut volumes = self.volumes.write().await;
volumes.insert(fingerprint.clone(), volume);
loaded_count += 1;
info!("Loaded cloud volume {} ({:?}) from database", db_volume.display_name.as_ref().unwrap_or(&"Unknown".to_string()), credential.service);
}
Err(e) => {
warn!("Failed to recreate cloud backend for volume {}: {}", fingerprint.0, e);
}
}
}
@@ -1295,6 +1405,7 @@ impl VolumeManager {
device_name: None, // TODO: Get from DeviceManager when available
volume_name: volume.name.clone(),
device_id: volume.device_id,
library_id: Uuid::nil(), // TODO: Populate from library context when available
};
if let Ok(content) = serde_json::to_string_pretty(&spacedrive_id) {

718
core/tests/sync_integration_test.rs
View File

@@ -412,7 +412,7 @@ impl MockTransportPeer {
/// Send response to the peer that made the request
async fn send_response_to_pending(
&self,
original_request: sd_core::service::network::protocol::sync::messages::SyncMessage,
_original_request: sd_core::service::network::protocol::sync::messages::SyncMessage,
response: sd_core::service::network::protocol::sync::messages::SyncMessage,
) -> anyhow::Result<()> {
// Send response through outgoing queue
@@ -605,6 +605,8 @@ impl SyncTestSetup {
updated_at: Set(Utc::now()),
sync_enabled: Set(true), // Enable sync by default
last_sync_at: Set(None),
last_state_watermark: Set(None),
last_shared_watermark: Set(None),
};
device_model.insert(library.db().conn()).await?;
@@ -1377,7 +1379,7 @@ async fn test_sync_transitive_three_devices() -> anyhow::Result<()> {
info!("️ Device C ID: {}", device_c_id);
// Create libraries (all same library ID for shared library scenario)
let library_id = Uuid::new_v4();
let _library_id = Uuid::new_v4();
// Create library on A
let library_a = core_a
@@ -1532,7 +1534,7 @@ async fn test_sync_transitive_three_devices() -> anyhow::Result<()> {
assert!(
tag_on_c.is_some(),
"Device C should have A's tag (received via B while A was offline) ✅"
"Device C should have A's tag (received via B while A was offline) "
);
let synced_tag = tag_on_c.unwrap();
@@ -1551,3 +1553,713 @@ async fn test_sync_transitive_three_devices() -> anyhow::Result<()> {
info!("TEST COMPLETE: Transitive sync validated (A → B → C)");
Ok(())
}
#[tokio::test]
async fn test_connection_state_tracking() -> anyhow::Result<()> {
info!("TEST: Connection State Tracking");
let setup = SyncTestSetup::new().await?;
// === VERIFY INITIAL STATE ===
// Both devices should be offline initially
let device_b_on_a = entities::device::Entity::find()
.filter(entities::device::Column::Uuid.eq(setup.device_b_id))
.one(setup.library_a.db().conn())
.await?
.expect("Device B should exist on A");
assert_eq!(device_b_on_a.is_online, false, "Device B should start offline");
info!("Initial state: Device B is offline on A's library");
let device_a_on_b = entities::device::Entity::find()
.filter(entities::device::Column::Uuid.eq(setup.device_a_id))
.one(setup.library_b.db().conn())
.await?
.expect("Device A should exist on B");
assert_eq!(device_a_on_b.is_online, false, "Device A should start offline");
info!("Initial state: Device A is offline on B's library");
// === SIMULATE CONNECTION ESTABLISHED ===
info!("Simulating ConnectionEstablished events");
// Device A's PeerSync receives ConnectionEstablished for Device B
if let Some(_sync_a) = setup.library_a.sync_service() {
// Simulate the connection event by directly updating the database
// (in production, this would be handled by NetworkEvent listener)
use chrono::Utc;
let now = Utc::now();
entities::device::Entity::update_many()
.col_expr(
entities::device::Column::IsOnline,
sea_orm::sea_query::Expr::value(true),
)
.col_expr(
entities::device::Column::LastSeenAt,
sea_orm::sea_query::Expr::value(now),
)
.col_expr(
entities::device::Column::UpdatedAt,
sea_orm::sea_query::Expr::value(now),
)
.filter(entities::device::Column::Uuid.eq(setup.device_b_id))
.exec(setup.library_a.db().conn())
.await?;
}
// Device B's PeerSync receives ConnectionEstablished for Device A
if let Some(_sync_b) = setup.library_b.sync_service() {
use chrono::Utc;
let now = Utc::now();
entities::device::Entity::update_many()
.col_expr(
entities::device::Column::IsOnline,
sea_orm::sea_query::Expr::value(true),
)
.col_expr(
entities::device::Column::LastSeenAt,
sea_orm::sea_query::Expr::value(now),
)
.col_expr(
entities::device::Column::UpdatedAt,
sea_orm::sea_query::Expr::value(now),
)
.filter(entities::device::Column::Uuid.eq(setup.device_a_id))
.exec(setup.library_b.db().conn())
.await?;
}
info!("Connection events processed");
// === VERIFY ONLINE STATE ===
let device_b_on_a = entities::device::Entity::find()
.filter(entities::device::Column::Uuid.eq(setup.device_b_id))
.one(setup.library_a.db().conn())
.await?
.expect("Device B should exist");
assert_eq!(
device_b_on_a.is_online, true,
"Device B should be online after ConnectionEstablished"
);
info!("Device B is now ONLINE on A's library");
let device_a_on_b = entities::device::Entity::find()
.filter(entities::device::Column::Uuid.eq(setup.device_a_id))
.one(setup.library_b.db().conn())
.await?
.expect("Device A should exist");
assert_eq!(
device_a_on_b.is_online, true,
"Device A should be online after ConnectionEstablished"
);
info!("Device A is now ONLINE on B's library");
// === SIMULATE CONNECTION LOST ===
info!("Simulating ConnectionLost events");
// Device A's PeerSync receives ConnectionLost for Device B
if let Some(_sync_a) = setup.library_a.sync_service() {
use chrono::Utc;
let now = Utc::now();
entities::device::Entity::update_many()
.col_expr(
entities::device::Column::IsOnline,
sea_orm::sea_query::Expr::value(false),
)
.col_expr(
entities::device::Column::LastSeenAt,
sea_orm::sea_query::Expr::value(now),
)
.col_expr(
entities::device::Column::UpdatedAt,
sea_orm::sea_query::Expr::value(now),
)
.filter(entities::device::Column::Uuid.eq(setup.device_b_id))
.exec(setup.library_a.db().conn())
.await?;
}
// Device B's PeerSync receives ConnectionLost for Device A
if let Some(_sync_b) = setup.library_b.sync_service() {
use chrono::Utc;
let now = Utc::now();
entities::device::Entity::update_many()
.col_expr(
entities::device::Column::IsOnline,
sea_orm::sea_query::Expr::value(false),
)
.col_expr(
entities::device::Column::LastSeenAt,
sea_orm::sea_query::Expr::value(now),
)
.col_expr(
entities::device::Column::UpdatedAt,
sea_orm::sea_query::Expr::value(now),
)
.filter(entities::device::Column::Uuid.eq(setup.device_a_id))
.exec(setup.library_b.db().conn())
.await?;
}
info!("Disconnection events processed");
// === VERIFY OFFLINE STATE ===
let device_b_on_a = entities::device::Entity::find()
.filter(entities::device::Column::Uuid.eq(setup.device_b_id))
.one(setup.library_a.db().conn())
.await?
.expect("Device B should exist");
assert_eq!(
device_b_on_a.is_online, false,
"Device B should be offline after ConnectionLost"
);
info!("Device B is now OFFLINE on A's library");
let device_a_on_b = entities::device::Entity::find()
.filter(entities::device::Column::Uuid.eq(setup.device_a_id))
.one(setup.library_b.db().conn())
.await?
.expect("Device A should exist");
assert_eq!(
device_a_on_b.is_online, false,
"Device A should be offline after ConnectionLost"
);
info!("Device A is now OFFLINE on B's library");
info!("TEST COMPLETE: Connection state tracking validated");
info!(" - ConnectionEstablished updates is_online=true and last_seen_at");
info!(" - ConnectionLost updates is_online=false and last_seen_at");
Ok(())
}
#[tokio::test]
async fn test_watermark_reconnection_sync() -> anyhow::Result<()> {
info!("TEST: Watermark-Based Reconnection Sync");
let setup = SyncTestSetup::new().await?;
// === PHASE 1: Initial sync with tags ===
info!("PHASE 1: Creating initial tags and syncing");
let mut initial_tag_uuids = Vec::new();
for i in 0..3 {
let tag_uuid = Uuid::new_v4();
let tag_model = entities::tag::ActiveModel {
id: sea_orm::ActiveValue::NotSet,
uuid: Set(tag_uuid),
canonical_name: Set(format!("Initial Tag {}", i + 1)),
display_name: Set(None),
formal_name: Set(None),
abbreviation: Set(None),
aliases: Set(None),
namespace: Set(Some("photos".to_string())),
tag_type: Set("standard".to_string()),
color: Set(None),
icon: Set(None),
description: Set(None),
is_organizational_anchor: Set(false),
privacy_level: Set("normal".to_string()),
search_weight: Set(100),
attributes: Set(None),
composition_rules: Set(None),
created_at: Set(chrono::Utc::now()),
updated_at: Set(chrono::Utc::now()),
created_by_device: Set(Some(setup.device_a_id)),
};
let tag_record = tag_model.insert(setup.library_a.db().conn()).await?;
setup
.library_a
.sync_model(&tag_record, ChangeType::Insert)
.await?;
initial_tag_uuids.push(tag_uuid);
info!("Created initial tag {}: {}", i + 1, tag_uuid);
}
// Pump messages to sync to Device B
setup.wait_for_sync(Duration::from_secs(2)).await?;
// Verify Device B received all initial tags
let tags_on_b = entities::tag::Entity::find()
.all(setup.library_b.db().conn())
.await?;
assert_eq!(
tags_on_b.len(),
3,
"Device B should have all 3 initial tags"
);
info!("Device B received all initial tags");
// === RECORD WATERMARK (simulating what Device B would track) ===
// Get the last HLC from SharedChange messages (tags use shared resources, not peer_log)
let messages_a_to_b = setup.transport.get_a_to_b_messages().await;
let last_hlc = messages_a_to_b
.iter()
.filter_map(|(_, msg)| {
if let sd_core::service::network::protocol::sync::messages::SyncMessage::SharedChange {
entry,
..
} = msg
{
Some(entry.hlc.clone())
} else {
None
}
})
.last()
.expect("Should have SharedChange messages with HLC");
info!("Device B's watermark: {:?}", last_hlc);
// === PHASE 2: Simulate disconnection and create more tags ===
info!("\nPHASE 2: Device B disconnects, Device A creates more tags");
tokio::time::sleep(Duration::from_millis(100)).await;
let mut new_tag_uuids = Vec::new();
for i in 0..2 {
let tag_uuid = Uuid::new_v4();
let tag_model = entities::tag::ActiveModel {
id: sea_orm::ActiveValue::NotSet,
uuid: Set(tag_uuid),
canonical_name: Set(format!("New Tag {}", i + 1)),
display_name: Set(None),
formal_name: Set(None),
abbreviation: Set(None),
aliases: Set(None),
namespace: Set(Some("photos".to_string())),
tag_type: Set("standard".to_string()),
color: Set(None),
icon: Set(None),
description: Set(None),
is_organizational_anchor: Set(false),
privacy_level: Set("normal".to_string()),
search_weight: Set(100),
attributes: Set(None),
composition_rules: Set(None),
created_at: Set(chrono::Utc::now()),
updated_at: Set(chrono::Utc::now()),
created_by_device: Set(Some(setup.device_a_id)),
};
let tag_record = tag_model.insert(setup.library_a.db().conn()).await?;
setup
.library_a
.sync_model(&tag_record, ChangeType::Insert)
.await?;
new_tag_uuids.push(tag_uuid);
info!("Created new tag {}: {} (while B offline)", i + 1, tag_uuid);
}
// === PHASE 3: Reconnection with incremental sync ===
info!("\nPHASE 3: Device B reconnects and requests only new changes");
// Device B requests changes since last watermark (not full backfill)
use sd_core::service::network::protocol::sync::messages::SyncMessage;
let request = SyncMessage::SharedChangeRequest {
library_id: setup.library_b.id(),
since_hlc: Some(last_hlc), // Request only changes AFTER this HLC
limit: 1000,
};
info!("Device B sending SharedChangeRequest with watermark");
// Send request and wait for response
tokio::spawn({
let transport_b = setup.transport_b.clone();
let device_a_id = setup.device_a_id;
async move {
transport_b
.send_sync_message(device_a_id, request)
.await
.unwrap();
}
});
// Pump messages
setup.wait_for_sync(Duration::from_secs(2)).await?;
// === VALIDATION ===
info!("\nValidating incremental sync results");
// Check messages sent from A to B
let messages_a_to_b = setup.transport.get_a_to_b_messages().await;
info!("Total messages A→B: {}", messages_a_to_b.len());
// Filter for SharedChangeResponse messages
let shared_change_responses: Vec<_> = messages_a_to_b
.iter()
.filter_map(|(_, msg)| {
if let SyncMessage::SharedChangeResponse {
entries,
current_state,
..
} = msg
{
Some((entries, current_state))
} else {
None
}
})
.collect();
assert!(
!shared_change_responses.is_empty(),
"Should have received SharedChangeResponse"
);
// Verify that current_state is NOT included (this is incremental, not backfill)
let has_full_state = shared_change_responses
.iter()
.any(|(_, state)| state.is_some());
assert!(
!has_full_state,
"Incremental sync should NOT include full state snapshot"
);
// Count entries in response
let total_entries: usize = shared_change_responses
.iter()
.map(|(entries, _)| entries.len())
.sum();
info!("Incremental changes received: {} entries", total_entries);
assert_eq!(
total_entries, 2,
"Should only receive 2 new tags (not all 5)"
);
// Verify Device B now has all 5 tags
let all_tags_on_b = entities::tag::Entity::find()
.all(setup.library_b.db().conn())
.await?;
assert_eq!(
all_tags_on_b.len(),
5,
"Device B should have all 5 tags after incremental sync"
);
// Verify the new tags are present
for new_tag_uuid in &new_tag_uuids {
let tag_exists = all_tags_on_b
.iter()
.any(|t| t.uuid == *new_tag_uuid);
assert!(
tag_exists,
"New tag {} should exist on Device B",
new_tag_uuid
);
}
info!("TEST COMPLETE: Watermark-based incremental sync validated");
info!(" - Device B tracked watermark after initial sync");
info!(" - Device A created 2 new tags while B offline");
info!(" - Device B requested only changes since watermark (not full backfill)");
info!(" - Device A sent only 2 new entries (not all 5)");
info!(" - Device B successfully applied incremental changes");
Ok(())
}
#[tokio::test]
async fn test_concurrent_tag_updates_hlc_conflict_resolution() -> anyhow::Result<()> {
info!("TEST: HLC-Based Conflict Resolution");
let setup = SyncTestSetup::new().await?;
// === SETUP: Create same tag on both devices with different HLCs ===
info!("Creating same tag UUID on both devices with concurrent edits");
let tag_uuid = Uuid::new_v4();
// === Device A creates tag with canonical_name "Version A" ===
let tag_model_a = entities::tag::ActiveModel {
id: sea_orm::ActiveValue::NotSet,
uuid: Set(tag_uuid),
canonical_name: Set("Version A".to_string()),
display_name: Set(None),
formal_name: Set(None),
abbreviation: Set(None),
aliases: Set(None),
namespace: Set(Some("photos".to_string())),
tag_type: Set("standard".to_string()),
color: Set(None),
icon: Set(None),
description: Set(None),
is_organizational_anchor: Set(false),
privacy_level: Set("normal".to_string()),
search_weight: Set(100),
attributes: Set(None),
composition_rules: Set(None),
created_at: Set(chrono::Utc::now()),
updated_at: Set(chrono::Utc::now()),
created_by_device: Set(Some(setup.device_a_id)),
};
let tag_record_a = tag_model_a.insert(setup.library_a.db().conn()).await?;
info!("Device A created tag: {} = '{}'", tag_uuid, "Version A");
// === Device B creates same tag with canonical_name "Version B" ===
// (simulating concurrent offline edits)
let tag_model_b = entities::tag::ActiveModel {
id: sea_orm::ActiveValue::NotSet,
uuid: Set(tag_uuid),
canonical_name: Set("Version B".to_string()),
display_name: Set(None),
formal_name: Set(None),
abbreviation: Set(None),
aliases: Set(None),
namespace: Set(Some("photos".to_string())),
tag_type: Set("standard".to_string()),
color: Set(None),
icon: Set(None),
description: Set(None),
is_organizational_anchor: Set(false),
privacy_level: Set("normal".to_string()),
search_weight: Set(100),
attributes: Set(None),
composition_rules: Set(None),
created_at: Set(chrono::Utc::now()),
updated_at: Set(chrono::Utc::now()),
created_by_device: Set(Some(setup.device_b_id)),
};
let tag_record_b = tag_model_b.insert(setup.library_b.db().conn()).await?;
info!("Device B created tag: {} = '{}'", tag_uuid, "Version B");
// === Both devices sync their versions (conflict!) ===
info!("Both devices sync their versions simultaneously");
// Add artificial delay to ensure HLCs are different
tokio::time::sleep(Duration::from_millis(50)).await;
// Device A syncs first (earlier HLC)
setup
.library_a
.sync_model(&tag_record_a, ChangeType::Insert)
.await?;
tokio::time::sleep(Duration::from_millis(100)).await;
// Device B syncs second (later HLC - should win)
setup
.library_b
.sync_model(&tag_record_b, ChangeType::Insert)
.await?;
info!("Both devices broadcasted their versions");
// === Pump messages in both directions ===
setup.wait_for_sync(Duration::from_secs(2)).await?;
// === VALIDATION: Higher HLC should win ===
info!("\nValidating conflict resolution");
// Check Device A's version (should have B's version, since B's HLC is later)
let tag_on_a = entities::tag::Entity::find()
.filter(entities::tag::Column::Uuid.eq(tag_uuid))
.one(setup.library_a.db().conn())
.await?
.expect("Tag should exist on A");
// Check Device B's version (should keep B's version)
let tag_on_b = entities::tag::Entity::find()
.filter(entities::tag::Column::Uuid.eq(tag_uuid))
.one(setup.library_b.db().conn())
.await?
.expect("Tag should exist on B");
info!("Tag on Device A: canonical_name = '{}'", tag_on_a.canonical_name);
info!("Tag on Device B: canonical_name = '{}'", tag_on_b.canonical_name);
// Both should converge to the same value (last write wins based on HLC)
assert_eq!(
tag_on_a.canonical_name, tag_on_b.canonical_name,
"Both devices should converge to same version"
);
// The winner should be "Version B" since it had the later HLC
assert_eq!(
tag_on_b.canonical_name, "Version B",
"Version B should win (later HLC)"
);
info!("TEST COMPLETE: HLC conflict resolution validated");
info!(" - Both devices created same tag UUID with different values");
info!(" - Device A synced first (earlier HLC) = 'Version A'");
info!(" - Device B synced second (later HLC) = 'Version B'");
info!(" - After bidirectional sync, both converged to 'Version B' (higher HLC wins)");
Ok(())
}
#[tokio::test]
async fn test_sync_update_and_delete_operations() -> anyhow::Result<()> {
info!("TEST: Update and Delete Operations");
let setup = SyncTestSetup::new().await?;
// === PHASE 1: Create tag on Device A ===
info!("PHASE 1: Creating tag on Device A");
let tag_uuid = Uuid::new_v4();
let tag_model = entities::tag::ActiveModel {
id: sea_orm::ActiveValue::NotSet,
uuid: Set(tag_uuid),
canonical_name: Set("Original Name".to_string()),
display_name: Set(None),
formal_name: Set(None),
abbreviation: Set(None),
aliases: Set(None),
namespace: Set(Some("photos".to_string())),
tag_type: Set("standard".to_string()),
color: Set(None),
icon: Set(None),
description: Set(Some("Original description".to_string())),
is_organizational_anchor: Set(false),
privacy_level: Set("normal".to_string()),
search_weight: Set(100),
attributes: Set(None),
composition_rules: Set(None),
created_at: Set(chrono::Utc::now()),
updated_at: Set(chrono::Utc::now()),
created_by_device: Set(Some(setup.device_a_id)),
};
let tag_record = tag_model.insert(setup.library_a.db().conn()).await?;
info!("Created tag: {} = '{}'", tag_uuid, "Original Name");
// Sync to Device B
setup
.library_a
.sync_model(&tag_record, ChangeType::Insert)
.await?;
setup.wait_for_sync(Duration::from_secs(2)).await?;
// Verify Device B received the tag
let tag_on_b = entities::tag::Entity::find()
.filter(entities::tag::Column::Uuid.eq(tag_uuid))
.one(setup.library_b.db().conn())
.await?;
assert!(tag_on_b.is_some(), "Tag should exist on Device B");
let initial_tag_on_b = tag_on_b.unwrap();
assert_eq!(initial_tag_on_b.canonical_name, "Original Name");
assert_eq!(
initial_tag_on_b.description,
Some("Original description".to_string())
);
info!("Device B received initial tag");
// === PHASE 2: Update tag on Device A ===
info!("\nPHASE 2: Updating tag on Device A");
// Update the tag
let mut tag_active_model: entities::tag::ActiveModel = tag_record.into();
tag_active_model.canonical_name = Set("Updated Name".to_string());
tag_active_model.description = Set(Some("Updated description".to_string()));
tag_active_model.updated_at = Set(chrono::Utc::now());
let updated_tag = tag_active_model
.update(setup.library_a.db().conn())
.await?;
info!("Updated tag: canonical_name = '{}'", updated_tag.canonical_name);
// Sync the update
setup
.library_a
.sync_model(&updated_tag, ChangeType::Update)
.await?;
setup.wait_for_sync(Duration::from_secs(2)).await?;
// Verify Device B received the update
let tag_on_b = entities::tag::Entity::find()
.filter(entities::tag::Column::Uuid.eq(tag_uuid))
.one(setup.library_b.db().conn())
.await?;
assert!(tag_on_b.is_some(), "Tag should still exist on Device B");
let updated_tag_on_b = tag_on_b.unwrap();
assert_eq!(updated_tag_on_b.canonical_name, "Updated Name", "Device B should have updated name");
assert_eq!(
updated_tag_on_b.description,
Some("Updated description".to_string()),
"Device B should have updated description"
);
info!("Device B received update");
// === PHASE 3: Delete tag on Device A ===
info!("\nPHASE 3: Deleting tag on Device A");
// Delete the tag
let delete_result = entities::tag::Entity::delete_by_id(updated_tag.id)
.exec(setup.library_a.db().conn())
.await?;
assert_eq!(delete_result.rows_affected, 1, "Should delete 1 row");
info!("Deleted tag on Device A");
// Sync the delete
setup
.library_a
.sync_model(&updated_tag, ChangeType::Delete)
.await?;
setup.wait_for_sync(Duration::from_secs(2)).await?;
// Verify Device B received the delete
let tag_on_b = entities::tag::Entity::find()
.filter(entities::tag::Column::Uuid.eq(tag_uuid))
.one(setup.library_b.db().conn())
.await?;
assert!(
tag_on_b.is_none(),
"Tag should be deleted on Device B"
);
info!("Device B received delete");
// === VALIDATION: Check messages ===
info!("\nValidating message types");
let messages_a_to_b = setup.transport.get_a_to_b_messages().await;
info!("Total messages A→B: {}", messages_a_to_b.len());
// Count SharedChange messages by type
let mut insert_count = 0;
let mut update_count = 0;
let mut delete_count = 0;
for (_, msg) in &messages_a_to_b {
if let sd_core::service::network::protocol::sync::messages::SyncMessage::SharedChange {
entry,
..
} = msg
{
match entry.change_type {
ChangeType::Insert => insert_count += 1,
ChangeType::Update => update_count += 1,
ChangeType::Delete => delete_count += 1,
}
}
}
info!("Message counts:");
info!(" Insert: {}", insert_count);
info!(" Update: {}", update_count);
info!(" Delete: {}", delete_count);
assert!(insert_count > 0, "Should have Insert message");
assert!(update_count > 0, "Should have Update message");
assert!(delete_count > 0, "Should have Delete message");
info!("TEST COMPLETE: CRUD operations validated");
info!(" - INSERT: Tag created on A, synced to B");
info!(" - UPDATE: Tag updated on A (name + description), changes synced to B");
info!(" - DELETE: Tag deleted on A, deletion synced to B");
info!(" - All operations successfully propagated across devices");
Ok(())
}