From b0235e05c87541ab9e5d5fdd75a2befd32b5b3e9 Mon Sep 17 00:00:00 2001 From: Dionisio Date: Tue, 24 Feb 2026 10:09:49 +0100 Subject: [PATCH] perf(issue#6): migrate ShareFsRepository to PostgreSQL - Add storage.shares table with indexes on token, (item_id, item_type), created_by - Create SharePgRepository with indexed SQL queries and window-function pagination - Rewire DI to inject SharePgRepository with PgPool instead of config - Delete legacy share_fs_repository.rs (295 lines of JSON file I/O) - Remove dead module declaration from repositories/mod.rs Eliminates O(n) full-file JSON reads/writes, TOCTOU races, and crash corruption risk. All share operations now use indexed PG queries. --- db/schema.sql | 24 ++ src/common/di.rs | 7 +- src/infrastructure/repositories/mod.rs | 2 - src/infrastructure/repositories/pg/mod.rs | 2 + .../repositories/pg/share_pg_repository.rs | 308 ++++++++++++++++++ .../repositories/share_fs_repository.rs | 285 ---------------- 6 files changed, 338 insertions(+), 290 deletions(-) create mode 100644 src/infrastructure/repositories/pg/share_pg_repository.rs delete mode 100644 src/infrastructure/repositories/share_fs_repository.rs diff --git a/db/schema.sql b/db/schema.sql index 4d48b4fc..9d2883bb 100644 --- a/db/schema.sql +++ b/db/schema.sql @@ -498,3 +498,27 @@ COMMENT ON FUNCTION storage.decrement_blob_ref() IS 'Auto-decrement blob ref_cou COMMENT ON TABLE storage.folders IS 'Virtual folder hierarchy with ltree — no physical directories on disk'; COMMENT ON TABLE storage.files IS 'File metadata pointing to content-addressable blobs'; COMMENT ON VIEW storage.trash_items IS 'Unified view of all trashed files and folders'; + +-- ── Share links ────────────────────────────────────────────────────────── +-- Replaces the legacy file-based shares.json with proper relational storage. +CREATE TABLE IF NOT EXISTS storage.shares ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + item_id TEXT NOT NULL, + item_name TEXT, + item_type TEXT NOT NULL CHECK (item_type IN ('file', 'folder')), + token VARCHAR(36) NOT NULL UNIQUE, + password_hash TEXT, + expires_at BIGINT, -- unix epoch seconds, NULL = never expires + permissions_read BOOLEAN NOT NULL DEFAULT TRUE, + permissions_write BOOLEAN NOT NULL DEFAULT FALSE, + permissions_reshare BOOLEAN NOT NULL DEFAULT FALSE, + created_at BIGINT NOT NULL, -- unix epoch seconds + created_by VARCHAR(36) NOT NULL, + access_count BIGINT NOT NULL DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS idx_shares_token ON storage.shares(token); +CREATE INDEX IF NOT EXISTS idx_shares_item ON storage.shares(item_id, item_type); +CREATE INDEX IF NOT EXISTS idx_shares_created_by ON storage.shares(created_by); + +COMMENT ON TABLE storage.shares IS 'Shared links for files and folders with token-based access'; diff --git a/src/common/di.rs b/src/common/di.rs index 9fb1f046..f227b384 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -37,7 +37,7 @@ use crate::domain::services::i18n_service::I18nService; use crate::infrastructure::repositories::pg::{ FileBlobReadRepository, FileBlobWriteRepository, FolderDbRepository, TrashDbRepository, }; -use crate::infrastructure::repositories::share_fs_repository::ShareFsRepository; +use crate::infrastructure::repositories::pg::SharePgRepository; use crate::infrastructure::services::file_content_cache::{ FileContentCache, FileContentCacheConfig, }; @@ -320,13 +320,14 @@ impl AppServiceFactory { pub fn create_share_service( &self, repos: &RepositoryServices, + db_pool: &Arc, ) -> Option> { if !self.config.features.enable_file_sharing { tracing::info!("File sharing service is disabled in configuration"); return None; } - let share_repository = Arc::new(ShareFsRepository::new(Arc::new(self.config.clone()))); + let share_repository = Arc::new(SharePgRepository::new(db_pool.clone())); // Build a password hasher for share password verification let password_hasher: Arc = @@ -435,7 +436,7 @@ impl AppServiceFactory { let mut apps = self.create_application_services(&core, &repos, trash_service.clone()); // 5. Share service - let share_service = self.create_share_service(&repos); + let share_service = self.create_share_service(&repos, &pool); apps.share_service = share_service.clone(); // 6. Database-dependent services (PgPool always available in blob model) diff --git a/src/infrastructure/repositories/mod.rs b/src/infrastructure/repositories/mod.rs index 94597b67..d79dc0e0 100644 --- a/src/infrastructure/repositories/mod.rs +++ b/src/infrastructure/repositories/mod.rs @@ -1,5 +1,3 @@ -pub mod share_fs_repository; - // Repositorios PostgreSQL (blob-storage model) pub mod pg; diff --git a/src/infrastructure/repositories/pg/mod.rs b/src/infrastructure/repositories/pg/mod.rs index ce95afc9..408749d2 100644 --- a/src/infrastructure/repositories/pg/mod.rs +++ b/src/infrastructure/repositories/pg/mod.rs @@ -8,6 +8,7 @@ mod favorites_pg_repository; mod recent_items_pg_repository; mod session_pg_repository; mod settings_pg_repository; +mod share_pg_repository; mod transaction_utils; mod user_pg_repository; @@ -30,5 +31,6 @@ pub use folder_db_repository::FolderDbRepository; pub use recent_items_pg_repository::RecentItemsPgRepository; pub use session_pg_repository::SessionPgRepository; pub use settings_pg_repository::SettingsPgRepository; +pub use share_pg_repository::SharePgRepository; pub use trash_db_repository::TrashDbRepository; pub use user_pg_repository::UserPgRepository; diff --git a/src/infrastructure/repositories/pg/share_pg_repository.rs b/src/infrastructure/repositories/pg/share_pg_repository.rs new file mode 100644 index 00000000..d5790def --- /dev/null +++ b/src/infrastructure/repositories/pg/share_pg_repository.rs @@ -0,0 +1,308 @@ +use async_trait::async_trait; +use sqlx::{PgPool, Row}; +use std::sync::Arc; + +use crate::{ + application::ports::share_ports::ShareStoragePort, + common::errors::DomainError, + domain::entities::share::{Share, ShareItemType, SharePermissions}, +}; + +/// PostgreSQL implementation of [`ShareStoragePort`]. +/// +/// Replaces the legacy file-based `ShareFsRepository` that read/wrote the +/// entire `shares.json` on every operation. Each method now issues a single +/// indexed SQL statement — O(1) lookups, ACID transactions, and no data-race +/// risk. +pub struct SharePgRepository { + db_pool: Arc, +} + +impl SharePgRepository { + pub fn new(db_pool: Arc) -> Self { + Self { db_pool } + } + + /// Maps a [`sqlx::postgres::PgRow`] to the domain [`Share`] entity. + fn row_to_entity(row: &sqlx::postgres::PgRow) -> Result { + let id: String = row.try_get("id").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read id: {e}")) + })?; + let item_id: String = row.try_get("item_id").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read item_id: {e}")) + })?; + let item_name: Option = row.try_get("item_name").unwrap_or(None); + let item_type_str: String = row.try_get("item_type").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read item_type: {e}")) + })?; + let token: String = row.try_get("token").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read token: {e}")) + })?; + let password_hash: Option = row.try_get("password_hash").unwrap_or(None); + let expires_at: Option = row.try_get("expires_at").unwrap_or(None); + let permissions_read: bool = row.try_get("permissions_read").unwrap_or(true); + let permissions_write: bool = row.try_get("permissions_write").unwrap_or(false); + let permissions_reshare: bool = row.try_get("permissions_reshare").unwrap_or(false); + let created_at: i64 = row.try_get("created_at").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read created_at: {e}")) + })?; + let created_by: String = row.try_get("created_by").map_err(|e| { + DomainError::internal_error("Share", format!("Failed to read created_by: {e}")) + })?; + let access_count: i64 = row.try_get("access_count").unwrap_or(0); + + let item_type = + ShareItemType::try_from(item_type_str.as_str()).unwrap_or(ShareItemType::File); + let permissions = SharePermissions::new(permissions_read, permissions_write, permissions_reshare); + + Ok(Share::from_raw( + id, + item_id, + item_name, + item_type, + token, + password_hash, + expires_at.map(|v| v as u64), + permissions, + created_at as u64, + created_by, + access_count as u64, + )) + } +} + +#[async_trait] +impl ShareStoragePort for SharePgRepository { + async fn save_share(&self, share: &Share) -> Result { + let row = sqlx::query( + r#" + INSERT INTO storage.shares + (id, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count) + VALUES + ($1::UUID, $2, $3, $4, $5, $6, + $7, $8, $9, $10, + $11, $12, $13) + ON CONFLICT (id) DO UPDATE SET + item_name = EXCLUDED.item_name, + password_hash = EXCLUDED.password_hash, + expires_at = EXCLUDED.expires_at, + permissions_read = EXCLUDED.permissions_read, + permissions_write = EXCLUDED.permissions_write, + permissions_reshare = EXCLUDED.permissions_reshare, + access_count = EXCLUDED.access_count + RETURNING + id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + "#, + ) + .bind(share.id()) + .bind(share.item_id()) + .bind(share.item_name()) + .bind(share.item_type().to_string()) + .bind(share.token()) + .bind(share.password_hash()) + .bind(share.expires_at().map(|v| v as i64)) + .bind(share.permissions().read()) + .bind(share.permissions().write()) + .bind(share.permissions().reshare()) + .bind(share.created_at() as i64) + .bind(share.created_by()) + .bind(share.access_count() as i64) + .fetch_one(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error saving share: {}", e); + DomainError::internal_error("Share", format!("Failed to save share: {e}")) + })?; + + Self::row_to_entity(&row) + } + + async fn find_share_by_id(&self, id: &str) -> Result { + let row = sqlx::query( + r#" + SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + FROM storage.shares + WHERE id = $1::UUID + "#, + ) + .bind(id) + .fetch_optional(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error finding share by id: {}", e); + DomainError::internal_error("Share", format!("Failed to find share: {e}")) + })?; + + match row { + Some(r) => Self::row_to_entity(&r), + None => Err(DomainError::not_found( + "Share", + format!("Share with ID {id} not found"), + )), + } + } + + async fn find_share_by_token(&self, token: &str) -> Result { + let row = sqlx::query( + r#" + SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + FROM storage.shares + WHERE token = $1 + "#, + ) + .bind(token) + .fetch_optional(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error finding share by token: {}", e); + DomainError::internal_error("Share", format!("Failed to find share by token: {e}")) + })?; + + match row { + Some(r) => Self::row_to_entity(&r), + None => Err(DomainError::not_found( + "Share", + format!("Share with token {token} not found"), + )), + } + } + + async fn find_shares_by_item( + &self, + item_id: &str, + item_type: &ShareItemType, + ) -> Result, DomainError> { + let rows = sqlx::query( + r#" + SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + FROM storage.shares + WHERE item_id = $1 AND item_type = $2 + ORDER BY created_at DESC + "#, + ) + .bind(item_id) + .bind(item_type.to_string()) + .fetch_all(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error finding shares by item: {}", e); + DomainError::internal_error("Share", format!("Failed to find shares by item: {e}")) + })?; + + rows.iter().map(Self::row_to_entity).collect() + } + + async fn update_share(&self, share: &Share) -> Result { + let row = sqlx::query( + r#" + UPDATE storage.shares SET + item_name = $2, + password_hash = $3, + expires_at = $4, + permissions_read = $5, + permissions_write = $6, + permissions_reshare = $7, + access_count = $8 + WHERE id = $1::UUID + RETURNING + id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count + "#, + ) + .bind(share.id()) + .bind(share.item_name()) + .bind(share.password_hash()) + .bind(share.expires_at().map(|v| v as i64)) + .bind(share.permissions().read()) + .bind(share.permissions().write()) + .bind(share.permissions().reshare()) + .bind(share.access_count() as i64) + .fetch_optional(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error updating share: {}", e); + DomainError::internal_error("Share", format!("Failed to update share: {e}")) + })?; + + match row { + Some(r) => Self::row_to_entity(&r), + None => Err(DomainError::not_found( + "Share", + format!("Share with ID {} not found for update", share.id()), + )), + } + } + + async fn delete_share(&self, id: &str) -> Result<(), DomainError> { + let result = sqlx::query( + "DELETE FROM storage.shares WHERE id = $1::UUID", + ) + .bind(id) + .execute(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error deleting share: {}", e); + DomainError::internal_error("Share", format!("Failed to delete share: {e}")) + })?; + + if result.rows_affected() == 0 { + return Err(DomainError::not_found( + "Share", + format!("Share with ID {id} not found for deletion"), + )); + } + + Ok(()) + } + + async fn find_shares_by_user( + &self, + user_id: &str, + offset: usize, + limit: usize, + ) -> Result<(Vec, usize), DomainError> { + // Single query with window function — count + rows in one roundtrip + let rows = sqlx::query( + r#" + SELECT id::TEXT, item_id, item_name, item_type, token, password_hash, + expires_at, permissions_read, permissions_write, permissions_reshare, + created_at, created_by, access_count, + COUNT(*) OVER() AS total_count + FROM storage.shares + WHERE created_by = $1 + ORDER BY created_at DESC + LIMIT $2 OFFSET $3 + "#, + ) + .bind(user_id) + .bind(limit as i64) + .bind(offset as i64) + .fetch_all(&*self.db_pool) + .await + .map_err(|e| { + tracing::error!("Database error finding shares by user: {}", e); + DomainError::internal_error("Share", format!("Failed to find shares by user: {e}")) + })?; + + let total: usize = rows + .first() + .and_then(|r| r.try_get::("total_count").ok()) + .unwrap_or(0) as usize; + + let shares: Result, DomainError> = + rows.iter().map(Self::row_to_entity).collect(); + + Ok((shares?, total)) + } +} diff --git a/src/infrastructure/repositories/share_fs_repository.rs b/src/infrastructure/repositories/share_fs_repository.rs deleted file mode 100644 index 50e8f6c7..00000000 --- a/src/infrastructure/repositories/share_fs_repository.rs +++ /dev/null @@ -1,285 +0,0 @@ -use std::{path::Path, sync::Arc}; - -use async_trait::async_trait; -use serde::{Deserialize, Serialize}; -use tokio::{fs, io}; - -use crate::{ - application::ports::share_ports::ShareStoragePort, - common::{config::AppConfig, errors::DomainError}, - domain::entities::share::{Share, ShareItemType}, -}; - -// Structure for storing in the file system -#[derive(Debug, Clone, Serialize, Deserialize)] -struct ShareRecord { - id: String, - item_id: String, - #[serde(default)] - item_name: Option, - item_type: String, - token: String, - password_hash: Option, - expires_at: Option, - permissions_read: bool, - permissions_write: bool, - permissions_reshare: bool, - created_at: u64, - created_by: String, - access_count: u64, -} - -pub struct ShareFsRepository { - config: Arc, -} - -impl ShareFsRepository { - pub fn new(config: Arc) -> Self { - Self { config } - } - - /// Gets the path to the JSON file where shared links are stored - fn get_shares_path(&self) -> String { - format!("{}/shares.json", self.config.storage_path.display()) - } - - /// Reads all shared links from the JSON file - async fn read_shares(&self) -> Result, io::Error> { - let path = self.get_shares_path(); - let path = Path::new(&path); - - if !path.exists() { - return Ok(Vec::new()); - } - - let content = fs::read_to_string(path).await?; - let shares: Vec = serde_json::from_str(&content).unwrap_or_default(); - - Ok(shares) - } - - /// Saves all shared links to the JSON file - async fn write_shares(&self, shares: &[ShareRecord]) -> Result<(), io::Error> { - let path = self.get_shares_path(); - let json = serde_json::to_string_pretty(shares)?; - - // Make sure the directory exists - let dir = Path::new(&path).parent().unwrap(); - if !dir.exists() { - fs::create_dir_all(dir).await? - } - - fs::write(path, json).await - } - - /// Converts a file system record to a domain entity - fn to_entity(&self, record: &ShareRecord) -> Share { - let item_type = - ShareItemType::try_from(record.item_type.as_str()).unwrap_or(ShareItemType::File); - - let permissions = crate::domain::entities::share::SharePermissions::new( - record.permissions_read, - record.permissions_write, - record.permissions_reshare, - ); - - Share::from_raw( - record.id.clone(), - record.item_id.clone(), - record.item_name.clone(), - item_type, - record.token.clone(), - record.password_hash.clone(), - record.expires_at, - permissions, - record.created_at, - record.created_by.clone(), - record.access_count, - ) - } - - /// Converts a domain entity to a file system record - fn to_record(&self, share: &Share) -> ShareRecord { - ShareRecord { - id: share.id().to_string(), - item_id: share.item_id().to_string(), - item_name: share.item_name().map(|s| s.to_string()), - item_type: share.item_type().to_string(), - token: share.token().to_string(), - password_hash: share.password_hash().map(|s| s.to_string()), - expires_at: share.expires_at(), - permissions_read: share.permissions().read(), - permissions_write: share.permissions().write(), - permissions_reshare: share.permissions().reshare(), - created_at: share.created_at(), - created_by: share.created_by().to_string(), - access_count: share.access_count(), - } - } -} - -#[async_trait] -impl ShareStoragePort for ShareFsRepository { - async fn save_share(&self, share: &Share) -> Result { - let mut shares = self - .read_shares() - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - // Check if the link already exists - let existing_index = shares.iter().position(|s| s.id == share.id()); - - let record = self.to_record(share); - - if let Some(index) = existing_index { - // Update - shares[index] = record; - } else { - // Insert - shares.push(record); - } - - self.write_shares(&shares) - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - Ok(share.clone()) - } - - async fn find_share_by_id(&self, id: &str) -> Result { - let shares = self - .read_shares() - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - let share = shares.iter().find(|s| s.id == id).ok_or_else(|| { - DomainError::not_found("Share", format!("Share with ID {} not found", id)) - }); - - match share { - Ok(record) => Ok(self.to_entity(record)), - Err(e) => Err(e), - } - } - - async fn find_share_by_token(&self, token: &str) -> Result { - let shares = self - .read_shares() - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - let share = shares.iter().find(|s| s.token == token).ok_or_else(|| { - DomainError::not_found("Share", format!("Share with token {} not found", token)) - }); - - match share { - Ok(record) => Ok(self.to_entity(record)), - Err(e) => Err(e), - } - } - - async fn find_shares_by_item( - &self, - item_id: &str, - item_type: &ShareItemType, - ) -> Result, DomainError> { - let shares = self - .read_shares() - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - let type_str = item_type.to_string(); - let result: Vec = shares - .iter() - .filter(|s| s.item_id == item_id && s.item_type == type_str) - .map(|record| self.to_entity(record)) - .collect(); - - Ok(result) - } - - async fn update_share(&self, share: &Share) -> Result { - let mut shares = self - .read_shares() - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - // Find the index of the link to update - let index = shares - .iter() - .position(|s| s.id == share.id()) - .ok_or_else(|| { - DomainError::not_found( - "Share", - format!("Share with ID {} not found for update", share.id()), - ) - })?; - - // Update the record - shares[index] = self.to_record(share); - - // Save changes - self.write_shares(&shares) - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - Ok(share.clone()) - } - - async fn delete_share(&self, id: &str) -> Result<(), DomainError> { - let mut shares = self - .read_shares() - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - // Find the index of the link to delete - let initial_len = shares.len(); - shares.retain(|s| s.id != id); - - // If no link was deleted, it means it didn't exist - if shares.len() == initial_len { - return Err(DomainError::not_found( - "Share", - format!("Share with ID {} not found for deletion", id), - )); - } - - // Save changes - self.write_shares(&shares) - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - Ok(()) - } - - async fn find_shares_by_user( - &self, - user_id: &str, - offset: usize, - limit: usize, - ) -> Result<(Vec, usize), DomainError> { - let shares = self - .read_shares() - .await - .map_err(|e| DomainError::internal_error("Share", e.to_string()))?; - - // Filter the user's links - let user_shares: Vec = shares - .into_iter() - .filter(|s| s.created_by == user_id) - .collect(); - - // Calculate the total - let total = user_shares.len(); - - // Apply pagination - let paginated: Vec = user_shares - .iter() - .skip(offset) - .take(limit) - .map(|record| self.to_entity(record)) - .collect(); - - Ok((paginated, total)) - } -}