Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/xcoder-es/media-cleaner-pro/llms.txt

Use this file to discover all available pages before exploring further.

MediaCleaner Pro is structured as an event-driven hexagonal monolith. The intent is clean internal boundaries that allow each layer to evolve independently — or be split into separate services later — without sacrificing the performance and deployment simplicity of a single native binary. The domain logic (image classification, duplicate detection, quality ranking) lives in a pure-Rust crate with zero I/O dependencies. All filesystem, database, and network concerns are isolated in a separate adapter crate. The Axum API layer sits on top, wiring everything together with Tokio async channels and a bounded streaming pipeline.

​
Layers

┌─────────────────────────────────────────────────────────────────┐
│                      API Layer  (Axum)                          │
│       REST + OpenAPI 3.1  │  SSE Progress  │  Job Control       │
└────────────────────────────┬────────────────────────────────────┘
                             │
┌────────────────────────────▼────────────────────────────────────┐
│               Domain Core  (mc-core)                            │
│   Pure Rust — no I/O.  All 10 stage algorithms as pure fns.    │
│   Port traits (FileSystem, JobRepository, ImageHasher, …)       │
└────────────────────────────┬────────────────────────────────────┘
                             │
┌────────────────────────────▼────────────────────────────────────┐
│               Adapters  (mc-infra)                              │
│   Filesystem  │  SHA-256  │  dHash  │  SQLite  │  FileScanner   │
└─────────────────────────────────────────────────────────────────┘

API Layer

Axum 0.7 handles all HTTP concerns: REST JSON endpoints, OpenAPI spec generation via utoipa, and Server-Sent Events for real-time pipeline progress. No domain logic lives here.

Domain Core (mc-core)

A pure-Rust library crate. No tokio, no filesystem, no database. Declares all domain types and port traits as Rust traits, and implements all 10 pipeline stages as pure functions that take &ImageMetadata and return StageResult.

Adapters (mc-infra)

Concrete implementations of every port trait: tokio::fs filesystem, sha2-based exact hasher, dHash image hasher, rusqlite job repository, walkdir file scanner, and the in-process notification bus.

​
Hexagonal Port / Adapter Pattern

All cross-boundary communication flows through Rust traits defined in mc-core/src/ports.rs. The root binary and mc-infra depend on these traits; mc-core itself depends on nothing outside the standard library and a handful of pure data crates (serde, thiserror, chrono).

​
FileSystem (async)

Abstracts all file and directory operations. The concrete adapter in mc-infra uses tokio::fs for non-blocking I/O. 
#[async_trait]
pub trait FileSystem: Send + Sync {
    async fn list_dir(&self, path: &Path) -> Result<Vec<DirEntry>, DomainError>;
    async fn read_file(&self, path: &Path) -> Result<Vec<u8>, DomainError>;
    async fn write_file(&self, path: &Path, data: &[u8]) -> Result<(), DomainError>;
    async fn create_dir(&self, path: &Path) -> Result<(), DomainError>;
    async fn delete_file(&self, path: &Path) -> Result<(), DomainError>;
    async fn copy_file(&self, src: &Path, dest: &Path) -> Result<(), DomainError>;
    async fn move_file(&self, src: &Path, dest: &Path) -> Result<(), DomainError>;
    async fn canonicalize(&self, path: &Path) -> Result<String, DomainError>;
}

​
JobRepository

Persists job state. The mc-infra adapter stores jobs in SQLite via rusqlite (bundled). The trait is designed so the backing store can be replaced with Supabase Postgres without touching domain code. 
#[async_trait]
pub trait JobRepository: Send + Sync {
    async fn create_job(&self, job: &Job) -> Result<(), DomainError>;
    async fn get_job(&self, id: &JobId) -> Result<Option<Job>, DomainError>;
    async fn update_job(&self, job: &Job) -> Result<(), DomainError>;
    async fn list_jobs(&self, user_id: &UserId, limit: usize) -> Result<Vec<Job>, DomainError>;
    async fn delete_job(&self, id: &JobId) -> Result<(), DomainError>;
    async fn query_by_team(&self, team_id: &TeamId) -> Result<Vec<Job>, DomainError>;
}

​
FileScanner

Recursively collects image paths matching a list of extensions. 
#[async_trait]
pub trait FileScanner: Send + Sync {
    async fn scan(&self, path: &Path, extensions: &[&str]) -> Result<Vec<PathBuf>, DomainError>;
}

​
ImageHasher

Computes perceptual difference hashes and Hamming distances for near-duplicate detection. 
pub trait ImageHasher: Send + Sync {
    fn compute_dhash(&self, data: &[u8]) -> Result<u64, DomainError>;
    fn hamming_distance(&self, a: u64, b: u64) -> u32;
}

​
ExactHasher

Computes SHA-256 fingerprints for byte-perfect duplicate detection. 
pub trait ExactHasher: Send + Sync {
    fn compute_sha256(&self, data: &[u8]) -> Result<String, DomainError>;
}

​
ImageDecoder

Decodes raw image bytes into a typed ImageInfo struct (width, height, format). Implemented in mc-infra using the image crate. 
pub trait ImageDecoder: Send + Sync {
    fn decode(&self, data: &[u8]) -> Result<ImageInfo, DomainError>;
}

​
PipelineStage

The uniform interface that every one of the 10 processing stages implements. 
pub trait PipelineStage: Send + Sync {
    fn name(&self) -> &str;
    fn description(&self) -> &str;
    fn process(
        &self,
        meta: &ImageMetadata,
        context: &StageContext,
    ) -> Result<StageResult, DomainError>;
}
StageContext carries the current PipelineConfig, the set of already-seen SHA-256 hashes, and the list of duplicate paths accumulated so far — all pure in-memory data, no I/O.

​
NotificationBus

Broadcasts PipelineEvent values to any registered listeners. The in-process implementation fans events out to the SSE stream that clients subscribe to via GET /api/progress. 
pub trait NotificationBus: Send + Sync {
    fn broadcast(&self, event: &PipelineEvent);
}
PipelineEvent covers the full lifecycle: StageStarted, StageProgress, StageCompleted, JobCompleted, JobPaused, JobResumed, JobCancelled, Error, and Log.

​
Pipeline Concurrency Model

The streaming pipeline in src/pipeline/mod.rs is designed around backpressure: no phase can outrun the one downstream of it, and memory usage is bounded regardless of how many images are in the source directory. 
Phase 1: walkdir scan ──────────────▶ Vec<PathBuf>
                                           │
Phase 2: 16-slot Semaphore ────────────────▼
         concurrent Tokio reader pool      │
         (async tokio::fs::read)           │
                                   mpsc channel (cap 1000)
                                           │
Phase 3: rayon block_in_place ─────────────▼
         SHA-256 + dHash + decode          │
         (CPU-bound, off the async         │
          runtime thread pool)    mpsc channel (cap 1000)
                                           │
Phase 4: sequential stage loop ────────────▼
         all 10 stages per image           │
         exact dup → perceptual dup →      │
         tiny → icon → thumbnail →         │
         screenshot → wallpaper →          │
         document → AI class → quality     │
                                           ▼
Phase 5: file routing ─────────── move to output dirs
         (dest_dir / category / filename)
Key concurrency decisions:
  • Phase 2 uses tokio::sync::Semaphore::new(16) — at most 16 files are read concurrently from disk, preventing fd exhaustion on large directories.
  • Phase 3 calls tokio::task::block_in_place to execute SHA-256 and dHash computation on the current Tokio worker thread without yielding, keeping CPU-bound work off the async scheduler so it does not block other async tasks from making progress.
  • Both channels are bounded with a capacity of 1 000 messages. If the stage loop falls behind, the reader pool backs off automatically — the semaphore ensures the system never buffers more than 16 + 1000 images worth of raw bytes in memory at once.
  • CancellationToken (from tokio-util) propagates a cancel signal across all phases without requiring explicit Arc<AtomicBool> polling.

​
Job State Machine

Job lifecycle is managed by StateMachine in src/state/machine.rs. All transitions mutate AppState behind an RwLock and emit a structured log entry. 
           start_job
               │
               ▼
         ┌─────────────┐   start_stage(0)    ┌──────────────┐
         │   Running   │ ──────────────────▶ │ Stage Running│
         │             │                     │              │
         │  pause_job  │◀── pause_job ────── │ update_stage │
         │  ┌────────┐ │                     │   _progress  │
         │  │ Paused │ │                     │              │
         │  └────┬───┘ │                     │  complete_   │
         │       │     │◀── resume_job ──────│   stage(n)   │
         └───────┼─────┘                     └──────────────┘
                 │                                  │
            cancel_job                        complete_job
                 │                                  │
                 ▼                                  ▼
           Job stopped                        Job stopped
           (is_running = false)               (is_running = false)
The available transitions are:
MethodEffect
StateMachine::start_jobSets is_running = true, clears logs, resets all stage statuses to Pending
StateMachine::start_stage(idx)Marks stage idx as Running, records started_at timestamp
StateMachine::update_stage_progress(idx, processed, total)Updates progress percentage, recalculates speed and ETA
StateMachine::complete_stage(idx)Marks stage idx as Completed, sets progress to 100 %
StateMachine::fail_stage(idx, error)Marks stage as Failed, records the error string
StateMachine::complete_jobSets is_running = false
StateMachine::pause_jobSets is_paused = true
StateMachine::resume_jobSets is_paused = false
StateMachine::cancel_jobSets both is_running and is_paused to false; the CancellationToken signals the pipeline task to stop reading from the channel

​
OpenAPI Generation

Endpoint types, request bodies, and response schemas are annotated with utoipa derive macros directly on the Rust handler functions and domain structs. The spec is assembled at startup and served at /api/openapi.json as OpenAPI 3.1 JSON. 
// Example: utoipa derives the schema directly from the Rust type
#[derive(Serialize, Deserialize, utoipa::ToSchema)]
pub struct StartJobRequest {
    pub source_dir: String,
    pub dest_dir: String,
    pub hamming_threshold: u32,
}
Because the spec is generated from the same types the handlers compile against, it is always in sync with the running binary — there is no separate spec file to maintain. Frontend clients can generate typed API bindings from this endpoint using openapi-typescript or any OpenAPI codegen tool.

​
Temporal Integration

docker-compose.yml includes a Temporal server (temporalio/auto-setup:1.24.2) backed by PostgreSQL. When the TEMPORAL_HOST environment variable is set (e.g. TEMPORAL_HOST=temporal:7233), the Rust binary connects to it for durable workflow orchestration — job steps can survive process restarts and machine reboots. In standalone mode (no TEMPORAL_HOST), the binary runs the in-process streaming pipeline directly without any external dependency.
The src/temporal/ module contains three files — client.rs, workflows.rs, and activities.rs — that implement the Temporal worker interface. The Temporal UI (port 8080 in Docker Compose) provides a visual workflow history dashboard.
mc-core has no I/O dependencies whatsoever — it does not import tokio, rusqlite, or any crate that touches the filesystem. This means every pipeline stage can be unit-tested with plain cargo test by constructing an ImageMetadata value inline and asserting on the returned StageResult, with no test doubles, mock filesystems, or async runtimes required.

Build docs developers (and LLMs) love

Get started for freeTalk to us