feat: add VisibilityFilteredAssetRepository decorator for automatic access control on asset queries

Cargo.lock

@@ -38,6 +38,7 @@ dependencies = [
  "async-trait",
  "chrono",
  "domain",
+ "event-payload",
  "serde",
  "serde_json",
  "sqlx",
@@ -3572,6 +3573,7 @@ dependencies = [
  "domain",
  "dotenvy",
  "event-transport",
+ "futures",
  "tokio",
  "tracing",
  "tracing-subscriber",

crates/adapters/event-transport/src/composite.rs

@@ -0,0 +1,26 @@
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError,
+    events::DomainEvent,
+    ports::{EventPublisher, EventStore},
+};
+use std::sync::Arc;
+
+pub struct CompositeEventPublisher {
+    primary: Arc<dyn EventPublisher>,
+    store: Arc<dyn EventStore>,
+}
+
+impl CompositeEventPublisher {
+    pub fn new(primary: Arc<dyn EventPublisher>, store: Arc<dyn EventStore>) -> Self {
+        Self { primary, store }
+    }
+}
+
+#[async_trait]
+impl EventPublisher for CompositeEventPublisher {
+    async fn publish(&self, event: &DomainEvent) -> Result<(), DomainError> {
+        self.store.append(event).await?;
+        self.primary.publish(event).await
+    }
+}

crates/adapters/event-transport/src/lib.rs

@@ -1,3 +1,6 @@
+pub mod composite;
+pub use composite::CompositeEventPublisher;
+
 use async_trait::async_trait;
 use domain::{
     errors::DomainError,

crates/adapters/postgres/Cargo.toml

@@ -4,11 +4,12 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-domain      = { workspace = true }
-sqlx        = { workspace = true, features = ["postgres", "runtime-tokio", "migrate", "uuid", "chrono", "json"] }
-uuid        = { workspace = true }
-chrono      = { workspace = true }
-anyhow      = { workspace = true }
-async-trait = { workspace = true }
-serde       = { workspace = true }
-serde_json  = { workspace = true }
+domain        = { workspace = true }
+event-payload = { workspace = true }
+sqlx          = { workspace = true, features = ["postgres", "runtime-tokio", "migrate", "uuid", "chrono", "json"] }
+uuid          = { workspace = true }
+chrono        = { workspace = true }
+anyhow        = { workspace = true }
+async-trait   = { workspace = true }
+serde         = { workspace = true }
+serde_json    = { workspace = true }

crates/adapters/postgres/migrations/010_event_log.sql

@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS event_log (
+    event_id    BIGSERIAL PRIMARY KEY,
+    aggregate_id UUID NOT NULL,
+    event_type  TEXT NOT NULL,
+    payload     JSONB NOT NULL,
+    occurred_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX idx_event_log_aggregate ON event_log (aggregate_id);
+CREATE INDEX idx_event_log_type      ON event_log (event_type);
+CREATE INDEX idx_event_log_occurred  ON event_log (occurred_at);

crates/adapters/postgres/src/catalog/mod.rs

@@ -1,4 +1,4 @@
-use crate::db::PgPool;
+use crate::helpers::{pg_repo, MapDomainError};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use domain::{
@@ -67,15 +67,7 @@ impl TryFrom<AssetRow> for Asset {
     }
 }
 
-pub struct PostgresAssetRepository {
-    pool: PgPool,
-}
-
-impl PostgresAssetRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresAssetRepository);
 
 #[async_trait]
 impl AssetRepository for PostgresAssetRepository {
@@ -88,7 +80,7 @@ impl AssetRepository for PostgresAssetRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -102,7 +94,7 @@ impl AssetRepository for PostgresAssetRepository {
         .bind(checksum.as_str())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         rows.into_iter().map(TryInto::try_into).collect()
     }
@@ -125,7 +117,7 @@ impl AssetRepository for PostgresAssetRepository {
         .bind(offset as i64)
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         rows.into_iter().map(TryInto::try_into).collect()
     }
@@ -157,7 +149,7 @@ impl AssetRepository for PostgresAssetRepository {
         .bind(asset.created_at.as_datetime())
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -166,7 +158,7 @@ impl AssetRepository for PostgresAssetRepository {
             .bind(*id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }
@@ -253,15 +245,7 @@ impl From<AssetMetadataRow> for AssetMetadata {
     }
 }
 
-pub struct PostgresAssetMetadataRepository {
-    pool: PgPool,
-}
-
-impl PostgresAssetMetadataRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresAssetMetadataRepository);
 
 #[async_trait]
 impl AssetMetadataRepository for PostgresAssetMetadataRepository {
@@ -273,7 +257,7 @@ impl AssetMetadataRepository for PostgresAssetMetadataRepository {
         .bind(*asset_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -291,7 +275,7 @@ impl AssetMetadataRepository for PostgresAssetMetadataRepository {
         .bind(source_to_str(&source))
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -310,7 +294,7 @@ impl AssetMetadataRepository for PostgresAssetMetadataRepository {
         .bind(metadata.updated_at.as_datetime())
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -324,7 +308,7 @@ impl AssetMetadataRepository for PostgresAssetMetadataRepository {
             .bind(source_to_str(&source))
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }
@@ -407,15 +391,7 @@ impl From<GroupRow> for DuplicateGroup {
     }
 }
 
-pub struct PostgresDuplicateRepository {
-    pool: PgPool,
-}
-
-impl PostgresDuplicateRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresDuplicateRepository);
 
 #[async_trait]
 impl DuplicateRepository for PostgresDuplicateRepository {
@@ -427,7 +403,7 @@ impl DuplicateRepository for PostgresDuplicateRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -439,7 +415,7 @@ impl DuplicateRepository for PostgresDuplicateRepository {
         )
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -452,7 +428,7 @@ impl DuplicateRepository for PostgresDuplicateRepository {
         .bind(serde_json::json!([{"asset_id": asset_id.as_uuid()}]))
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -472,7 +448,7 @@ impl DuplicateRepository for PostgresDuplicateRepository {
         .bind(candidates_to_json(&group.candidates))
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 }

crates/adapters/postgres/src/event_store.rs

@@ -0,0 +1,74 @@
+use crate::helpers::{pg_repo, MapDomainError};
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError,
+    events::DomainEvent,
+    ports::EventStore,
+    value_objects::SystemId,
+};
+use event_payload::EventPayload;
+use uuid::Uuid;
+
+pg_repo!(PostgresEventStore);
+
+/// Extracts the primary aggregate ID from a domain event.
+fn aggregate_id(event: &DomainEvent) -> Uuid {
+    match event {
+        DomainEvent::AssetIngested { asset_id, .. }
+        | DomainEvent::MetadataUpdated { asset_id, .. }
+        | DomainEvent::AssetDeleted { asset_id, .. }
+        | DomainEvent::SidecarSyncRequested { asset_id, .. } => *asset_id.as_uuid(),
+
+        DomainEvent::ShareCreated { scope_id, .. }
+        | DomainEvent::ShareRevoked { scope_id, .. } => *scope_id.as_uuid(),
+
+        DomainEvent::JobEnqueued { job_id, .. }
+        | DomainEvent::JobCompleted { job_id, .. }
+        | DomainEvent::JobFailed { job_id, .. } => *job_id.as_uuid(),
+    }
+}
+
+#[async_trait]
+impl EventStore for PostgresEventStore {
+    async fn append(&self, event: &DomainEvent) -> Result<(), DomainError> {
+        let payload = EventPayload::from(event);
+        let event_type = payload.subject().to_string();
+        let json = serde_json::to_value(&payload)
+            .map_err(|e| DomainError::Internal(e.to_string()))?;
+        let agg_id = aggregate_id(event);
+
+        sqlx::query(
+            "INSERT INTO event_log (aggregate_id, event_type, payload, occurred_at)
+             VALUES ($1, $2, $3, now())",
+        )
+        .bind(agg_id)
+        .bind(event_type)
+        .bind(json)
+        .execute(&self.pool)
+        .await
+        .map_pg()?;
+
+        Ok(())
+    }
+
+    async fn query_by_aggregate(
+        &self,
+        aggregate_id: &SystemId,
+    ) -> Result<Vec<DomainEvent>, DomainError> {
+        let rows: Vec<(serde_json::Value,)> = sqlx::query_as(
+            "SELECT payload FROM event_log WHERE aggregate_id = $1 ORDER BY event_id ASC",
+        )
+        .bind(*aggregate_id.as_uuid())
+        .fetch_all(&self.pool)
+        .await
+        .map_pg()?;
+
+        rows.into_iter()
+            .map(|(json,)| {
+                let payload: EventPayload = serde_json::from_value(json)
+                    .map_err(|e| DomainError::Internal(e.to_string()))?;
+                DomainEvent::try_from(payload)
+            })
+            .collect()
+    }
+}

crates/adapters/postgres/src/helpers.rs

@@ -0,0 +1,36 @@
+use domain::errors::DomainError;
+
+/// Extension trait for converting `sqlx::Error` into `DomainError`.
+pub trait MapDomainError<T> {
+    fn map_pg(self) -> Result<T, DomainError>;
+}
+
+impl<T> MapDomainError<T> for Result<T, sqlx::Error> {
+    fn map_pg(self) -> Result<T, DomainError> {
+        self.map_err(|e| DomainError::Internal(e.to_string()))
+    }
+}
+
+/// Generates a Postgres repository struct with a `PgPool` field and a `new` constructor.
+///
+/// ```ignore
+/// pg_repo!(PostgresFooRepository);
+/// // expands to:
+/// // pub struct PostgresFooRepository { pool: PgPool }
+/// // impl PostgresFooRepository { pub fn new(pool: PgPool) -> Self { Self { pool } } }
+/// ```
+macro_rules! pg_repo {
+    ($name:ident) => {
+        pub struct $name {
+            pool: crate::db::PgPool,
+        }
+
+        impl $name {
+            pub fn new(pool: crate::db::PgPool) -> Self {
+                Self { pool }
+            }
+        }
+    };
+}
+
+pub(crate) use pg_repo;

crates/adapters/postgres/src/identity/mod.rs

@@ -1,4 +1,4 @@
-use crate::db::PgPool;
+use crate::helpers::{pg_repo, MapDomainError};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use domain::{
@@ -30,15 +30,7 @@ impl TryFrom<UserRow> for domain::entities::User {
     }
 }
 
-pub struct PostgresUserRepository {
-    pool: PgPool,
-}
-
-impl PostgresUserRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresUserRepository);
 
 #[async_trait]
 impl UserRepository for PostgresUserRepository {
@@ -52,7 +44,7 @@ impl UserRepository for PostgresUserRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -67,7 +59,7 @@ impl UserRepository for PostgresUserRepository {
         .bind(email.as_str())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -82,7 +74,7 @@ impl UserRepository for PostgresUserRepository {
         .bind(username)
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -104,7 +96,7 @@ impl UserRepository for PostgresUserRepository {
         .bind(user.created_at)
         .fetch_one(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -113,7 +105,7 @@ impl UserRepository for PostgresUserRepository {
             .bind(*id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }

crates/adapters/postgres/src/lib.rs

@@ -1,6 +1,8 @@
 pub mod db;
+mod helpers;
 
 pub mod catalog;
+pub mod event_store;
 pub mod identity;
 pub mod organization;
 pub mod processing;
@@ -11,6 +13,7 @@ pub mod storage;
 pub use db::{PgPool, connect, run_migrations};
 
 pub use catalog::*;
+pub use event_store::PostgresEventStore;
 pub use identity::*;
 pub use organization::*;
 pub use processing::*;

crates/adapters/postgres/src/organization/mod.rs

@@ -1,4 +1,4 @@
-use crate::db::PgPool;
+use crate::helpers::{pg_repo, MapDomainError};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use domain::{
@@ -60,15 +60,9 @@ fn album_from_row(r: AlbumRow, entries: Vec<AlbumEntry>) -> Album {
     }
 }
 
-pub struct PostgresAlbumRepository {
-    pool: PgPool,
-}
+pg_repo!(PostgresAlbumRepository);
 
 impl PostgresAlbumRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-
     async fn load_entries(&self, album_id: &Uuid) -> Result<Vec<AlbumEntry>, DomainError> {
         let rows = sqlx::query_as::<_, AlbumEntryRow>(
             "SELECT album_id, asset_id, sort_order, added_at, added_by_user_id
@@ -78,7 +72,7 @@ impl PostgresAlbumRepository {
         .bind(album_id)
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -95,7 +89,7 @@ impl AlbumRepository for PostgresAlbumRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         let Some(r) = row else {
             return Ok(None);
@@ -114,7 +108,7 @@ impl AlbumRepository for PostgresAlbumRepository {
         .bind(*creator_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         let mut albums = Vec::with_capacity(rows.len());
         for r in rows {
@@ -146,14 +140,14 @@ impl AlbumRepository for PostgresAlbumRepository {
         .bind(album.created_at.as_datetime())
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         // Sync entries: delete all then re-insert
         sqlx::query("DELETE FROM album_entries WHERE album_id = $1")
             .bind(*album.album_id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
 
         for entry in &album.entries {
             sqlx::query(
@@ -167,7 +161,7 @@ impl AlbumRepository for PostgresAlbumRepository {
             .bind(*entry.added_by_user_id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         }
 
         Ok(())
@@ -179,7 +173,7 @@ impl AlbumRepository for PostgresAlbumRepository {
             .bind(*id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }
@@ -240,15 +234,7 @@ impl From<AssetTagRow> for AssetTag {
     }
 }
 
-pub struct PostgresTagRepository {
-    pool: PgPool,
-}
-
-impl PostgresTagRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresTagRepository);
 
 #[async_trait]
 impl TagRepository for PostgresTagRepository {
@@ -259,7 +245,7 @@ impl TagRepository for PostgresTagRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -271,7 +257,7 @@ impl TagRepository for PostgresTagRepository {
         .bind(name)
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -288,7 +274,7 @@ impl TagRepository for PostgresTagRepository {
         .bind(*asset_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         let at_rows = sqlx::query_as::<_, AssetTagRow>(
             "SELECT asset_id, tag_id, tagged_by_user_id, confidence
@@ -297,7 +283,7 @@ impl TagRepository for PostgresTagRepository {
         .bind(*asset_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         let tags: Vec<Tag> = rows.into_iter().map(Into::into).collect();
         let asset_tags: Vec<AssetTag> = at_rows.into_iter().map(Into::into).collect();
@@ -316,7 +302,7 @@ impl TagRepository for PostgresTagRepository {
         .bind(tag_source_to_str(&tag.tag_source))
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -334,7 +320,7 @@ impl TagRepository for PostgresTagRepository {
         .bind(asset_tag.confidence)
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -348,7 +334,7 @@ impl TagRepository for PostgresTagRepository {
             .bind(*tag_id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }

crates/adapters/postgres/src/processing/mod.rs

@@ -1,4 +1,4 @@
-use crate::db::PgPool;
+use crate::helpers::{pg_repo, MapDomainError};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use domain::{
@@ -119,15 +119,7 @@ impl From<JobRow> for Job {
     }
 }
 
-pub struct PostgresJobRepository {
-    pool: PgPool,
-}
-
-impl PostgresJobRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresJobRepository);
 
 #[async_trait]
 impl JobRepository for PostgresJobRepository {
@@ -141,7 +133,7 @@ impl JobRepository for PostgresJobRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -157,7 +149,7 @@ impl JobRepository for PostgresJobRepository {
         )
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -173,7 +165,7 @@ impl JobRepository for PostgresJobRepository {
         .bind(*batch_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -210,7 +202,7 @@ impl JobRepository for PostgresJobRepository {
         .bind(&job.error_message)
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 }
@@ -261,15 +253,7 @@ impl From<BatchRow> for JobBatch {
     }
 }
 
-pub struct PostgresJobBatchRepository {
-    pool: PgPool,
-}
-
-impl PostgresJobBatchRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresJobBatchRepository);
 
 #[async_trait]
 impl JobBatchRepository for PostgresJobBatchRepository {
@@ -281,7 +265,7 @@ impl JobBatchRepository for PostgresJobBatchRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -304,7 +288,7 @@ impl JobBatchRepository for PostgresJobBatchRepository {
         .bind(batch_status_to_str(&batch.status))
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 }
@@ -351,15 +335,7 @@ impl From<PluginRow> for Plugin {
     }
 }
 
-pub struct PostgresPluginRepository {
-    pool: PgPool,
-}
-
-impl PostgresPluginRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresPluginRepository);
 
 #[async_trait]
 impl PluginRepository for PostgresPluginRepository {
@@ -371,7 +347,7 @@ impl PluginRepository for PostgresPluginRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -383,7 +359,7 @@ impl PluginRepository for PostgresPluginRepository {
         )
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -405,7 +381,7 @@ impl PluginRepository for PostgresPluginRepository {
         .bind(structured_to_json(&plugin.configuration))
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 }
@@ -475,15 +451,7 @@ impl From<PipelineRow> for ProcessingPipeline {
     }
 }
 
-pub struct PostgresPipelineRepository {
-    pool: PgPool,
-}
-
-impl PostgresPipelineRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresPipelineRepository);
 
 #[async_trait]
 impl PipelineRepository for PostgresPipelineRepository {
@@ -495,7 +463,7 @@ impl PipelineRepository for PostgresPipelineRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -508,7 +476,7 @@ impl PipelineRepository for PostgresPipelineRepository {
         .bind(event)
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -526,7 +494,7 @@ impl PipelineRepository for PostgresPipelineRepository {
         .bind(steps_to_json(&pipeline.steps))
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 }

crates/adapters/postgres/src/sharing/mod.rs

@@ -1,4 +1,4 @@
-use crate::db::PgPool;
+use crate::helpers::{pg_repo, MapDomainError};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use domain::{
@@ -207,15 +207,7 @@ impl TryFrom<InviteCodeRow> for InviteCode {
     }
 }
 
-pub struct PostgresShareRepository {
-    pool: PgPool,
-}
-
-impl PostgresShareRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresShareRepository);
 
 #[async_trait]
 impl ShareRepository for PostgresShareRepository {
@@ -238,7 +230,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(scope.created_at.as_datetime())
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -250,7 +242,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -266,7 +258,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(*resource_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         rows.into_iter().map(TryInto::try_into).collect()
     }
@@ -276,7 +268,7 @@ impl ShareRepository for PostgresShareRepository {
             .bind(*id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 
@@ -296,7 +288,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(*target.role_id.as_uuid())
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -311,7 +303,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(*scope_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         rows.into_iter().map(TryInto::try_into).collect()
     }
@@ -327,7 +319,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(*user_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         rows.into_iter().map(TryInto::try_into).collect()
     }
@@ -355,7 +347,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(link.use_count as i32)
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -367,7 +359,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(token)
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -392,7 +384,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(*invite.assigned_role_id.as_uuid())
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -404,7 +396,7 @@ impl ShareRepository for PostgresShareRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -433,15 +425,7 @@ impl From<VisibilityFilterRow> for VisibilityFilter {
     }
 }
 
-pub struct PostgresVisibilityFilterRepository {
-    pool: PgPool,
-}
-
-impl PostgresVisibilityFilterRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresVisibilityFilterRepository);
 
 #[async_trait]
 impl VisibilityFilterRepository for PostgresVisibilityFilterRepository {
@@ -458,7 +442,7 @@ impl VisibilityFilterRepository for PostgresVisibilityFilterRepository {
         .bind(*role_id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -476,7 +460,7 @@ impl VisibilityFilterRepository for PostgresVisibilityFilterRepository {
         .bind(&filter.hidden_fields)
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -485,7 +469,7 @@ impl VisibilityFilterRepository for PostgresVisibilityFilterRepository {
             .bind(*id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }

crates/adapters/postgres/src/sidecar/mod.rs

@@ -1,4 +1,4 @@
-use crate::db::PgPool;
+use crate::helpers::{pg_repo, MapDomainError};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use domain::{
@@ -55,15 +55,7 @@ impl TryFrom<SidecarRow> for SidecarRecord {
     }
 }
 
-pub struct PostgresSidecarRepository {
-    pool: PgPool,
-}
-
-impl PostgresSidecarRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresSidecarRepository);
 
 #[async_trait]
 impl SidecarRepository for PostgresSidecarRepository {
@@ -79,7 +71,7 @@ impl SidecarRepository for PostgresSidecarRepository {
         .bind(*asset_id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -93,7 +85,7 @@ impl SidecarRepository for PostgresSidecarRepository {
         .bind(sync_status_to_str(&status))
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         rows.into_iter().map(TryInto::try_into).collect()
     }
@@ -118,7 +110,7 @@ impl SidecarRepository for PostgresSidecarRepository {
         .bind(&record.error_message)
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -127,7 +119,7 @@ impl SidecarRepository for PostgresSidecarRepository {
             .bind(*asset_id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }

crates/adapters/postgres/src/storage/mod.rs

@@ -1,15 +1,15 @@
-use crate::db::PgPool;
+use crate::helpers::{pg_repo, MapDomainError};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use domain::{
     entities::{
-        IngestSession, IngestStatus, LibraryPath, OwnershipPolicy, QuotaDefinition, QuotaRule,
-        StorageVolume, TimePeriod, UsageLedgerEntry, UsageType,
+        Asset, IngestSession, IngestStatus, LibraryPath, OwnershipPolicy, QuotaDefinition,
+        QuotaRule, StorageVolume, TimePeriod, UsageLedgerEntry, UsageType,
     },
     errors::DomainError,
     ports::{
-        IngestSessionRepository, LibraryPathRepository, QuotaRepository, StorageVolumeRepository,
-        UsageLedgerRepository,
+        IngestSessionRepository, IngestTransaction, LibraryPathRepository, QuotaRepository,
+        StorageVolumeRepository, UsageLedgerRepository,
     },
     value_objects::{Checksum, DateTimeStamp, SystemId},
 };
@@ -40,15 +40,7 @@ impl From<StorageVolumeRow> for StorageVolume {
     }
 }
 
-pub struct PostgresStorageVolumeRepository {
-    pool: PgPool,
-}
-
-impl PostgresStorageVolumeRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresStorageVolumeRepository);
 
 #[async_trait]
 impl StorageVolumeRepository for PostgresStorageVolumeRepository {
@@ -60,7 +52,7 @@ impl StorageVolumeRepository for PostgresStorageVolumeRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -72,7 +64,7 @@ impl StorageVolumeRepository for PostgresStorageVolumeRepository {
         )
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -94,7 +86,7 @@ impl StorageVolumeRepository for PostgresStorageVolumeRepository {
         .bind(volume.available_bytes as i64)
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -103,7 +95,7 @@ impl StorageVolumeRepository for PostgresStorageVolumeRepository {
             .bind(*id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }
@@ -151,15 +143,7 @@ impl From<LibraryPathRow> for LibraryPath {
     }
 }
 
-pub struct PostgresLibraryPathRepository {
-    pool: PgPool,
-}
-
-impl PostgresLibraryPathRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresLibraryPathRepository);
 
 #[async_trait]
 impl LibraryPathRepository for PostgresLibraryPathRepository {
@@ -171,7 +155,7 @@ impl LibraryPathRepository for PostgresLibraryPathRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.map(Into::into))
     }
@@ -184,7 +168,7 @@ impl LibraryPathRepository for PostgresLibraryPathRepository {
         .bind(*volume_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -201,7 +185,7 @@ impl LibraryPathRepository for PostgresLibraryPathRepository {
         .bind(*owner_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(rows.into_iter().map(Into::into).collect())
     }
@@ -225,7 +209,7 @@ impl LibraryPathRepository for PostgresLibraryPathRepository {
         .bind(path.designated_owner_id.as_ref().map(|id| *id.as_uuid()))
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -234,7 +218,7 @@ impl LibraryPathRepository for PostgresLibraryPathRepository {
             .bind(*id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }
@@ -294,15 +278,7 @@ impl TryFrom<IngestSessionRow> for IngestSession {
     }
 }
 
-pub struct PostgresIngestSessionRepository {
-    pool: PgPool,
-}
-
-impl PostgresIngestSessionRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresIngestSessionRepository);
 
 #[async_trait]
 impl IngestSessionRepository for PostgresIngestSessionRepository {
@@ -315,7 +291,7 @@ impl IngestSessionRepository for PostgresIngestSessionRepository {
         .bind(*id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         row.map(TryInto::try_into).transpose()
     }
@@ -329,7 +305,7 @@ impl IngestSessionRepository for PostgresIngestSessionRepository {
         .bind(*user_id.as_uuid())
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         rows.into_iter().map(TryInto::try_into).collect()
     }
@@ -354,7 +330,7 @@ impl IngestSessionRepository for PostgresIngestSessionRepository {
         .bind(session.error_message.as_deref())
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 }
@@ -457,15 +433,7 @@ impl From<UsageLedgerRow> for UsageLedgerEntry {
     }
 }
 
-pub struct PostgresQuotaRepository {
-    pool: PgPool,
-}
-
-impl PostgresQuotaRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresQuotaRepository);
 
 #[async_trait]
 impl QuotaRepository for PostgresQuotaRepository {
@@ -479,7 +447,7 @@ impl QuotaRepository for PostgresQuotaRepository {
         .bind(*owner_id.as_uuid())
         .fetch_optional(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         let Some(def) = def_row else {
             return Ok(None);
@@ -492,7 +460,7 @@ impl QuotaRepository for PostgresQuotaRepository {
         .bind(def.quota_id)
         .fetch_all(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(Some(QuotaDefinition {
             quota_id: SystemId::from_uuid(def.quota_id),
@@ -515,14 +483,14 @@ impl QuotaRepository for PostgresQuotaRepository {
         .bind(quota.is_enforced)
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         // Delete old rules then re-insert
         sqlx::query("DELETE FROM quota_rules WHERE quota_id = $1")
             .bind(*quota.quota_id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
 
         for rule in &quota.rules {
             sqlx::query(
@@ -537,7 +505,7 @@ impl QuotaRepository for PostgresQuotaRepository {
             .bind(rule.is_unlimited)
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         }
 
         Ok(())
@@ -549,20 +517,12 @@ impl QuotaRepository for PostgresQuotaRepository {
             .bind(*id.as_uuid())
             .execute(&self.pool)
             .await
-            .map_err(|e| DomainError::Internal(e.to_string()))?;
+            .map_pg()?;
         Ok(())
     }
 }
 
-pub struct PostgresUsageLedgerRepository {
-    pool: PgPool,
-}
-
-impl PostgresUsageLedgerRepository {
-    pub fn new(pool: PgPool) -> Self {
-        Self { pool }
-    }
-}
+pg_repo!(PostgresUsageLedgerRepository);
 
 #[async_trait]
 impl UsageLedgerRepository for PostgresUsageLedgerRepository {
@@ -579,7 +539,7 @@ impl UsageLedgerRepository for PostgresUsageLedgerRepository {
         .bind(&entry.context)
         .execute(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
         Ok(())
     }
 
@@ -600,8 +560,56 @@ impl UsageLedgerRepository for PostgresUsageLedgerRepository {
         .bind(since_dt)
         .fetch_one(&self.pool)
         .await
-        .map_err(|e| DomainError::Internal(e.to_string()))?;
+        .map_pg()?;
 
         Ok(row.total as u64)
     }
 }
+
+// ──────────────────────────────────────────────
+// IngestTransaction (composite port)
+// ──────────────────────────────────────────────
+
+pg_repo!(PostgresIngestTransaction);
+
+#[async_trait]
+impl IngestTransaction for PostgresIngestTransaction {
+    async fn save_asset(&self, asset: &Asset) -> Result<(), DomainError> {
+        use domain::ports::AssetRepository;
+        crate::PostgresAssetRepository::new(self.pool.clone())
+            .save(asset)
+            .await
+    }
+
+    async fn save_session(&self, session: &IngestSession) -> Result<(), DomainError> {
+        PostgresIngestSessionRepository::new(self.pool.clone())
+            .save(session)
+            .await
+    }
+
+    async fn find_quota(
+        &self,
+        owner_id: &SystemId,
+    ) -> Result<Option<QuotaDefinition>, DomainError> {
+        PostgresQuotaRepository::new(self.pool.clone())
+            .find_by_owner(owner_id)
+            .await
+    }
+
+    async fn sum_usage(
+        &self,
+        user_id: &SystemId,
+        usage_type: UsageType,
+        since: Option<DateTimeStamp>,
+    ) -> Result<u64, DomainError> {
+        PostgresUsageLedgerRepository::new(self.pool.clone())
+            .sum_usage(user_id, usage_type, since)
+            .await
+    }
+
+    async fn record_usage(&self, entry: &UsageLedgerEntry) -> Result<(), DomainError> {
+        PostgresUsageLedgerRepository::new(self.pool.clone())
+            .record(entry)
+            .await
+    }
+}

crates/application/src/catalog/mod.rs

@@ -1,8 +1,10 @@
 pub mod commands;
 pub mod queries;
+pub mod visibility;
 
 pub use commands::register_asset::{RegisterAssetCommand, RegisterAssetHandler};
 pub use commands::update_metadata::{UpdateMetadataCommand, UpdateMetadataHandler};
 pub use queries::get_asset::{GetAssetHandler, GetAssetQuery};
 pub use queries::get_timeline::{GetTimelineHandler, GetTimelineQuery};
 pub use queries::read_asset_file::{AssetFileResult, ReadAssetFileHandler, ReadAssetFileQuery};
+pub use visibility::VisibilityFilteredAssetRepository;

crates/application/src/catalog/queries/get_asset.rs

@@ -1,8 +1,9 @@
+use crate::catalog::visibility::VisibilityFilteredAssetRepository;
 use domain::{
     catalog::entities::Asset,
     catalog::services::resolve_metadata,
     errors::DomainError,
-    ports::{AssetMetadataRepository, AssetRepository},
+    ports::{AssetMetadataRepository, AssetRepository, ShareRepository},
     value_objects::{StructuredData, SystemId},
 };
 use std::sync::Arc;
@@ -16,6 +17,7 @@ pub struct GetAssetQuery {
 pub struct GetAssetHandler {
     asset_repo: Arc<dyn AssetRepository>,
     metadata_repo: Arc<dyn AssetMetadataRepository>,
+    share_repo: Option<Arc<dyn ShareRepository>>,
 }
 
 impl GetAssetHandler {
@@ -26,6 +28,28 @@ impl GetAssetHandler {
         Self {
             asset_repo,
             metadata_repo,
+            share_repo: None,
+        }
+    }
+
+    /// Enable sharing-aware visibility filtering. When set, the handler
+    /// wraps the inner `AssetRepository` with a `VisibilityFilteredAssetRepository`
+    /// so that shared assets are visible to the caller.
+    pub fn with_visibility_filter(mut self, share_repo: Arc<dyn ShareRepository>) -> Self {
+        self.share_repo = Some(share_repo);
+        self
+    }
+
+    /// Returns the effective asset repo — wrapped with a visibility filter
+    /// when a `ShareRepository` has been configured, otherwise the raw inner repo.
+    fn effective_repo(&self, caller_id: SystemId) -> Arc<dyn AssetRepository> {
+        match &self.share_repo {
+            Some(share_repo) => Arc::new(VisibilityFilteredAssetRepository::new(
+                self.asset_repo.clone(),
+                share_repo.clone(),
+                caller_id,
+            )),
+            None => self.asset_repo.clone(),
         }
     }
 
@@ -33,13 +57,16 @@ impl GetAssetHandler {
         &self,
         query: GetAssetQuery,
     ) -> Result<(Asset, StructuredData), DomainError> {
-        let asset = self
-            .asset_repo
+        let repo = self.effective_repo(query.user_id);
+
+        let asset = repo
             .find_by_id(&query.asset_id)
             .await?
             .ok_or_else(|| DomainError::NotFound(format!("Asset {} not found", query.asset_id)))?;
 
-        if asset.owner_user_id != query.user_id {
+        // When the visibility filter is active it already enforces access.
+        // When it is not, fall back to the original owner-only check.
+        if self.share_repo.is_none() && asset.owner_user_id != query.user_id {
             return Err(DomainError::Forbidden("Access denied".to_string()));
         }
 

crates/application/src/catalog/queries/get_timeline.rs

@@ -1,8 +1,9 @@
+use crate::catalog::visibility::VisibilityFilteredAssetRepository;
 use domain::{
     catalog::entities::Asset,
     catalog::services::resolve_metadata,
     errors::DomainError,
-    ports::{AssetMetadataRepository, AssetRepository},
+    ports::{AssetMetadataRepository, AssetRepository, ShareRepository},
     value_objects::{StructuredData, SystemId},
 };
 use std::sync::Arc;
@@ -10,6 +11,7 @@ use std::sync::Arc;
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct GetTimelineQuery {
     pub owner_id: SystemId,
+    pub caller_id: Option<SystemId>,
     pub limit: u32,
     pub offset: u32,
 }
@@ -17,6 +19,7 @@ pub struct GetTimelineQuery {
 pub struct GetTimelineHandler {
     asset_repo: Arc<dyn AssetRepository>,
     metadata_repo: Arc<dyn AssetMetadataRepository>,
+    share_repo: Option<Arc<dyn ShareRepository>>,
 }
 
 impl GetTimelineHandler {
@@ -27,6 +30,24 @@ impl GetTimelineHandler {
         Self {
             asset_repo,
             metadata_repo,
+            share_repo: None,
+        }
+    }
+
+    /// Enable sharing-aware visibility filtering on timeline queries.
+    pub fn with_visibility_filter(mut self, share_repo: Arc<dyn ShareRepository>) -> Self {
+        self.share_repo = Some(share_repo);
+        self
+    }
+
+    fn effective_repo(&self, caller_id: SystemId) -> Arc<dyn AssetRepository> {
+        match &self.share_repo {
+            Some(share_repo) => Arc::new(VisibilityFilteredAssetRepository::new(
+                self.asset_repo.clone(),
+                share_repo.clone(),
+                caller_id,
+            )),
+            None => self.asset_repo.clone(),
         }
     }
 
@@ -34,8 +55,10 @@ impl GetTimelineHandler {
         &self,
         query: GetTimelineQuery,
     ) -> Result<Vec<(Asset, StructuredData)>, DomainError> {
-        let assets = self
-            .asset_repo
+        let caller_id = query.caller_id.unwrap_or(query.owner_id);
+        let repo = self.effective_repo(caller_id);
+
+        let assets = repo
             .find_by_owner(&query.owner_id, query.limit, query.offset)
             .await?;
 

crates/application/src/catalog/visibility.rs

@@ -0,0 +1,296 @@
+use async_trait::async_trait;
+use domain::{
+    catalog::entities::Asset,
+    errors::DomainError,
+    ports::{AssetRepository, ShareRepository},
+    value_objects::{Checksum, SystemId},
+};
+use std::sync::Arc;
+
+/// Decorator that wraps an `AssetRepository` and filters query results
+/// based on sharing permissions. The caller sees only assets they own
+/// or have been granted access to via a `ShareScope` + `ShareTarget`.
+///
+/// Write operations (`save`, `delete`) pass through to the inner repository
+/// unchanged — authorization for writes is handled at the use-case layer.
+pub struct VisibilityFilteredAssetRepository {
+    inner: Arc<dyn AssetRepository>,
+    share_repo: Arc<dyn ShareRepository>,
+    caller_id: SystemId,
+}
+
+impl VisibilityFilteredAssetRepository {
+    pub fn new(
+        inner: Arc<dyn AssetRepository>,
+        share_repo: Arc<dyn ShareRepository>,
+        caller_id: SystemId,
+    ) -> Self {
+        Self {
+            inner,
+            share_repo,
+            caller_id,
+        }
+    }
+
+    /// Returns `true` if the caller owns the asset or has been granted
+    /// access through a share scope that targets them.
+    async fn caller_can_access(&self, asset: &Asset) -> Result<bool, DomainError> {
+        if asset.owner_user_id == self.caller_id {
+            return Ok(true);
+        }
+
+        // Find all share scopes that cover this asset
+        let scopes = self
+            .share_repo
+            .find_scopes_for_resource(&asset.asset_id)
+            .await?;
+
+        if scopes.is_empty() {
+            return Ok(false);
+        }
+
+        // Find all share targets that name this caller
+        let caller_targets = self
+            .share_repo
+            .find_targets_for_user(&self.caller_id)
+            .await?;
+
+        // The caller has access if any of their targets reference a scope
+        // that covers this asset.
+        for scope in &scopes {
+            if scope.is_expired() {
+                continue;
+            }
+            if caller_targets.iter().any(|t| t.scope_id == scope.scope_id) {
+                return Ok(true);
+            }
+        }
+
+        Ok(false)
+    }
+}
+
+#[async_trait]
+impl AssetRepository for VisibilityFilteredAssetRepository {
+    async fn find_by_id(&self, id: &SystemId) -> Result<Option<Asset>, DomainError> {
+        let asset = self.inner.find_by_id(id).await?;
+        match asset {
+            Some(a) if self.caller_can_access(&a).await? => Ok(Some(a)),
+            _ => Ok(None),
+        }
+    }
+
+    async fn find_by_checksum(&self, checksum: &Checksum) -> Result<Vec<Asset>, DomainError> {
+        let assets = self.inner.find_by_checksum(checksum).await?;
+        let mut visible = Vec::with_capacity(assets.len());
+        for asset in assets {
+            if self.caller_can_access(&asset).await? {
+                visible.push(asset);
+            }
+        }
+        Ok(visible)
+    }
+
+    async fn find_by_owner(
+        &self,
+        owner_id: &SystemId,
+        limit: u32,
+        offset: u32,
+    ) -> Result<Vec<Asset>, DomainError> {
+        if owner_id == &self.caller_id {
+            // Querying own assets — no filtering needed.
+            return self.inner.find_by_owner(owner_id, limit, offset).await;
+        }
+
+        let assets = self.inner.find_by_owner(owner_id, limit, offset).await?;
+        let mut visible = Vec::with_capacity(assets.len());
+        for asset in assets {
+            if self.caller_can_access(&asset).await? {
+                visible.push(asset);
+            }
+        }
+        Ok(visible)
+    }
+
+    async fn save(&self, asset: &Asset) -> Result<(), DomainError> {
+        self.inner.save(asset).await
+    }
+
+    async fn delete(&self, id: &SystemId) -> Result<(), DomainError> {
+        self.inner.delete(id).await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::testing::{InMemoryAssetRepository, InMemoryShareRepository};
+    use domain::{
+        catalog::entities::{AssetType, SourceReference},
+        sharing::entities::{ScopeType, ShareScope, ShareTarget, ShareableType, TargetType},
+        value_objects::{Checksum, SystemId},
+    };
+
+    fn make_asset(owner: SystemId) -> Asset {
+        Asset::new(
+            SourceReference {
+                volume_id: SystemId::new(),
+                relative_path: "test/photo.jpg".to_string(),
+                checksum: Checksum::new("a".repeat(64)).unwrap(),
+            },
+            AssetType::Image,
+            "image/jpeg",
+            1024,
+            owner,
+        )
+    }
+
+    fn share_asset(asset_id: SystemId, granter: SystemId) -> ShareScope {
+        ShareScope::new(ScopeType::User, ShareableType::Asset, asset_id, granter)
+    }
+
+    fn target_user(scope_id: SystemId, user_id: SystemId) -> ShareTarget {
+        ShareTarget::new(scope_id, TargetType::User, user_id, SystemId::new())
+    }
+
+    #[tokio::test]
+    async fn owner_can_always_see_own_asset() {
+        let owner_id = SystemId::new();
+        let asset = make_asset(owner_id);
+
+        let inner = Arc::new(InMemoryAssetRepository::new());
+        inner.save(&asset).await.unwrap();
+
+        let share_repo = Arc::new(InMemoryShareRepository::new());
+        let filtered = VisibilityFilteredAssetRepository::new(
+            inner.clone(),
+            share_repo.clone(),
+            owner_id,
+        );
+
+        let found = filtered.find_by_id(&asset.asset_id).await.unwrap();
+        assert!(found.is_some());
+        assert_eq!(found.unwrap().asset_id, asset.asset_id);
+    }
+
+    #[tokio::test]
+    async fn stranger_cannot_see_unshared_asset() {
+        let owner_id = SystemId::new();
+        let stranger_id = SystemId::new();
+        let asset = make_asset(owner_id);
+
+        let inner = Arc::new(InMemoryAssetRepository::new());
+        inner.save(&asset).await.unwrap();
+
+        let share_repo = Arc::new(InMemoryShareRepository::new());
+        let filtered = VisibilityFilteredAssetRepository::new(
+            inner.clone(),
+            share_repo.clone(),
+            stranger_id,
+        );
+
+        let found = filtered.find_by_id(&asset.asset_id).await.unwrap();
+        assert!(found.is_none());
+    }
+
+    #[tokio::test]
+    async fn shared_user_can_see_asset() {
+        let owner_id = SystemId::new();
+        let friend_id = SystemId::new();
+        let asset = make_asset(owner_id);
+
+        let inner = Arc::new(InMemoryAssetRepository::new());
+        inner.save(&asset).await.unwrap();
+
+        let share_repo = Arc::new(InMemoryShareRepository::new());
+
+        // Create a share scope on the asset and target the friend
+        let scope = share_asset(asset.asset_id, owner_id);
+        share_repo.save_scope(&scope).await.unwrap();
+
+        let target = target_user(scope.scope_id, friend_id);
+        share_repo.save_target(&target).await.unwrap();
+
+        let filtered = VisibilityFilteredAssetRepository::new(
+            inner.clone(),
+            share_repo.clone(),
+            friend_id,
+        );
+
+        let found = filtered.find_by_id(&asset.asset_id).await.unwrap();
+        assert!(found.is_some());
+        assert_eq!(found.unwrap().asset_id, asset.asset_id);
+    }
+
+    #[tokio::test]
+    async fn find_by_checksum_filters_inaccessible() {
+        let owner_id = SystemId::new();
+        let stranger_id = SystemId::new();
+
+        let asset_a = make_asset(owner_id);
+        let mut asset_b = make_asset(stranger_id);
+        // Give asset_b the same checksum as asset_a
+        asset_b.source_reference.checksum = asset_a.source_reference.checksum.clone();
+
+        let inner = Arc::new(InMemoryAssetRepository::new());
+        inner.save(&asset_a).await.unwrap();
+        inner.save(&asset_b).await.unwrap();
+
+        let share_repo = Arc::new(InMemoryShareRepository::new());
+
+        // Stranger queries by checksum — should only see their own
+        let filtered = VisibilityFilteredAssetRepository::new(
+            inner.clone(),
+            share_repo.clone(),
+            stranger_id,
+        );
+
+        let results = filtered
+            .find_by_checksum(&asset_a.source_reference.checksum)
+            .await
+            .unwrap();
+        assert_eq!(results.len(), 1);
+        assert_eq!(results[0].owner_user_id, stranger_id);
+    }
+
+    #[tokio::test]
+    async fn find_by_owner_skips_filter_for_own_assets() {
+        let owner_id = SystemId::new();
+        let asset = make_asset(owner_id);
+
+        let inner = Arc::new(InMemoryAssetRepository::new());
+        inner.save(&asset).await.unwrap();
+
+        let share_repo = Arc::new(InMemoryShareRepository::new());
+        let filtered = VisibilityFilteredAssetRepository::new(
+            inner.clone(),
+            share_repo.clone(),
+            owner_id,
+        );
+
+        let results = filtered.find_by_owner(&owner_id, 10, 0).await.unwrap();
+        assert_eq!(results.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn find_by_owner_filters_others_assets() {
+        let owner_id = SystemId::new();
+        let stranger_id = SystemId::new();
+        let asset = make_asset(owner_id);
+
+        let inner = Arc::new(InMemoryAssetRepository::new());
+        inner.save(&asset).await.unwrap();
+
+        let share_repo = Arc::new(InMemoryShareRepository::new());
+
+        // Stranger queries owner's assets without a share — should get nothing
+        let filtered = VisibilityFilteredAssetRepository::new(
+            inner.clone(),
+            share_repo.clone(),
+            stranger_id,
+        );
+
+        let results = filtered.find_by_owner(&owner_id, 10, 0).await.unwrap();
+        assert!(results.is_empty());
+    }
+}

crates/application/src/storage/commands/ingest_asset.rs

@@ -5,10 +5,7 @@ use domain::{
     },
     errors::DomainError,
     events::DomainEvent,
-    ports::{
-        AssetRepository, EventPublisher, FileStoragePort, IngestSessionRepository,
-        LibraryPathRepository, QuotaRepository, UsageLedgerRepository,
-    },
+    ports::{EventPublisher, FileStoragePort, IngestTransaction, LibraryPathRepository},
     value_objects::{Checksum, DateTimeStamp, SystemId},
 };
 use sha2::{Digest, Sha256};
@@ -26,31 +23,22 @@ pub struct IngestAssetCommand {
 }
 
 pub struct IngestAssetHandler {
-    ingest_repo: Arc<dyn IngestSessionRepository>,
+    tx: Arc<dyn IngestTransaction>,
     path_repo: Arc<dyn LibraryPathRepository>,
-    quota_repo: Arc<dyn QuotaRepository>,
-    ledger_repo: Arc<dyn UsageLedgerRepository>,
-    asset_repo: Arc<dyn AssetRepository>,
     file_storage: Arc<dyn FileStoragePort>,
     event_pub: Arc<dyn EventPublisher>,
 }
 
 impl IngestAssetHandler {
     pub fn new(
-        ingest_repo: Arc<dyn IngestSessionRepository>,
+        tx: Arc<dyn IngestTransaction>,
         path_repo: Arc<dyn LibraryPathRepository>,
-        quota_repo: Arc<dyn QuotaRepository>,
-        ledger_repo: Arc<dyn UsageLedgerRepository>,
-        asset_repo: Arc<dyn AssetRepository>,
         file_storage: Arc<dyn FileStoragePort>,
         event_pub: Arc<dyn EventPublisher>,
     ) -> Self {
         Self {
-            ingest_repo,
+            tx,
             path_repo,
-            quota_repo,
-            ledger_repo,
-            asset_repo,
             file_storage,
             event_pub,
         }
@@ -79,9 +67,9 @@ impl IngestAssetHandler {
             ));
         }
 
-        if let Some(quota) = self.quota_repo.find_by_owner(&cmd.uploader_id).await? {
+        if let Some(quota) = self.tx.find_quota(&cmd.uploader_id).await? {
             let current = self
-                .ledger_repo
+                .tx
                 .sum_usage(&cmd.uploader_id, UsageType::StorageBytes, None)
                 .await?;
             let result = domain::storage::services::check_quota(
@@ -131,10 +119,10 @@ impl IngestAssetHandler {
             cmd.uploader_id,
         );
 
-        self.asset_repo.save(&asset).await?;
+        self.tx.save_asset(&asset).await?;
 
         session.advance_to(IngestStatus::AwaitingProcessing)?;
-        self.ingest_repo.save(&session).await?;
+        self.tx.save_session(&session).await?;
 
         let entry = UsageLedgerEntry::new(
             cmd.uploader_id,
@@ -142,7 +130,7 @@ impl IngestAssetHandler {
             cmd.file_size,
             format!("Ingested {}", cmd.filename),
         );
-        self.ledger_repo.record(&entry).await?;
+        self.tx.record_usage(&entry).await?;
 
         self.event_pub
             .publish(&DomainEvent::AssetIngested {

crates/application/src/testing/repositories.rs

@@ -9,10 +9,10 @@ use domain::{
     errors::DomainError,
     ports::{
         AlbumRepository, AssetMetadataRepository, AssetRepository, DuplicateRepository,
-        GroupRepository, IngestSessionRepository, JobBatchRepository, JobRepository,
-        LibraryPathRepository, PipelineRepository, PluginRepository, QuotaRepository,
-        RoleRepository, ShareRepository, SidecarRepository, StorageVolumeRepository, TagRepository,
-        UsageLedgerRepository, UserRepository,
+        GroupRepository, IngestSessionRepository, IngestTransaction, JobBatchRepository,
+        JobRepository, LibraryPathRepository, PipelineRepository, PluginRepository,
+        QuotaRepository, RoleRepository, ShareRepository, SidecarRepository,
+        StorageVolumeRepository, TagRepository, UsageLedgerRepository, UserRepository,
     },
     value_objects::{Checksum, DateTimeStamp, Email, SystemId},
 };
@@ -1126,3 +1126,93 @@ impl PipelineRepository for InMemoryPipelineRepository {
         Ok(())
     }
 }
+
+// --- InMemoryIngestTransaction ---
+
+pub struct InMemoryIngestTransaction {
+    assets: Mutex<HashMap<String, Asset>>,
+    sessions: Mutex<HashMap<String, IngestSession>>,
+    quotas: Mutex<HashMap<String, QuotaDefinition>>,
+    ledger: Mutex<Vec<UsageLedgerEntry>>,
+}
+
+impl InMemoryIngestTransaction {
+    pub fn new() -> Self {
+        Self {
+            assets: Mutex::new(HashMap::new()),
+            sessions: Mutex::new(HashMap::new()),
+            quotas: Mutex::new(HashMap::new()),
+            ledger: Mutex::new(Vec::new()),
+        }
+    }
+
+    /// Pre-seed a quota for testing.
+    pub async fn insert_quota(&self, quota: &QuotaDefinition) {
+        self.quotas
+            .lock()
+            .await
+            .insert(quota.owner_scope.to_string(), quota.clone());
+    }
+}
+
+impl Default for InMemoryIngestTransaction {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl IngestTransaction for InMemoryIngestTransaction {
+    async fn save_asset(&self, asset: &Asset) -> Result<(), DomainError> {
+        self.assets
+            .lock()
+            .await
+            .insert(asset.asset_id.to_string(), asset.clone());
+        Ok(())
+    }
+
+    async fn save_session(&self, session: &IngestSession) -> Result<(), DomainError> {
+        self.sessions
+            .lock()
+            .await
+            .insert(session.session_id.to_string(), session.clone());
+        Ok(())
+    }
+
+    async fn find_quota(
+        &self,
+        owner_id: &SystemId,
+    ) -> Result<Option<QuotaDefinition>, DomainError> {
+        Ok(self
+            .quotas
+            .lock()
+            .await
+            .values()
+            .find(|q| &q.owner_scope == owner_id)
+            .cloned())
+    }
+
+    async fn sum_usage(
+        &self,
+        user_id: &SystemId,
+        usage_type: UsageType,
+        since: Option<DateTimeStamp>,
+    ) -> Result<u64, DomainError> {
+        let entries = self.ledger.lock().await;
+        let total = entries
+            .iter()
+            .filter(|e| &e.user_id == user_id && e.usage_type == usage_type)
+            .filter(|e| match &since {
+                Some(ts) => &e.timestamp >= ts,
+                None => true,
+            })
+            .map(|e| e.consumed_amount)
+            .sum();
+        Ok(total)
+    }
+
+    async fn record_usage(&self, entry: &UsageLedgerEntry) -> Result<(), DomainError> {
+        self.ledger.lock().await.push(entry.clone());
+        Ok(())
+    }
+}

crates/application/tests/catalog/queries/get_timeline.rs

@@ -31,6 +31,7 @@ async fn returns_paginated_assets() {
     let page = handler
         .execute(GetTimelineQuery {
             owner_id: owner,
+            caller_id: None,
             limit: 3,
             offset: 0,
         })
@@ -50,6 +51,7 @@ async fn returns_empty_for_no_assets() {
     let page = handler
         .execute(GetTimelineQuery {
             owner_id: SystemId::new(),
+            caller_id: None,
             limit: 10,
             offset: 0,
         })

crates/application/tests/storage/commands/ingest_asset.rs

@@ -3,23 +3,18 @@ use application::storage::{
     RegisterVolumeCommand, RegisterVolumeHandler,
 };
 use application::testing::{
-    InMemoryAssetRepository, InMemoryFileStorage, InMemoryIngestSessionRepository,
-    InMemoryLibraryPathRepository, InMemoryQuotaRepository, InMemoryStorageVolumeRepository,
-    InMemoryUsageLedgerRepository, StubEventPublisher,
+    InMemoryFileStorage, InMemoryIngestTransaction, InMemoryLibraryPathRepository,
+    InMemoryStorageVolumeRepository, StubEventPublisher,
 };
 use bytes::Bytes;
 use domain::entities::{IngestStatus, QuotaDefinition, TimePeriod, UsageType};
 use domain::errors::DomainError;
-use domain::ports::QuotaRepository;
 use domain::value_objects::SystemId;
 use std::sync::Arc;
 
 struct Harness {
-    ingest_repo: Arc<InMemoryIngestSessionRepository>,
+    tx: Arc<InMemoryIngestTransaction>,
     path_repo: Arc<InMemoryLibraryPathRepository>,
-    quota_repo: Arc<InMemoryQuotaRepository>,
-    ledger_repo: Arc<InMemoryUsageLedgerRepository>,
-    asset_repo: Arc<InMemoryAssetRepository>,
     file_storage: Arc<InMemoryFileStorage>,
     event_pub: Arc<StubEventPublisher>,
     vol_repo: Arc<InMemoryStorageVolumeRepository>,
@@ -28,11 +23,8 @@ struct Harness {
 impl Harness {
     fn new() -> Self {
         Self {
-            ingest_repo: Arc::new(InMemoryIngestSessionRepository::new()),
+            tx: Arc::new(InMemoryIngestTransaction::new()),
             path_repo: Arc::new(InMemoryLibraryPathRepository::new()),
-            quota_repo: Arc::new(InMemoryQuotaRepository::new()),
-            ledger_repo: Arc::new(InMemoryUsageLedgerRepository::new()),
-            asset_repo: Arc::new(InMemoryAssetRepository::new()),
             file_storage: Arc::new(InMemoryFileStorage::new()),
             event_pub: Arc::new(StubEventPublisher::new()),
             vol_repo: Arc::new(InMemoryStorageVolumeRepository::new()),
@@ -41,11 +33,8 @@ impl Harness {
 
     fn ingest_handler(&self) -> IngestAssetHandler {
         IngestAssetHandler::new(
-            self.ingest_repo.clone(),
+            self.tx.clone(),
             self.path_repo.clone(),
-            self.quota_repo.clone(),
-            self.ledger_repo.clone(),
-            self.asset_repo.clone(),
             self.file_storage.clone(),
             self.event_pub.clone(),
         )
@@ -111,7 +100,7 @@ async fn rejects_quota_exceeded() {
 
     let mut quota = QuotaDefinition::new(user);
     quota.add_rule(UsageType::StorageBytes, 500, TimePeriod::Lifetime);
-    h.quota_repo.save(&quota).await.unwrap();
+    h.tx.insert_quota(&quota).await;
 
     let handler = h.ingest_handler();
     let result = handler

crates/bootstrap/src/factory.rs

@@ -7,55 +7,12 @@ use tower_http::{
     trace::TraceLayer,
 };
 
-use adapters_auth::{BcryptPasswordHasher, JwtTokenIssuer};
-
-use adapters_postgres::{
-    PostgresAlbumRepository, PostgresAssetMetadataRepository, PostgresAssetRepository,
-    PostgresDuplicateRepository, PostgresIngestSessionRepository, PostgresJobBatchRepository,
-    PostgresJobRepository, PostgresLibraryPathRepository, PostgresPipelineRepository,
-    PostgresPluginRepository, PostgresQuotaRepository, PostgresShareRepository,
-    PostgresSidecarRepository, PostgresStorageVolumeRepository, PostgresTagRepository,
-    PostgresUsageLedgerRepository, PostgresUserRepository, PostgresVisibilityFilterRepository,
-    connect, run_migrations,
-};
-
+use adapters_postgres::{connect, run_migrations};
 use adapters_storage::LocalFileStorage;
-
-use application::{
-    catalog::{
-        GetAssetHandler, GetTimelineHandler, ReadAssetFileHandler, RegisterAssetHandler,
-        UpdateMetadataHandler,
-    },
-    identity::{GetProfileHandler, LoginUserHandler, RegisterUserHandler},
-    organization::{
-        CreateAlbumHandler, GetAlbumHandler, ManageAlbumEntriesHandler, TagAssetHandler,
-    },
-    processing::{
-        CompleteJobHandler, ConfigurePipelineHandler, EnqueueJobHandler, FailJobHandler,
-        ManagePluginHandler, ReportBatchProgressHandler, StartJobHandler,
-    },
-    sharing::{
-        AccessSharedResourceHandler, GenerateShareLinkHandler, RevokeShareHandler,
-        ShareResourceHandler,
-    },
-    sidecar::{
-        DetectExternalChangesHandler, ExportSidecarHandler, FullExportHandler, FullImportHandler,
-        ImportSidecarHandler, ResolveConflictHandler,
-    },
-    storage::{
-        CheckQuotaHandler, IngestAssetHandler, RegisterLibraryPathHandler, RegisterVolumeHandler,
-    },
-};
-use presentation::{
-    routes::app_router,
-    state::{
-        AppState, CatalogHandlers, IdentityHandlers, OrganizationHandlers, ProcessingHandlers,
-        SharingHandlers, SidecarHandlers, StorageHandlers,
-    },
-};
+use presentation::{routes::app_router, state::AppState};
 
 use crate::config::Config;
-use crate::log_sidecar_writer::LogSidecarWriter;
+use crate::services;
 
 pub async fn build_app(config: &Config) -> Result<Router> {
     let pool = connect(&config.database_url).await?;
@@ -64,222 +21,35 @@ pub async fn build_app(config: &Config) -> Result<Router> {
     let nats_client = async_nats::connect(&config.nats_url).await?;
     adapters_nats::ensure_stream(&nats_client).await?;
 
-    // Identity
-    let user_repo = Arc::new(PostgresUserRepository::new(pool.clone()));
-    let hasher = Arc::new(BcryptPasswordHasher);
-    let issuer = Arc::new(JwtTokenIssuer::new(&config.jwt_secret));
-
-    let register_handler = Arc::new(RegisterUserHandler::new(user_repo.clone(), hasher.clone()));
-    let login_handler = Arc::new(LoginUserHandler::new(
-        user_repo.clone(),
-        hasher,
-        issuer.clone(),
-    ));
-    let get_profile_handler = Arc::new(GetProfileHandler::new(user_repo));
-
-    // Repos
-    let album_repo = Arc::new(PostgresAlbumRepository::new(pool.clone()));
-    let asset_repo = Arc::new(PostgresAssetRepository::new(pool.clone()));
-    let metadata_repo = Arc::new(PostgresAssetMetadataRepository::new(pool.clone()));
-    let volume_repo = Arc::new(PostgresStorageVolumeRepository::new(pool.clone()));
-    let path_repo = Arc::new(PostgresLibraryPathRepository::new(pool.clone()));
-    let session_repo = Arc::new(PostgresIngestSessionRepository::new(pool.clone()));
-    let quota_repo = Arc::new(PostgresQuotaRepository::new(pool.clone()));
-    let ledger_repo = Arc::new(PostgresUsageLedgerRepository::new(pool.clone()));
-    let tag_repo = Arc::new(PostgresTagRepository::new(pool.clone()));
-    let duplicate_repo = Arc::new(PostgresDuplicateRepository::new(pool.clone()));
-    let sidecar_repo = Arc::new(PostgresSidecarRepository::new(pool.clone()));
-    let job_repo = Arc::new(PostgresJobRepository::new(pool.clone()));
-    let batch_repo = Arc::new(PostgresJobBatchRepository::new(pool.clone()));
-    let plugin_repo = Arc::new(PostgresPluginRepository::new(pool.clone()));
-    let pipeline_repo = Arc::new(PostgresPipelineRepository::new(pool.clone()));
     let transport = adapters_nats::NatsTransport::new(nats_client);
-    let event_publisher: Arc<dyn domain::ports::EventPublisher> =
+    let nats_publisher: Arc<dyn domain::ports::EventPublisher> =
         Arc::new(event_transport::EventPublisherAdapter::new(transport));
-    let sidecar_writer: Arc<LogSidecarWriter> = Arc::new(LogSidecarWriter);
+    let event_store: Arc<dyn domain::ports::EventStore> =
+        Arc::new(adapters_postgres::PostgresEventStore::new(pool.clone()));
+    let event_publisher: Arc<dyn domain::ports::EventPublisher> =
+        Arc::new(event_transport::CompositeEventPublisher::new(nats_publisher, event_store));
 
-    // File storage
     let storage_path = std::env::var("STORAGE_PATH").unwrap_or_else(|_| "./data/media".to_string());
     let file_storage: Arc<LocalFileStorage> = Arc::new(LocalFileStorage::new(&storage_path));
 
-    // Album handlers
-    let create_album_handler = Arc::new(CreateAlbumHandler::new(album_repo.clone()));
-    let get_album_handler = Arc::new(GetAlbumHandler::new(album_repo.clone()));
-    let manage_album_entries_handler = Arc::new(ManageAlbumEntriesHandler::new(album_repo));
-
-    // Asset handlers
-    let ingest_asset_handler = Arc::new(IngestAssetHandler::new(
-        session_repo,
-        path_repo.clone(),
-        quota_repo.clone(),
-        ledger_repo.clone(),
-        asset_repo.clone(),
-        file_storage.clone(),
-        event_publisher.clone(),
-    ));
-    let get_asset_handler = Arc::new(GetAssetHandler::new(
-        asset_repo.clone(),
-        metadata_repo.clone(),
-    ));
-    let get_timeline_handler = Arc::new(GetTimelineHandler::new(
-        asset_repo.clone(),
-        metadata_repo.clone(),
-    ));
-    let update_metadata_handler = Arc::new(UpdateMetadataHandler::new(
-        asset_repo.clone(),
-        metadata_repo.clone(),
-        event_publisher.clone(),
-    ));
-    let read_asset_file_handler =
-        Arc::new(ReadAssetFileHandler::new(asset_repo.clone(), file_storage));
-
-    // Register asset handler
-    let register_asset_handler = Arc::new(RegisterAssetHandler::new(
-        asset_repo.clone(),
-        duplicate_repo,
-        event_publisher.clone(),
-    ));
-
-    // Tag handler
-    let tag_asset_handler = Arc::new(TagAssetHandler::new(asset_repo.clone(), tag_repo));
-
-    // Check quota handler
-    let check_quota_handler = Arc::new(CheckQuotaHandler::new(quota_repo, ledger_repo));
-
-    // Sidecar handlers
-    let export_sidecar_handler = Arc::new(ExportSidecarHandler::new(
-        metadata_repo.clone(),
-        sidecar_repo.clone(),
-        sidecar_writer.clone(),
-    ));
-    let detect_changes_handler = Arc::new(DetectExternalChangesHandler::new(
-        sidecar_repo.clone(),
-        sidecar_writer.clone(),
-    ));
-    let import_sidecar_handler = Arc::new(ImportSidecarHandler::new(
-        sidecar_repo.clone(),
-        sidecar_writer.clone(),
-        metadata_repo.clone(),
-    ));
-    let resolve_conflict_handler = Arc::new(ResolveConflictHandler::new(
-        sidecar_repo.clone(),
-        sidecar_writer.clone(),
-        metadata_repo.clone(),
-    ));
-    let full_export_handler = Arc::new(FullExportHandler::new(
-        asset_repo.clone(),
-        metadata_repo.clone(),
-        sidecar_repo.clone(),
-        sidecar_writer.clone(),
-    ));
-    let full_import_handler = Arc::new(FullImportHandler::new(
-        asset_repo,
-        metadata_repo,
-        sidecar_repo,
-        sidecar_writer,
-    ));
-
-    // Processing handlers
-    let enqueue_job_handler = Arc::new(EnqueueJobHandler::new(
-        job_repo.clone(),
-        event_publisher.clone(),
-    ));
-    let start_job_handler = Arc::new(StartJobHandler::new(job_repo.clone()));
-    let complete_job_handler = Arc::new(CompleteJobHandler::new(
-        job_repo.clone(),
-        batch_repo.clone(),
-        event_publisher.clone(),
-    ));
-    let fail_job_handler = Arc::new(FailJobHandler::new(
-        job_repo.clone(),
-        batch_repo.clone(),
-        event_publisher.clone(),
-    ));
-    let batch_progress_handler = Arc::new(ReportBatchProgressHandler::new(batch_repo, job_repo));
-    let manage_plugin_handler = Arc::new(ManagePluginHandler::new(plugin_repo.clone()));
-    let configure_pipeline_handler =
-        Arc::new(ConfigurePipelineHandler::new(pipeline_repo, plugin_repo));
-
-    // Sharing repos & handlers
-    let share_repo = Arc::new(PostgresShareRepository::new(pool.clone()));
-    let _visibility_filter_repo = Arc::new(PostgresVisibilityFilterRepository::new(pool));
-
-    let share_resource_handler = Arc::new(ShareResourceHandler::new(
-        share_repo.clone(),
-        event_publisher.clone(),
-    ));
-    let generate_link_handler = Arc::new(GenerateShareLinkHandler::new(share_repo.clone()));
-    let revoke_handler = Arc::new(RevokeShareHandler::new(share_repo.clone(), event_publisher));
-    let access_handler = Arc::new(AccessSharedResourceHandler::new(share_repo));
-
-    // Storage handlers
-    let register_volume_handler = Arc::new(RegisterVolumeHandler::new(volume_repo.clone()));
-    let register_library_path_handler =
-        Arc::new(RegisterLibraryPathHandler::new(volume_repo, path_repo));
-
-    let identity = IdentityHandlers {
-        register: register_handler,
-        login: login_handler,
-        get_profile: get_profile_handler,
-    };
-
-    let catalog = CatalogHandlers {
-        ingest_asset: ingest_asset_handler,
-        get_asset: get_asset_handler,
-        get_timeline: get_timeline_handler,
-        update_metadata: update_metadata_handler,
-        read_asset_file: read_asset_file_handler,
-        register_asset: register_asset_handler,
-    };
-
-    let organization = OrganizationHandlers {
-        create_album: create_album_handler,
-        get_album: get_album_handler,
-        manage_album_entries: manage_album_entries_handler,
-        tag_asset: tag_asset_handler,
-    };
-
-    let storage_handlers = StorageHandlers {
-        register_volume: register_volume_handler,
-        register_library_path: register_library_path_handler,
-        check_quota: check_quota_handler,
-    };
-
-    let sidecar = SidecarHandlers {
-        export: export_sidecar_handler,
-        detect_changes: detect_changes_handler,
-        import: import_sidecar_handler,
-        resolve: resolve_conflict_handler,
-        full_export: full_export_handler,
-        full_import: full_import_handler,
-    };
-
-    let processing = ProcessingHandlers {
-        enqueue_job: enqueue_job_handler,
-        start_job: start_job_handler,
-        complete_job: complete_job_handler,
-        fail_job: fail_job_handler,
-        batch_progress: batch_progress_handler,
-        manage_plugin: manage_plugin_handler,
-        configure_pipeline: configure_pipeline_handler,
-    };
-
-    let sharing = SharingHandlers {
-        share_resource: share_resource_handler,
-        generate_link: generate_link_handler,
-        revoke: revoke_handler,
-        access: access_handler,
-    };
+    // Build per-context services
+    let identity = services::identity::build(&pool, &config.jwt_secret);
+    let (storage_repos, storage) = services::storage::build(&pool);
+    let catalog = services::catalog::build(&pool, &storage_repos, file_storage, event_publisher.clone());
+    let organization = services::organization::build(&pool);
+    let sidecar = services::sidecar::build(&pool);
+    let processing = services::processing::build(&pool, event_publisher.clone());
+    let sharing = services::sharing::build(&pool, event_publisher);
 
     let state = AppState {
-        identity,
+        identity: identity.handlers,
         catalog,
         organization,
-        storage: storage_handlers,
+        storage,
         sharing,
         sidecar,
         processing,
-        token_issuer: issuer,
+        token_issuer: identity.token_issuer,
     };
 
     let cors = CorsLayer::new()

crates/bootstrap/src/lib.rs

@@ -1,3 +1,4 @@
 pub mod config;
 pub mod factory;
 pub mod log_sidecar_writer;
+pub mod services;

crates/bootstrap/src/main.rs

@@ -4,6 +4,7 @@ use tracing::info;
 mod config;
 mod factory;
 mod log_sidecar_writer;
+mod services;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {

crates/bootstrap/src/services/catalog.rs

@@ -0,0 +1,68 @@
+use std::sync::Arc;
+
+use adapters_postgres::{
+    PgPool, PostgresAssetMetadataRepository, PostgresAssetRepository,
+    PostgresDuplicateRepository, PostgresIngestTransaction,
+};
+use adapters_storage::LocalFileStorage;
+use application::catalog::{
+    GetAssetHandler, GetTimelineHandler, ReadAssetFileHandler, RegisterAssetHandler,
+    UpdateMetadataHandler,
+};
+use application::storage::IngestAssetHandler;
+use domain::ports::EventPublisher;
+use presentation::state::CatalogHandlers;
+
+use super::storage::StorageRepos;
+
+pub fn build(
+    pool: &PgPool,
+    storage_repos: &StorageRepos,
+    file_storage: Arc<LocalFileStorage>,
+    event_publisher: Arc<dyn EventPublisher>,
+) -> CatalogHandlers {
+    let asset_repo = Arc::new(PostgresAssetRepository::new(pool.clone()));
+    let metadata_repo = Arc::new(PostgresAssetMetadataRepository::new(pool.clone()));
+    let duplicate_repo = Arc::new(PostgresDuplicateRepository::new(pool.clone()));
+    let ingest_tx = Arc::new(PostgresIngestTransaction::new(pool.clone()));
+
+    let ingest_asset = Arc::new(IngestAssetHandler::new(
+        ingest_tx,
+        storage_repos.path_repo.clone(),
+        file_storage.clone(),
+        event_publisher.clone(),
+    ));
+
+    let get_asset = Arc::new(GetAssetHandler::new(
+        asset_repo.clone(),
+        metadata_repo.clone(),
+    ));
+
+    let get_timeline = Arc::new(GetTimelineHandler::new(
+        asset_repo.clone(),
+        metadata_repo.clone(),
+    ));
+
+    let update_metadata = Arc::new(UpdateMetadataHandler::new(
+        asset_repo.clone(),
+        metadata_repo.clone(),
+        event_publisher.clone(),
+    ));
+
+    let read_asset_file = Arc::new(ReadAssetFileHandler::new(asset_repo.clone(), file_storage));
+
+    let register_asset = Arc::new(RegisterAssetHandler::new(
+        asset_repo,
+        duplicate_repo,
+        event_publisher,
+    ));
+
+    CatalogHandlers {
+        ingest_asset,
+        get_asset,
+        get_timeline,
+        update_metadata,
+        read_asset_file,
+        register_asset,
+    }
+}

crates/bootstrap/src/services/identity.rs

@@ -0,0 +1,35 @@
+use std::sync::Arc;
+
+use adapters_auth::{BcryptPasswordHasher, JwtTokenIssuer};
+use adapters_postgres::{PgPool, PostgresUserRepository};
+use application::identity::{GetProfileHandler, LoginUserHandler, RegisterUserHandler};
+use domain::ports::TokenIssuer;
+use presentation::state::IdentityHandlers;
+
+pub struct IdentityServices {
+    pub handlers: IdentityHandlers,
+    pub token_issuer: Arc<dyn TokenIssuer>,
+}
+
+pub fn build(pool: &PgPool, jwt_secret: &str) -> IdentityServices {
+    let user_repo = Arc::new(PostgresUserRepository::new(pool.clone()));
+    let hasher = Arc::new(BcryptPasswordHasher);
+    let issuer: Arc<JwtTokenIssuer> = Arc::new(JwtTokenIssuer::new(jwt_secret));
+
+    let register = Arc::new(RegisterUserHandler::new(user_repo.clone(), hasher.clone()));
+    let login = Arc::new(LoginUserHandler::new(
+        user_repo.clone(),
+        hasher,
+        issuer.clone(),
+    ));
+    let get_profile = Arc::new(GetProfileHandler::new(user_repo));
+
+    IdentityServices {
+        handlers: IdentityHandlers {
+            register,
+            login,
+            get_profile,
+        },
+        token_issuer: issuer,
+    }
+}

crates/bootstrap/src/services/mod.rs

@@ -0,0 +1,7 @@
+pub mod catalog;
+pub mod identity;
+pub mod organization;
+pub mod processing;
+pub mod sharing;
+pub mod sidecar;
+pub mod storage;

crates/bootstrap/src/services/organization.rs

@@ -0,0 +1,27 @@
+use std::sync::Arc;
+
+use adapters_postgres::{
+    PgPool, PostgresAlbumRepository, PostgresAssetRepository, PostgresTagRepository,
+};
+use application::organization::{
+    CreateAlbumHandler, GetAlbumHandler, ManageAlbumEntriesHandler, TagAssetHandler,
+};
+use presentation::state::OrganizationHandlers;
+
+pub fn build(pool: &PgPool) -> OrganizationHandlers {
+    let album_repo = Arc::new(PostgresAlbumRepository::new(pool.clone()));
+    let asset_repo = Arc::new(PostgresAssetRepository::new(pool.clone()));
+    let tag_repo = Arc::new(PostgresTagRepository::new(pool.clone()));
+
+    let create_album = Arc::new(CreateAlbumHandler::new(album_repo.clone()));
+    let get_album = Arc::new(GetAlbumHandler::new(album_repo.clone()));
+    let manage_album_entries = Arc::new(ManageAlbumEntriesHandler::new(album_repo));
+    let tag_asset = Arc::new(TagAssetHandler::new(asset_repo, tag_repo));
+
+    OrganizationHandlers {
+        create_album,
+        get_album,
+        manage_album_entries,
+        tag_asset,
+    }
+}

crates/bootstrap/src/services/processing.rs

@@ -0,0 +1,48 @@
+use std::sync::Arc;
+
+use adapters_postgres::{
+    PgPool, PostgresJobBatchRepository, PostgresJobRepository, PostgresPipelineRepository,
+    PostgresPluginRepository,
+};
+use application::processing::{
+    CompleteJobHandler, ConfigurePipelineHandler, EnqueueJobHandler, FailJobHandler,
+    ManagePluginHandler, ReportBatchProgressHandler, StartJobHandler,
+};
+use domain::ports::EventPublisher;
+use presentation::state::ProcessingHandlers;
+
+pub fn build(pool: &PgPool, event_publisher: Arc<dyn EventPublisher>) -> ProcessingHandlers {
+    let job_repo = Arc::new(PostgresJobRepository::new(pool.clone()));
+    let batch_repo = Arc::new(PostgresJobBatchRepository::new(pool.clone()));
+    let plugin_repo = Arc::new(PostgresPluginRepository::new(pool.clone()));
+    let pipeline_repo = Arc::new(PostgresPipelineRepository::new(pool.clone()));
+
+    let enqueue_job = Arc::new(EnqueueJobHandler::new(
+        job_repo.clone(),
+        event_publisher.clone(),
+    ));
+    let start_job = Arc::new(StartJobHandler::new(job_repo.clone()));
+    let complete_job = Arc::new(CompleteJobHandler::new(
+        job_repo.clone(),
+        batch_repo.clone(),
+        event_publisher.clone(),
+    ));
+    let fail_job = Arc::new(FailJobHandler::new(
+        job_repo.clone(),
+        batch_repo.clone(),
+        event_publisher,
+    ));
+    let batch_progress = Arc::new(ReportBatchProgressHandler::new(batch_repo, job_repo));
+    let manage_plugin = Arc::new(ManagePluginHandler::new(plugin_repo.clone()));
+    let configure_pipeline = Arc::new(ConfigurePipelineHandler::new(pipeline_repo, plugin_repo));
+
+    ProcessingHandlers {
+        enqueue_job,
+        start_job,
+        complete_job,
+        fail_job,
+        batch_progress,
+        manage_plugin,
+        configure_pipeline,
+    }
+}

crates/bootstrap/src/services/sharing.rs

@@ -0,0 +1,31 @@
+use std::sync::Arc;
+
+use adapters_postgres::{
+    PgPool, PostgresShareRepository, PostgresVisibilityFilterRepository,
+};
+use application::sharing::{
+    AccessSharedResourceHandler, GenerateShareLinkHandler, RevokeShareHandler,
+    ShareResourceHandler,
+};
+use domain::ports::EventPublisher;
+use presentation::state::SharingHandlers;
+
+pub fn build(pool: &PgPool, event_publisher: Arc<dyn EventPublisher>) -> SharingHandlers {
+    let share_repo = Arc::new(PostgresShareRepository::new(pool.clone()));
+    let _visibility_filter_repo = Arc::new(PostgresVisibilityFilterRepository::new(pool.clone()));
+
+    let share_resource = Arc::new(ShareResourceHandler::new(
+        share_repo.clone(),
+        event_publisher.clone(),
+    ));
+    let generate_link = Arc::new(GenerateShareLinkHandler::new(share_repo.clone()));
+    let revoke = Arc::new(RevokeShareHandler::new(share_repo.clone(), event_publisher));
+    let access = Arc::new(AccessSharedResourceHandler::new(share_repo));
+
+    SharingHandlers {
+        share_resource,
+        generate_link,
+        revoke,
+        access,
+    }
+}

crates/bootstrap/src/services/sidecar.rs

@@ -0,0 +1,65 @@
+use std::sync::Arc;
+
+use adapters_postgres::{
+    PgPool, PostgresAssetMetadataRepository, PostgresAssetRepository, PostgresSidecarRepository,
+};
+use application::sidecar::{
+    DetectExternalChangesHandler, ExportSidecarHandler, FullExportHandler, FullImportHandler,
+    ImportSidecarHandler, ResolveConflictHandler,
+};
+use presentation::state::SidecarHandlers;
+
+use crate::log_sidecar_writer::LogSidecarWriter;
+
+pub fn build(pool: &PgPool) -> SidecarHandlers {
+    let metadata_repo = Arc::new(PostgresAssetMetadataRepository::new(pool.clone()));
+    let asset_repo = Arc::new(PostgresAssetRepository::new(pool.clone()));
+    let sidecar_repo = Arc::new(PostgresSidecarRepository::new(pool.clone()));
+    let sidecar_writer: Arc<LogSidecarWriter> = Arc::new(LogSidecarWriter);
+
+    let export = Arc::new(ExportSidecarHandler::new(
+        metadata_repo.clone(),
+        sidecar_repo.clone(),
+        sidecar_writer.clone(),
+    ));
+
+    let detect_changes = Arc::new(DetectExternalChangesHandler::new(
+        sidecar_repo.clone(),
+        sidecar_writer.clone(),
+    ));
+
+    let import = Arc::new(ImportSidecarHandler::new(
+        sidecar_repo.clone(),
+        sidecar_writer.clone(),
+        metadata_repo.clone(),
+    ));
+
+    let resolve = Arc::new(ResolveConflictHandler::new(
+        sidecar_repo.clone(),
+        sidecar_writer.clone(),
+        metadata_repo.clone(),
+    ));
+
+    let full_export = Arc::new(FullExportHandler::new(
+        asset_repo.clone(),
+        metadata_repo.clone(),
+        sidecar_repo.clone(),
+        sidecar_writer.clone(),
+    ));
+
+    let full_import = Arc::new(FullImportHandler::new(
+        asset_repo,
+        metadata_repo,
+        sidecar_repo,
+        sidecar_writer,
+    ));
+
+    SidecarHandlers {
+        export,
+        detect_changes,
+        import,
+        resolve,
+        full_export,
+        full_import,
+    }
+}

crates/bootstrap/src/services/storage.rs

@@ -0,0 +1,35 @@
+use std::sync::Arc;
+
+use adapters_postgres::{
+    PgPool, PostgresLibraryPathRepository, PostgresQuotaRepository,
+    PostgresStorageVolumeRepository, PostgresUsageLedgerRepository,
+};
+use application::storage::{CheckQuotaHandler, RegisterLibraryPathHandler, RegisterVolumeHandler};
+use presentation::state::StorageHandlers;
+
+/// Shared storage repos needed by other bounded contexts (catalog ingest, etc.).
+pub struct StorageRepos {
+    pub path_repo: Arc<PostgresLibraryPathRepository>,
+}
+
+pub fn build(pool: &PgPool) -> (StorageRepos, StorageHandlers) {
+    let volume_repo = Arc::new(PostgresStorageVolumeRepository::new(pool.clone()));
+    let path_repo = Arc::new(PostgresLibraryPathRepository::new(pool.clone()));
+    let quota_repo = Arc::new(PostgresQuotaRepository::new(pool.clone()));
+    let ledger_repo = Arc::new(PostgresUsageLedgerRepository::new(pool.clone()));
+
+    let register_volume = Arc::new(RegisterVolumeHandler::new(volume_repo.clone()));
+    let register_library_path =
+        Arc::new(RegisterLibraryPathHandler::new(volume_repo, path_repo.clone()));
+    let check_quota = Arc::new(CheckQuotaHandler::new(quota_repo, ledger_repo));
+
+    let handlers = StorageHandlers {
+        register_volume,
+        register_library_path,
+        check_quota,
+    };
+
+    let repos = StorageRepos { path_repo };
+
+    (repos, handlers)
+}

crates/domain/src/common/ports.rs

@@ -1,5 +1,6 @@
 use crate::common::errors::DomainError;
 use crate::common::events::{DomainEvent, EventEnvelope};
+use crate::common::value_objects::SystemId;
 use async_trait::async_trait;
 use futures::stream::BoxStream;
 
@@ -11,3 +12,9 @@ pub trait EventPublisher: Send + Sync {
 pub trait EventConsumer: Send + Sync {
     fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>>;
 }
+
+#[async_trait]
+pub trait EventStore: Send + Sync {
+    async fn append(&self, event: &DomainEvent) -> Result<(), DomainError>;
+    async fn query_by_aggregate(&self, aggregate_id: &SystemId) -> Result<Vec<DomainEvent>, DomainError>;
+}

crates/domain/src/storage/ports.rs

@@ -1,6 +1,7 @@
 use super::entities::{
     IngestSession, LibraryPath, QuotaDefinition, StorageVolume, UsageLedgerEntry, UsageType,
 };
+use crate::catalog::entities::Asset;
 use crate::common::errors::DomainError;
 use crate::common::value_objects::{DateTimeStamp, SystemId};
 use async_trait::async_trait;
@@ -63,6 +64,25 @@ pub trait UsageLedgerRepository: Send + Sync {
     ) -> Result<u64, DomainError>;
 }
 
+// --- IngestTransaction ---
+
+/// Bundles the four persistence concerns that the ingest use-case touches
+/// (asset, session, quota, ledger) behind a single port so the handler only
+/// needs one `Arc<dyn IngestTransaction>` instead of four separate repos.
+#[async_trait]
+pub trait IngestTransaction: Send + Sync {
+    async fn save_asset(&self, asset: &Asset) -> Result<(), DomainError>;
+    async fn save_session(&self, session: &IngestSession) -> Result<(), DomainError>;
+    async fn find_quota(&self, owner_id: &SystemId) -> Result<Option<QuotaDefinition>, DomainError>;
+    async fn sum_usage(
+        &self,
+        user_id: &SystemId,
+        usage_type: UsageType,
+        since: Option<DateTimeStamp>,
+    ) -> Result<u64, DomainError>;
+    async fn record_usage(&self, entry: &UsageLedgerEntry) -> Result<(), DomainError>;
+}
+
 // --- FileStoragePort ---
 
 #[derive(Debug, Clone)]

crates/presentation/src/handlers/assets.rs

@@ -82,6 +82,7 @@ pub async fn timeline(
 ) -> Result<Json<TimelineResponse>, AppError> {
     let query = GetTimelineQuery {
         owner_id: claims.user_id,
+        caller_id: None,
         limit: params.limit.unwrap_or(DEFAULT_PAGE_SIZE).min(MAX_PAGE_SIZE),
         offset: params.offset.unwrap_or(0),
     };

crates/worker/Cargo.toml

@@ -17,6 +17,7 @@ adapters-nats      = { workspace = true }
 event-transport    = { workspace = true }
 async-nats         = { workspace = true }
 
+futures           = { workspace = true }
 tokio             = { workspace = true }
 anyhow            = { workspace = true }
 tracing           = { workspace = true }

crates/worker/src/config.rs

@@ -2,7 +2,7 @@
 pub struct WorkerConfig {
     pub database_url: String,
     pub nats_url: String,
-    pub poll_interval_secs: u64,
+    pub fallback_sweep_secs: u64,
     pub storage_path: String,
 }
 
@@ -12,10 +12,10 @@ impl WorkerConfig {
         Self {
             database_url: std::env::var("DATABASE_URL").expect("DATABASE_URL must be set"),
             nats_url: std::env::var("NATS_URL").unwrap_or_else(|_| "nats://localhost:4222".into()),
-            poll_interval_secs: std::env::var("POLL_INTERVAL_SECS")
+            fallback_sweep_secs: std::env::var("FALLBACK_SWEEP_SECS")
                 .ok()
                 .and_then(|v| v.parse().ok())
-                .unwrap_or(5),
+                .unwrap_or(60),
             storage_path: std::env::var("STORAGE_PATH").unwrap_or_else(|_| "./storage".into()),
         }
     }

crates/worker/src/main.rs

@@ -1,13 +1,18 @@
 use std::sync::Arc;
 use std::time::Duration;
-use tracing::{error, info};
+
+use futures::StreamExt;
+use tracing::{error, info, warn};
+
+use application::processing::ProcessNextJobCommand;
+use domain::events::DomainEvent;
+use domain::ports::EventConsumer;
 
 mod config;
 mod factories;
 mod plugin_registry;
 mod plugins;
 
-use application::processing::ProcessNextJobCommand;
 use factories::{Repos, build_plugin_registry, build_process_next_handler};
 
 #[tokio::main]
@@ -28,32 +33,90 @@ async fn main() -> anyhow::Result<()> {
     adapters_nats::ensure_stream(&nats_client).await?;
     info!(nats_url = %config.nats_url, "NATS connected");
 
+    let event_store: Arc<dyn domain::ports::EventStore> =
+        Arc::new(adapters_postgres::PostgresEventStore::new(pool.clone()));
     let repos = Repos::new(pool);
     let file_storage = Arc::new(adapters_storage::LocalFileStorage::new(
         &config.storage_path,
     ));
     let sidecar_writer: Arc<dyn domain::ports::SidecarWriterPort> = Arc::new(LogSidecarWriter);
 
-    let transport = adapters_nats::NatsTransport::new(nats_client);
+    // Publisher transport consumes a client clone; the consumer gets another.
+    let pub_transport = adapters_nats::NatsTransport::new(nats_client.clone());
+    let nats_publisher: Arc<dyn domain::ports::EventPublisher> =
+        Arc::new(event_transport::EventPublisherAdapter::new(pub_transport));
     let event_pub: Arc<dyn domain::ports::EventPublisher> =
-        Arc::new(event_transport::EventPublisherAdapter::new(transport));
+        Arc::new(event_transport::CompositeEventPublisher::new(nats_publisher, event_store));
 
     let registry = Arc::new(build_plugin_registry(&repos, file_storage, sidecar_writer));
-    let process_next = build_process_next_handler(&repos, registry, event_pub);
+    let process_next = Arc::new(build_process_next_handler(&repos, registry, event_pub));
 
-    let poll_interval = Duration::from_secs(config.poll_interval_secs);
-    info!(poll_secs = config.poll_interval_secs, "Worker running");
+    // ── Fallback sweep task ────────────────────────────────────────────
+    let sweep_interval = Duration::from_secs(config.fallback_sweep_secs);
+    let sweep_handler = Arc::clone(&process_next);
+    tokio::spawn(async move {
+        info!(every_secs = config.fallback_sweep_secs, "fallback sweep task started");
+        loop {
+            tokio::time::sleep(sweep_interval).await;
+            info!("fallback sweep: draining queued jobs");
+            loop {
+                match sweep_handler.execute(ProcessNextJobCommand).await {
+                    Ok(Some(job)) => {
+                        info!(job_id = %job.job_id, status = ?job.status, "sweep: processed job");
+                    }
+                    Ok(None) => break,
+                    Err(e) => {
+                        error!(error = %e, "sweep: error processing job");
+                        break;
+                    }
+                }
+            }
+        }
+    });
 
-    loop {
-        match process_next.execute(ProcessNextJobCommand).await {
-            Ok(Some(job)) => info!(job_id = %job.job_id, status = ?job.status, "processed job"),
-            Ok(None) => tokio::time::sleep(poll_interval).await,
+    // ── Event-driven loop via NATS ─────────────────────────────────────
+    let consumer_source = adapters_nats::NatsMessageSource::new(nats_client);
+    let event_consumer = event_transport::EventConsumerAdapter::new(consumer_source);
+
+    info!("event loop: listening for NATS events");
+    let mut stream = event_consumer.consume();
+
+    while let Some(result) = stream.next().await {
+        let envelope = match result {
+            Ok(env) => env,
             Err(e) => {
-                error!(error = %e, "worker error");
-                tokio::time::sleep(poll_interval).await;
+                error!(error = %e, "event loop: consumer error");
+                continue;
+            }
+        };
+
+        match &envelope.event {
+            DomainEvent::JobEnqueued {
+                job_id, job_type, ..
+            } => {
+                info!(job_id = %job_id, job_type = %job_type, "event loop: JobEnqueued received");
+                (envelope.ack)();
+                match process_next.execute(ProcessNextJobCommand).await {
+                    Ok(Some(job)) => {
+                        info!(job_id = %job.job_id, status = ?job.status, "event loop: processed job");
+                    }
+                    Ok(None) => {
+                        warn!("event loop: JobEnqueued event but no queued job found");
+                    }
+                    Err(e) => {
+                        error!(error = %e, "event loop: error processing job");
+                    }
+                }
+            }
+            other => {
+                info!(event = ?other, "event loop: non-job event, acking");
+                (envelope.ack)();
             }
         }
     }
+
+    error!("event loop: NATS stream ended unexpectedly");
+    Ok(())
 }
 
 struct LogSidecarWriter;