feat(job-registry): wire engine + wire trash cleaner

src/common/di.rs

@@ -39,6 +39,7 @@ use crate::infrastructure::repositories::pg::{
     FileBlobReadRepository, FileBlobWriteRepository, FileMetadataRepository, FolderDbRepository,
     TrashDbRepository,
 };
+use crate::infrastructure::scheduler::{JobRegistry, SchedulerEngine};
 use crate::infrastructure::services::file_content_cache::{
     FileContentCache, FileContentCacheConfig,
 };
@@ -429,6 +430,12 @@ impl AppServiceFactory {
         }
         let file_lifecycle = Arc::new(fls);
 
+        // Empty periodic-job registry; services register themselves
+        // downstream during their own creation. `SchedulerEngine::start`
+        // fires at the end of `build_app_state` once all registrations
+        // have landed.
+        let job_registry = Arc::new(JobRegistry::new());
+
         Ok(CoreServices {
             path_service,
             file_content_cache,
@@ -441,6 +448,7 @@ impl AppServiceFactory {
             dedup_service,
             zip_service: None, // Placeholder - replaced after app services init
             config: self.config.clone(),
+            job_registry,
         })
     }
 
@@ -865,14 +873,32 @@ impl AppServiceFactory {
         // Initialize cleanup service (bulk-deletes expired items in 2 SQL
         // queries, then GCs zero-reference blobs — including chunks orphaned
         // by aborted streaming uploads).
-        let cleanup_service = TrashCleanupService::new(
+        //
+        // Registers with the periodic-job scheduler
+        // (`docs/plan/job-registry.md` Part 1) instead of spawning its own
+        // tokio interval loop. `SchedulerEngine::start` fires the actual
+        // supervisor task at the end of `build_app_state`.
+        let cleanup_service = Arc::new(TrashCleanupService::new(
             trash_repo.clone(),
             core.dedup_service.clone(),
             24, // Run cleanup every 24 hours
-        );
-
-        cleanup_service.start_cleanup_job().await;
-        tracing::info!("Trash service initialized with daily cleanup schedule");
+        ));
+        let interval = cleanup_service.interval();
+        if let Err(e) = core
+            .job_registry
+            .register(cleanup_service.clone(), interval, None)
+            .await
+        {
+            // Duplicate registration is the only failure mode today and
+            // shouldn't happen in the normal DI flow. Log + continue so
+            // trash service still lands even if scheduling didn't.
+            tracing::error!("Failed to register trash_cleanup job with scheduler: {e}");
+        } else {
+            tracing::info!(
+                "Trash cleanup registered with scheduler (interval {} h)",
+                interval.as_secs() / 3600
+            );
+        }
 
         Some(service as Arc<TrashService>)
     }
@@ -1805,6 +1831,11 @@ impl AppServiceFactory {
                     50_000,
                 ),
             ),
+            // Populated below once every service has finished registering
+            // with `core.job_registry`. Starting the engine before all
+            // registrations land would race the first tick against
+            // late-registered jobs.
+            scheduler_engine: None,
         };
         let email_bundle = build_email_sender(&self.config.smtp);
         app_state.email_sender = email_bundle.sender;
@@ -2079,6 +2110,21 @@ impl AppServiceFactory {
             }
         }
 
+        // Start the periodic-job scheduler AFTER every native service has
+        // finished registering its jobs on `core.job_registry`. Starting
+        // it earlier would race the first tick against late registrations.
+        // See `docs/plan/job-registry.md` Part 1.
+        let registered = app_state.core.job_registry.len().await;
+        let engine = SchedulerEngine::start(app_state.core.job_registry.clone());
+        app_state.scheduler_engine = Some(Arc::new(engine));
+        tracing::info!(
+            target: "oxicloud::scheduler",
+            event = "scheduler.ready",
+            registered = registered,
+            "periodic scheduler ready ({} job(s) registered)",
+            registered
+        );
+
         Ok(app_state)
     }
 }
@@ -2099,6 +2145,11 @@ pub struct CoreServices {
     pub dedup_service: Arc<DedupService>,
     pub zip_service: Option<Arc<ZipService>>,
     pub config: AppConfig,
+    /// Periodic-job scheduler registry. Services that satisfy the
+    /// migration criterion (`docs/plan/job-registry.md`) `register()`
+    /// themselves here during their creation; `SchedulerEngine::start`
+    /// spins up the supervisor loop at the end of `build_app_state`.
+    pub job_registry: Arc<JobRegistry>,
 }
 
 /// Container for repository services
@@ -2305,6 +2356,14 @@ pub struct AppState {
     /// Authenticated callers bypass this limit.
     pub magic_link_send_per_ip_rate_limiter:
         Arc<crate::interfaces::middleware::rate_limit::RateLimiter>,
+    /// Handle to the periodic-job scheduler's supervisor task, spawned
+    /// at the end of `build_app_state` after every native service has
+    /// registered its jobs on `core.job_registry`. `Option` because
+    /// tests that assemble a partial `AppState` (no full DI) skip the
+    /// scheduler; production always populates it. Held here purely so
+    /// the tokio task isn't dropped — the supervisor loop runs off its
+    /// internal `JoinHandle`, not off this reference.
+    pub scheduler_engine: Option<Arc<SchedulerEngine>>,
 }
 
 // All AppState construction is done via struct literal in build_app_state().

src/infrastructure/services/trash_cleanup_service.rs

@@ -6,7 +6,9 @@ use tracing::{debug, error, info, instrument};
 use crate::common::errors::Result;
 use crate::domain::repositories::trash_repository::TrashRepository;
 use crate::infrastructure::repositories::pg::trash_db_repository::TrashDbRepository;
+use crate::infrastructure::scheduler::{JobHandler, JobOutcome};
 use crate::infrastructure::services::dedup_service::DedupService;
+use async_trait::async_trait;
 
 /// Service for automatic cleanup of expired items in the trash.
 ///
@@ -27,6 +29,8 @@ pub struct TrashCleanupService {
 }
 
 impl TrashCleanupService {
+    pub const JOB_NAME: &'static str = "trash_cleanup";
+
     pub fn new(
         trash_repository: Arc<TrashDbRepository>,
         dedup_service: Arc<DedupService>,
@@ -39,6 +43,12 @@ impl TrashCleanupService {
         }
     }
 
+    /// Registered interval as a `Duration` — helper for DI wiring so
+    /// the composition root doesn't reinvent the `hours × 3600` cast.
+    pub fn interval(&self) -> Duration {
+        Duration::from_secs(self.cleanup_interval_hours * 3600)
+    }
+
     /// Starts the periodic cleanup job
     #[instrument(skip(self))]
     pub async fn start_cleanup_job(&self) {
@@ -107,4 +117,69 @@ impl TrashCleanupService {
 
         Ok(())
     }
+
+    /// One-shot execution used by both the legacy `start_cleanup_job`
+    /// timer AND the new `JobHandler::run` path. Returns the counts a
+    /// caller can turn into either a log line (legacy) or a `JobOutcome`
+    /// (scheduler).
+    async fn run_once(&self) -> Result<TrashCleanupStats> {
+        let (files, folders) = self.trash_repository.delete_expired_bulk().await?;
+        // GC failure is non-fatal — the expiry itself succeeded. Report
+        // reclaimed bytes when possible; log + swallow otherwise.
+        let (gc_items, gc_bytes) = match self.dedup_service.garbage_collect().await {
+            Ok(pair) => pair,
+            Err(e) => {
+                error!("Trash cleanup GC failed: {:?}", e);
+                (0, 0)
+            }
+        };
+        Ok(TrashCleanupStats {
+            files_purged: files,
+            folders_purged: folders,
+            gc_items,
+            gc_bytes,
+        })
+    }
+}
+
+/// Structured counters for one trash-cleanup sweep. Consumed by the
+/// scheduler's `JobHandler::run` to shape `JobOutcome::Ok.extra`.
+#[derive(Debug, Clone, Copy)]
+struct TrashCleanupStats {
+    files_purged: u64,
+    folders_purged: u64,
+    gc_items: u64,
+    gc_bytes: u64,
+}
+
+#[async_trait]
+impl JobHandler for TrashCleanupService {
+    fn name(&self) -> &str {
+        Self::JOB_NAME
+    }
+
+    /// Runs one bulk-delete-expired + GC sweep. `count` on the returned
+    /// `JobOutcome::Ok` is the total number of rows this tick removed
+    /// from the trash (files + folders); `extra` carries GC reclaim
+    /// counts so operators can see "how much did this actually free."
+    ///
+    /// Failure of the trash sweep itself → `Err`. GC failure alone is
+    /// non-fatal and stays logged only.
+    async fn run(&self) -> JobOutcome {
+        match self.run_once().await {
+            Ok(stats) => {
+                let removed = stats.files_purged + stats.folders_purged;
+                JobOutcome::ok_with(
+                    removed,
+                    serde_json::json!({
+                        "files_purged": stats.files_purged,
+                        "folders_purged": stats.folders_purged,
+                        "gc_items": stats.gc_items,
+                        "gc_bytes": stats.gc_bytes,
+                    }),
+                )
+            }
+            Err(e) => JobOutcome::Err(format!("trash cleanup failed: {e}")),
+        }
+    }
 }