From 31bb1245da6712f67a41b9b370e0d2f5ee5bef07 Mon Sep 17 00:00:00 2001 From: Pierre Date: Mon, 25 May 2026 23:48:42 +0200 Subject: [PATCH] improve doc and bug fix --- README.md | 1149 +++++++---------------- docs/fast-hash-global-id.md | 10 +- docs/plugin-system.md | 781 +-------------- docs/streaming-pipeline-architecture.md | 1011 +++++--------------- packages/async-utils/README.md | 2 +- 5 files changed, 600 insertions(+), 2353 deletions(-) diff --git a/README.md b/README.md index 2e15356..35ade65 100644 --- a/README.md +++ b/README.md @@ -1,865 +1,400 @@ # meta-sort -A standalone file sorting and metadata extraction service for the MetaMesh ecosystem. meta-sort monitors watch folders, extracts metadata from media files, stores it in a distributed key-value store, and manages remote drive mounting for shared access. +File-watcher and metadata-extraction service for the MetaMesh ecosystem. +meta-sort consumes file events from meta-core (over SSE), drives metadata +through a streaming pipeline + container plugins, and writes results back +to Redis through meta-core's API. It exposes a Fastify REST API and serves +the `/files` volume over WebDAV so that other services and plugin +containers can read media without mounting the same volume. + +> **Service role:** meta-sort is **Redis-consuming**, not Redis-managing. +> It does not run leader election, it does not spawn Redis, and it does +> not own `kv-leader.lock`. Those responsibilities belong to **meta-core** +> (a separate Go service). meta-sort reads +> `/meta-core/locks/kv-leader.info` via `LeaderClient` to discover the +> Redis URL, API URL, and WebDAV URLs, and registers its own URLs into +> `/meta-core/services/.json`. + +--- ## Overview -meta-sort is a **process-write** service that serves as the primary data ingestion point for the MetaMesh platform. It: +What this service does: -1. **Watches folders** for new media files (video, audio, subtitles) -2. **Extracts metadata** using FFmpeg and filename parsing -3. **Computes content identifiers** using midhash256 (fast) and SHA-256 (background) -4. **Stores metadata** in a distributed KV store (Redis or etcd) -5. **Manages remote mounts** (NFS, SMB, cloud storage) for shared access +1. **Subscribes** to meta-core's `/api/events/files` SSE stream (file + discovery events: `add`, `change`, `delete`, `rename`, `reset`). +2. **Drives the streaming pipeline** (`StreamingPipeline`): extension + filter → light phase (filename parse, basic stats, midhash256) → write + to Redis → dispatch container plugins → background phase (SHA-256). +3. **Manages container plugins**: spawns plugin containers via the Docker + socket, exposes a callback endpoint, lets plugins write metadata back + into Redis via the meta-core-compatible `/meta/*` API on this service. +4. **Serves files over WebDAV** at `/webdav/*` so container plugins and + sibling services (meta-fuse, meta-stremio) can read media without + needing the underlying bind/SMB/rclone mounts. + +What this service does **not** do (anymore): + +- It does not run leader election. +- It does not spawn or supervise Redis. +- It does not watch the filesystem directly (chokidar lives in meta-core). +- It does not manage remote mounts (mounts are now in meta-core / rclone-smb). +- It does not have an in-process plugin runtime — all plugins are + containerized. +- `/api/scan/trigger` and `/api/metadata/clear` no longer live here; use + meta-core's endpoints instead. --- -## Meta-Core Architecture +## Architecture -MetaMesh uses **two shared volumes** for operation: +### Two-volume layout -### 1. META_CORE_VOLUME - Infrastructure +meta-sort consumes two host-side volumes via standard bind mounts: -The infrastructure volume stores all service coordination data: +| Volume | Container path | Owner | Purpose | +|--------|----------------|-------|---------| +| `META_CORE_VOLUME` | `/meta-core` | meta-core writes, all services read | Leader-info file, Redis data, service registry, SSE cursor | +| `FILES_VOLUME` | `/files` | meta-sort serves over WebDAV | Watched media, plugin output, dynamic mounts | -| Context | Path | Description | -|---------|------|-------------| -| **Inside Container** | `/meta-core` | Mounted shared volume | -| **Production (Host)** | `/DATA/Apps/meta-core` | External shared drive mount point | -| **Development** | `./data/meta-core` | Local development directory | +File paths stored in Redis are **relative to `/files`**, which makes the +storage portable across hosts. See the root `CLAUDE.md` for the +authoritative volume layout description. + +### Leader discovery (read-only) ``` -META_CORE_VOLUME (/meta-core or /DATA/Apps/meta-core) -├── db/ # KV Database storage -│ ├── redis/ # Redis persistence (RDB + AOF) -│ │ ├── dump.rdb # Redis snapshot -│ │ └── appendonly.aof # Redis append-only log -│ └── etcd/ # etcd data (alternative) -│ └── member/ -│ -├── locks/ # Distributed locking -│ └── kv-leader.lock # Leader election lock file -│ # Contains: {"host":"service-name","api":"http://host:port"} -│ -├── services/ # Service discovery registry -│ ├── meta-sort.json # {"name":"meta-sort","api":"http://...","status":"running","pid":123} -│ ├── meta-fuse.json # {"name":"meta-fuse","api":"http://...","status":"running","pid":456} -│ ├── meta-stremio.json # {"name":"meta-stremio","api":"http://...","status":"running","pid":789} -│ └── meta-orbit.json # {"name":"meta-orbit","api":"http://...","status":"running","pid":321} -│ -└── config/ # Shared configuration - ├── remote-storage.conf # Remote mount definitions - └── smb-credentials.txt # SMB credentials (chmod 600) +meta-core (separate container) + ├── runs Redis + ├── runs file watcher (chokidar) and publishes file:events + ├── runs HTTP API on /api/urls, /api/events/files, /meta/*, /api/services, ... + └── writes /meta-core/locks/kv-leader.info + │ + ▼ + meta-sort (this service) + ├── LeaderClient reads kv-leader.info → Redis URL, API URL, + │ WebDAV URLs, etc. + ├── KVManager creates RedisKVClient (writes are routed + │ through meta-core's HTTP API for the api-mediated-access + │ model) + ├── ServiceRegistration writes + │ /meta-core/services/meta-sort-.json + └── FileEventConsumer subscribes to + ${apiUrl}/api/events/files (SSE) ``` -### 2. FILES_VOLUME - Shared Media Drive +Source: `packages/meta-sort-core/src/kv/{KVManager,LeaderClient,RedisClient,ServiceRegistration}.ts`, +`packages/meta-sort-core/src/events/{FileEventConsumer,SSEEventClient}.ts`. -The **FILES_VOLUME** (canonical name: `/files`) is the shared drive for all media files. This is where all actual media content resides, and file paths in the KV database reference this volume. - -| Context | Path | Description | -|---------|------|-------------| -| **Inside Container** | `/files` | Mounted shared media volume | -| **Production (Host)** | Varies (NFS/SMB/local) | External shared drive | -| **Development** | `./data/files` | Local development directory | +### Processing pipeline ``` -FILES_VOLUME (/files) -├── watch/ # Host media folder (read-only bind mount) -│ ├── Movies/ -│ │ └── Inception (2010)/ -│ │ └── Inception.mkv -│ └── TV Shows/ -│ └── Breaking Bad/ -│ -├── test/ # Test media folder (development) -│ -├── plugin/ # Plugin output files (read-write) -│ └── {plugin-id}/ # Per-plugin output directory -│ -└── corn/ # SMB/rclone mounts (mounted inside container) - ├── nas-share/ # SMB mount via rclone - └── gdrive/ # Google Drive mount via rclone +meta-core SSE ─► FileEventConsumer ─► StreamingPipeline + │ + ├─ validationQueue (extension filter) + │ + ├─ fastQueue + │ processLightPhase: + │ midhash256 (permanent ID) + │ basic metadata + │ write to Redis ───► file visible in VFS + │ dispatchAllPlugins (containers) + │ + └─ backgroundQueue + processHashPhase: + SHA-256, heavier hashes + await ContainerPluginScheduler + file:complete ───► DONE ``` -### WebDAV File Access - -meta-sort exposes `/files` via **WebDAV** at the `/webdav/` endpoint. This allows: -- **Container plugins** to access files without direct volume mounts -- **meta-fuse** and **meta-stremio** to read files from meta-sort -- Access to dynamic mounts (SMB, rclone) that can't be pre-mounted in other containers - -``` -Container Plugin ──► WebDAV (http://meta-sort/webdav/) ──► /files/ - GET, PUT, DELETE, MKCOL supported -``` - -Environment variable: `PLUGIN_WEBDAV_URL=http://meta-sort-dev/webdav` - -### File Path References - -All file paths stored in the KV database are **relative to FILES_VOLUME**: - -``` -KV Key: /file/{midhash256}/filePath -KV Value: "media1/Movies/Inception (2010)/Inception.mkv" - ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ - Relative to FILES_VOLUME (/files) - -Full path construction: - Container: /files + /media1/Movies/... = /files/media1/Movies/Inception (2010)/Inception.mkv - Any host: $FILES_VOLUME + /media1/Movies/... = $FILES_VOLUME/media1/Movies/Inception (2010)/Inception.mkv -``` - -This allows services to be deployed on different hosts while sharing the same volume. The FILES_VOLUME can be: -- A local folder bind mount -- An NFS share mounted on the host -- An SMB/CIFS share -- A FUSE mount (rclone, sshfs, etc.) - -### Mount Management - -meta-sort manages remote mounts **inside** the FILES_VOLUME for shared access: - -```bash -# meta-sort mounts remote storage INTO /files -/files/smb/nas-share ← SMB mount managed by meta-sort -/files/nfs/server ← NFS mount managed by meta-sort -/files/gdrive/ ← rclone mount managed by meta-sort - -# Other folders are bind mounts from host -/files/media1 ← Host folder mounted into container -/files/media2 ← Another host folder -``` - -All services access files through `/files/*`, regardless of whether the underlying storage is local, NFS, SMB, or cloud. - ---- - -## Leader Election and Lock Mechanism - -### How It Works - -MetaMesh uses a **filesystem-based distributed consensus** mechanism using `flock(2)`. The shared filesystem acts as the consensus layer. - -``` -┌─────────────────────────────────────────────────────────────────────────────┐ -│ Leader Election via flock │ -├─────────────────────────────────────────────────────────────────────────────┤ -│ │ -│ Step 1: Service Startup │ -│ ┌─────────────┐ │ -│ │ Service │ ──▶ Try to acquire flock on /meta-core/locks/kv-leader.lock│ -│ └─────────────┘ │ -│ │ │ -│ ▼ │ -│ ┌─────────────────────────────────────────────────────────────────────┐ │ -│ │ flock() Result │ │ -│ │ ┌─────────────────────┐ ┌──────────────────────────────────┐ │ │ -│ │ │ Lock Acquired │ │ Lock Blocked │ │ │ -│ │ │ ────────────── │ │ ───────────── │ │ │ -│ │ │ │ │ │ │ │ -│ │ │ BECOME LEADER │ │ BECOME FOLLOWER │ │ │ -│ │ │ 1. Spawn Redis │ │ 1. Read lock file │ │ │ -│ │ │ 2. Write API │ │ 2. Extract leader API │ │ │ -│ │ │ endpoint to │ │ 3. Connect to leader's │ │ │ -│ │ │ lock file │ │ Redis │ │ │ -│ │ │ 3. Accept clients │ │ 4. Start reconnect loop │ │ │ -│ │ └─────────────────────┘ └──────────────────────────────────┘ │ │ -│ └─────────────────────────────────────────────────────────────────────┘ │ -│ │ -└─────────────────────────────────────────────────────────────────────────────┘ -``` - -### Lock File Format - -The lock file at `/meta-core/locks/kv-leader.lock` contains JSON: - -```json -{ - "host": "meta-sort-instance-1", - "api": "redis://10.0.1.50:6379", - "http": "http://10.0.1.50:3000", - "timestamp": 1703808000000, - "pid": 12345 -} -``` - -### flock Implementation - -```typescript -import { open, flock } from 'fs'; -import { promisify } from 'util'; - -const LOCK_FILE = '/meta-core/locks/kv-leader.lock'; - -async function tryBecomeLeader(): Promise { - const fd = await promisify(open)(LOCK_FILE, 'w'); - - try { - // LOCK_EX = exclusive lock, LOCK_NB = non-blocking - await promisify(flock)(fd, 'exnb'); // Throws if cannot acquire - - // We are the leader - spawn KV database and write our endpoint - await spawnRedis(); - await fs.writeFile(LOCK_FILE, JSON.stringify({ - host: hostname(), - api: `redis://${myIP}:6379`, - http: `http://${myIP}:3000`, - timestamp: Date.now(), - pid: process.pid - })); - - return true; // We are leader - } catch (err) { - if (err.code === 'EAGAIN' || err.code === 'EWOULDBLOCK') { - // Lock held by another process - become follower - return false; - } - throw err; - } -} -``` - -### Failure Detection and Recovery - -``` -┌──────────────────────────────────────────────────────────────────────────┐ -│ Failure Detection Loop │ -├──────────────────────────────────────────────────────────────────────────┤ -│ │ -│ Follower Service: │ -│ ┌─────────────────────────────────────────────────────────────────────┐ │ -│ │ while (true) { │ │ -│ │ try { │ │ -│ │ await redis.ping(); // Health check │ │ -│ │ await sleep(5000); // Wait 5 seconds │ │ -│ │ } catch (connectionError) { │ │ -│ │ // Leader might be down │ │ -│ │ await tryBecomeLeader(); // Attempt to take over │ │ -│ │ if (isLeader) break; │ │ -│ │ await reconnectToNewLeader(); // Or connect to new leader │ │ -│ │ } │ │ -│ │ } │ │ -│ └─────────────────────────────────────────────────────────────────────┘ │ -│ │ -│ When Leader Dies: │ -│ 1. flock is automatically released by OS │ -│ 2. First follower to call flock() wins │ -│ 3. New leader spawns Redis, writes new endpoint │ -│ 4. Other followers detect connection failure, read new endpoint │ -│ │ -└──────────────────────────────────────────────────────────────────────────┘ -``` - ---- - -## Service Discovery - -### Registration Mechanism - -Each service registers itself in `/meta-core/services/{service-name}.json`: - -```json -{ - "name": "meta-sort", - "version": "1.0.0", - "api": "http://10.0.1.50:3000", - "status": "running", - "pid": 12345, - "hostname": "meta-sort-instance-1", - "startedAt": "2024-12-28T10:00:00Z", - "lastHeartbeat": "2024-12-28T12:30:00Z", - "capabilities": ["write", "mount", "monitor"], - "endpoints": { - "health": "/health", - "api": "/api", - "metrics": "/api/metrics" - } -} -``` - -### Service Lifecycle - -``` -┌──────────────────────────────────────────────────────────────────────────┐ -│ Service Lifecycle │ -├──────────────────────────────────────────────────────────────────────────┤ -│ │ -│ Startup: │ -│ 1. Create/update /meta-core/services/{name}.json │ -│ 2. Set status = "starting" │ -│ 3. Initialize (connect to KV, etc.) │ -│ 4. Set status = "running" │ -│ 5. Start heartbeat loop (update lastHeartbeat every 30s) │ -│ │ -│ Running: │ -│ - Heartbeat updates lastHeartbeat timestamp │ -│ - Other services read this file to discover endpoints │ -│ - stale detection: if (now - lastHeartbeat > 60s) → service dead │ -│ │ -│ Shutdown: │ -│ 1. Set status = "stopping" │ -│ 2. Graceful shutdown (finish current work) │ -│ 3. Delete /meta-core/services/{name}.json OR set status = "stopped" │ -│ │ -└──────────────────────────────────────────────────────────────────────────┘ -``` - -### Discovering Other Services - -```typescript -async function discoverService(name: string): Promise { - const servicePath = `/meta-core/services/${name}.json`; - - try { - const content = await fs.readFile(servicePath, 'utf-8'); - const service = JSON.parse(content); - - // Check if service is alive (heartbeat within last 60 seconds) - const lastHeartbeat = new Date(service.lastHeartbeat).getTime(); - const isAlive = Date.now() - lastHeartbeat < 60_000; - - if (!isAlive) { - console.warn(`Service ${name} appears stale (last heartbeat: ${service.lastHeartbeat})`); - return null; - } - - return service; - } catch (err) { - if (err.code === 'ENOENT') { - return null; // Service not registered - } - throw err; - } -} -``` - ---- - -## MetaMesh Ecosystem Architecture - -### Service Hierarchy - -``` -┌─────────────────────────────────────────────────────────────────────────────┐ -│ MetaMesh Ecosystem │ -│ │ -│ Shared Volumes: │ -│ ┌──────────────────────────────────┐ ┌──────────────────────────────────┐ │ -│ │ META_CORE_VOLUME │ │ FILES_VOLUME │ │ -│ │ (/meta-core) │ │ (/files) │ │ -│ │ ┌────────┐ ┌────────┐ ┌──────┐ │ │ ┌────────┐ ┌────────┐ ┌──────┐ │ │ -│ │ │ db/ │ │ locks/ │ │svc/ │ │ │ │media1/ │ │ smb/ │ │ nfs/ │ │ │ -│ │ │(Redis) │ │(flock) │ │(JSON)│ │ │ │(local) │ │(mount) │ │(mnt) │ │ │ -│ │ └────┬───┘ └───┬────┘ └──┬───┘ │ │ └───┬────┘ └───┬────┘ └──┬───┘ │ │ -│ └───────┼─────────┼─────────┼─────┘ └──────┼──────────┼─────────┼─────┘ │ -│ │ │ │ │ │ │ │ -│ ▼ ▼ ▼ ▼ ▼ ▼ │ -│ ┌───────────────────────────────────────────────────────────────────────┐│ -│ │ ││ -│ │ [1] meta-sort PROCESS-WRITE ││ -│ │ ├── Watches /files/* for new media files ││ -│ │ ├── Extracts metadata → writes to KV (Redis in /meta-core/db) ││ -│ │ ├── Manages remote mounts INTO /files (NFS, SMB, rclone) ││ -│ │ └── Leader election (spawns Redis if leader) ││ -│ │ ││ -│ │ [2] meta-fuse PROCESS-READ ││ -│ │ ├── Reads metadata from KV (Redis) ││ -│ │ ├── Builds virtual filesystem from metadata ││ -│ │ ├── Serves FUSE mount + WebDAV ││ -│ │ └── File paths resolve to /files/* ││ -│ │ ││ -│ │ [3] meta-stremio PROCESS-READ ││ -│ │ ├── Reads metadata from KV (Redis) ││ -│ │ ├── Stremio addon server (manifest, catalog, streams) ││ -│ │ ├── HLS transcoding via FFmpeg ││ -│ │ └── Video files from /files/* ││ -│ │ ││ -│ │ [4] meta-orbit SHARING-READ-WRITE ││ -│ │ ├── P2P metadata synchronization (libp2p + OrbitDB) ││ -│ │ ├── Decentralized search across network ││ -│ │ ├── Reads from local KV, shares to network ││ -│ │ └── Receives from network, writes to local KV ││ -│ │ ││ -│ └───────────────────────────────────────────────────────────────────────┘│ -│ │ -└─────────────────────────────────────────────────────────────────────────────┘ -``` - -### Data Flow - -``` -┌─────────────────────────────────────────────────────────────────────────────┐ -│ Data Flow │ -├─────────────────────────────────────────────────────────────────────────────┤ -│ │ -│ File Discovery: │ -│ │ -│ /files/media1/Movies/Inception (2010)/Inception.mkv │ -│ │ │ -│ ▼ │ -│ [meta-sort] ──▶ Extract metadata ──▶ Write to Redis (in /meta-core/db) │ -│ │ │ │ -│ │ Key: file:{midhash256} │ │ -│ │ filePath: "media1/Movies/Inception (2010)/Inception.mkv" │ -│ │ title: "Inception" │ -│ │ year: "2010" │ -│ │ duration: 8880 │ -│ │ │ │ -│ ▼ ▼ │ -│ │ -│ File Access: │ -│ │ -│ [meta-fuse] ◀── Read from Redis │ -│ │ │ -│ ▼ │ -│ Virtual FS: /mnt/virtual/Movies/Inception (2010)/Inception.mkv │ -│ │ ↓ │ -│ │ symlink to /files/media1/Movies/Inception (2010)/... │ -│ │ │ -│ [meta-stremio] ◀── Read from Redis │ -│ │ │ -│ ▼ │ -│ Stream: GET /stremio/stream/{midhash256}/playlist.m3u8 │ -│ │ ↓ │ -│ │ transcode /files/media1/Movies/Inception (2010)/... │ -│ │ │ -│ [meta-orbit] ◀──▶ Sync with P2P network │ -│ │ │ -│ ▼ │ -│ Share metadata to peers / Receive metadata from peers │ -│ │ -└─────────────────────────────────────────────────────────────────────────────┘ -``` - ---- - -## Deployment - -### Docker Compose (Production) - -```yaml -version: '3.8' - -services: - meta-sort: - image: metazla/meta-sort:latest - volumes: - # Infrastructure volume (KV DB, locks, services) - - /DATA/Apps/meta-core:/meta-core:rw - # Shared media volume - - /DATA/Apps/meta-core/files:/files:rw,shared - # OR mount multiple sources into /files - # - /mnt/nas/media:/files/nas:ro - # - /mnt/local/media:/files/local:ro - environment: - - META_CORE_PATH=/meta-core - - FILES_PATH=/files - - WATCH_PATHS=/files - - REDIS_DATA_DIR=/meta-core/db/redis - privileged: true # Required for rclone FUSE mounts - - meta-fuse: - image: metazla/meta-fuse:latest - volumes: - # Infrastructure (read-only for followers) - - /DATA/Apps/meta-core:/meta-core:ro - # Shared media volume (read-only) - - /DATA/Apps/meta-core/files:/files:ro - # FUSE mount output - - /mnt/metamesh:/mnt/metamesh:rw,shared - environment: - - META_CORE_PATH=/meta-core - - FILES_PATH=/files - privileged: true # Required for FUSE - cap_add: - - SYS_ADMIN - devices: - - /dev/fuse - - meta-stremio: - image: metazla/meta-stremio:latest - volumes: - # Infrastructure (read-only) - - /DATA/Apps/meta-core:/meta-core:ro - # Shared media volume (read-only) - - /DATA/Apps/meta-core/files:/files:ro - environment: - - META_CORE_PATH=/meta-core - - FILES_PATH=/files - ports: - - "7000:7000" - - meta-orbit: - image: metazla/meta-orbit:latest - volumes: - # Infrastructure (read-write for P2P sync) - - /DATA/Apps/meta-core:/meta-core:rw - environment: - - META_CORE_PATH=/meta-core - ports: - - "4001:4001" # libp2p -``` - -### Environment Variables - -| Variable | Default | Description | -|----------|---------|-------------| -| `META_CORE_PATH` | `/meta-core` | Path to META_CORE_VOLUME (infrastructure) | -| `FILES_PATH` | `/files` | Path to FILES_VOLUME (shared media) | -| `REDIS_URL` | Auto-discover | Redis connection (from leader lock file) | -| `SERVICE_NAME` | Container hostname | Unique service identifier | -| `BASE_URL` | `http://localhost:8180` | External URL for this service | - -**Container Plugins:** - -| Variable | Default | Description | -|----------|---------|-------------| -| `CONTAINER_PLUGINS_CONFIG` | `/app/plugins.yml` | Path to plugins configuration | -| `DOCKER_SOCKET_PATH` | `/var/run/docker.sock` | Docker socket for container management | -| `CONTAINER_CALLBACK_URL` | `http://meta-sort:80` | URL plugins use to call back | -| `CONTAINER_NETWORK` | `meta-network` | Docker network for plugin containers | -| `PLUGIN_WEBDAV_URL` | - | WebDAV URL for plugin file access (e.g., `http://meta-sort-dev/webdav`) | -| `PLUGIN_STACK_NAME` | - | Docker Compose project name for grouping containers | - ---- - -## File Processing Pipeline - -meta-sort uses a **plugin-based architecture** with a two-queue task scheduler: - -``` -┌──────────────┐ ┌────────────────────────────────────────────────────┐ -│ Discovery │ │ Task Scheduler │ -│ (Stream) │───▶│ ┌─────────────────┐ ┌──────────────────────────┐ │ -└──────────────┘ │ │ Fast Queue │ │ Background Queue │ │ - Watch folder │ │ (16 workers) │ │ (4 workers) │ │ - + manual scan │ │ <1s tasks │ │ >=1s tasks │ │ - │ └─────────────────┘ └──────────────────────────┘ │ - └────────────────────────────────────────────────────┘ -``` - -Tasks are automatically classified and scheduled based on plugin dependencies: - -``` -Discovery → file-info → ffmpeg → filename-parser → jellyfin-nfo → tmdb - ↓ - anime-detector ← language ← subtitle ← torrent ← full-hash -``` - ---- - -## Plugin System - -Plugins are the core processing units that extract and enrich metadata. Each plugin: -- Has a `manifest.yml` defining metadata schema, config, and dependencies -- Runs in either **fast** (high concurrency) or **background** (low concurrency) queue -- Can depend on other plugins to ensure correct execution order - -### Container Plugins - -All plugins run as **Docker containers** for isolation and flexibility. Plugins are configured in `plugins.yml` and communicate via HTTP API. - -| Plugin | Language | Queue | Priority | Description | -|--------|----------|-------|----------|-------------| -| **file-info** | TypeScript | fast | 10 | Basic file info (type, MIME, size, midhash256) | -| **ffmpeg** | TypeScript | fast | 15 | Video/audio/subtitle stream metadata via FFprobe | -| **filename-parser** | TypeScript | fast | 20 | Parse title, season, episode, year, quality from filename | -| **jellyfin-nfo** | TypeScript | fast | 25 | Parse Jellyfin/Kodi NFO metadata files | -| **tmdb** | TypeScript | background | 30 | Fetch metadata from The Movie Database API | -| **anime-detector** | TypeScript | fast | 35 | Detect anime content from keywords and audio tracks | -| **language** | TypeScript | fast | 40 | Aggregate languages from all streams (ISO 639-3) | -| **subtitle** | TypeScript | fast | 45 | Find sibling subtitle files and detect language | -| **torrent** | TypeScript | fast | 50 | Parse .torrent file metadata | -| **full-hash** | TypeScript | background | 100 | Compute SHA-256/SHA-1/MD5 hashes (JS implementation) | -| **fast-full-hash** | **Rust** | background | 100 | High-performance multi-algorithm hashing (Rust/Axum) | - -### Plugin File Access - -Plugins access files via **WebDAV** served by meta-sort nginx. This allows plugins to: -- Access files without requiring direct volume mounts -- See SMB/rclone mounts that are mounted dynamically inside meta-sort -- Read and write to the plugin output directory at `/files/plugin/` - -### Plugin Manifest Example - -```yaml -id: ffmpeg -name: FFmpeg Metadata -version: 1.0.0 -description: Extracts video/audio/subtitle stream metadata using FFprobe -priority: 15 -color: "#4CAF50" -defaultQueue: fast - -dependencies: - - file-info - -config: {} - -schema: - fileinfo/duration: - label: Duration - type: string - readonly: true - fileinfo/streamdetails/video: - label: Video Streams - type: json - readonly: true - pattern: "fileinfo/streamdetails/video/{n}/*" -``` - -### Task Queue Architecture - -- **Fast Queue** (16 workers): High-throughput for quick operations (<1s) - - File info extraction, FFprobe, filename parsing, NFO reading -- **Background Queue** (4 workers): Low-concurrency for slow operations (>=1s) - - TMDB API calls, full SHA-256 hashing - - Automatically pauses when fast queue has work, resumes when idle - -### Plugin Management - -Plugins can be enabled/disabled and configured via: -- **Dashboard**: https://dev.localhost:8180/ → Plugins tab -- **Config file**: `plugins.json` in cache directory - ---- - -## Processing States - -1. **Pending** - File discovered in watch folder -2. **Processing** - Plugins running (fast queue first, then background) -3. **Done** - All plugins completed successfully - ---- - -## Package Structure - -``` -meta-sort/ -├── packages/ -│ ├── meta-sort-core/ # Core processing service (@meta-sort/core) -│ │ ├── plugins/ # Built-in plugins -│ │ │ ├── file-info/ # Basic file information -│ │ │ ├── ffmpeg/ # FFprobe metadata extraction -│ │ │ ├── filename-parser/# Filename parsing -│ │ │ ├── jellyfin-nfo/ # NFO file parsing -│ │ │ ├── tmdb/ # TMDB API integration -│ │ │ ├── anime-detector/ # Anime content detection -│ │ │ ├── language/ # Language aggregation -│ │ │ ├── subtitle/ # Subtitle file finder -│ │ │ ├── torrent/ # Torrent file parsing -│ │ │ └── full-hash/ # SHA-256 full file hashing -│ │ ├── src/ -│ │ │ ├── api/ # Fastify REST API (UnifiedAPIServer) -│ │ │ ├── config/ # Environment configuration -│ │ │ ├── jellyfin/ # Jellyfin integration -│ │ │ ├── kv/ # KV manager with leader election (Redis) -│ │ │ ├── logic/ # File processing logic -│ │ │ ├── metrics/ # Performance tracking -│ │ │ ├── plugin-engine/ # Plugin loader and task scheduler -│ │ │ ├── types/ # TypeScript types -│ │ │ ├── utils/ # Utility functions -│ │ │ └── index.ts # Entry point -│ │ └── package.json -│ │ -│ ├── meta-sort-ui/ # Monitoring dashboard (@meta-sort/ui) -│ │ └── ... # React/Vite app -│ │ -│ └── async-utils/ # Async utility library -│ └── ... # MultiQueue, PromiseQueue, etc. -│ -├── docker/ -│ ├── nginx.conf # Reverse proxy config -│ ├── redis.conf # Redis configuration -│ └── supervisord.conf # Process management -│ -├── Dockerfile -├── docker-compose.yml -├── package.json # Workspace root -├── pnpm-workspace.yaml -└── README.md -``` - -### Key Components (meta-sort-core) - -| Component | File | Purpose | -|-----------|------|---------| -| Entry Point | `src/index.ts` | Initializes KV manager, plugin engine, API server, file watching | -| File Processor | `src/logic/WatchedFileProcessor.ts` | File discovery and plugin task scheduling | -| Plugin Loader | `src/plugin-engine/PluginLoader.ts` | Loads plugins from manifest.yml, manages plugin lifecycle | -| Task Scheduler | `src/plugin-engine/TaskScheduler.ts` | Two-queue scheduler (fast/background) with dependency resolution | -| KV Manager | `src/kv/KVManager.ts` | Leader election via flock, Redis spawning, service discovery | -| Unified API | `src/api/UnifiedAPIServer.ts` | Fastify REST API for monitoring and control | -| Performance Metrics | `src/metrics/PerformanceMetrics.ts` | Tracks processing throughput and timing | -| Config | `src/config/EnvConfig.ts` | Environment variable parsing | - ---- - -## Core Features - -### 1. Folder Watching - -- **Real-time monitoring** using chokidar for file system events -- **Streaming discovery** via async generators for efficient memory usage -- **Configurable watch paths** supporting local and remote mounted directories -- **File type filtering** (video, audio, subtitle, torrent files) - -### 2. Plugin-Based Metadata Extraction - -Metadata extraction is handled by plugins that can be enabled/disabled and configured: - -- **file-info** - Basic file info, MIME type, midhash256 content identifier -- **ffmpeg** - Duration, codec, resolution, bitrate, audio tracks, subtitles via FFprobe -- **filename-parser** - Extract title, year, season, episode, quality from filenames -- **jellyfin-nfo** - Parse NFO sidecar files (Jellyfin/Kodi format) -- **tmdb** - Fetch rich metadata from The Movie Database API -- **anime-detector** - Detect anime content from keywords and audio tracks -- **language** - Aggregate languages from all streams (ISO 639-3) -- **subtitle** - Find sibling subtitle files and detect language -- **torrent** - Parse .torrent file metadata -- **full-hash** - Compute SHA-256 hash of entire file (background) - -### 3. KV Storage - -- **Nested key architecture** for property-level granularity -- **Leader election** via flock for high availability -- **Service discovery** via shared filesystem -- **CRDT-compatible** for distributed synchronization - -``` -Key Structure: -/file/{midhash256}/title → "Inception" -/file/{midhash256}/year → "2010" -/file/{midhash256}/video/codec → "h265" -/file/{midhash256}/audio/0/language → "en" -/file/{midhash256}/filePath → "media1/Movies/Inception (2010)/Inception.mkv" - ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ - Relative to FILES_VOLUME (/files) -``` - -### 4. Remote Mount Management - -- **NFS shares** - Network file systems -- **SMB/CIFS** - Windows network shares with credentials -- **rclone** - Cloud storage (Google Drive, Dropbox, S3, etc.) -- **Automatic mounting** on startup -- **Mounts created inside FILES_VOLUME** for shared access by all services - -Mounts are created inside `/files/` (FILES_VOLUME) so all services can access them: - -```bash -# Example mount structure -/files/ # FILES_VOLUME -├── media1/ # Local folder bind mount from host -├── media2/ # Another local folder -├── smb/ # SMB/CIFS remote mounts -│ └── nas-share/ # SMB: //server/share -├── nfs/ # NFS remote mounts -│ └── server-media/ # NFS: nas.local:/volume1/media -└── gdrive/ # rclone: gdrive:Media -``` +`midhash256` is the **permanent** file identifier — there is no `tempId` +and no rename step when SHA-256 finishes. Full details in +[`docs/streaming-pipeline-architecture.md`](docs/streaming-pipeline-architecture.md). + +### Container plugins + +All plugins run as separate Docker containers. meta-sort orchestrates +them via the Docker socket and a callback endpoint: + +- **Built-in plugins** (each is a separate git submodule under + `packages/plugins/metamesh-plugin-*` at the repo root): `file-info`, + `ffmpeg`, `filename-parser`, `jellyfin-nfo`, `tmdb`, `anime-detector`, + `language`, `subtitle`, `subtitle-extractor`, `torrent`, `fullhash`. +- **Dispatch:** `ContainerPluginScheduler.dispatchAllPlugins(hashId, + filePath, metaFlat)` — fire-and-forget. The scheduler keeps its own + fast / background queues based on each plugin manifest's + `defaultQueue`. +- **Plugin → meta-sort:** plugins POST results to + `/api/plugins/callback`, and they can read/write Redis metadata + through the meta-core-compatible `/meta/:hash` routes that meta-sort + also exposes (so plugins only need one HTTP target). +- **Plugins access files** by hitting meta-sort's WebDAV + (`http://metasort-app/webdav/...` inside the docker network). + +For the plugin HTTP contract, see +[`docs/containerized-plugin-architecture.md`](docs/containerized-plugin-architecture.md) +and [`docs/plugin-task-queue-architecture.md`](docs/plugin-task-queue-architecture.md). --- ## Configuration -### Environment Variables +### Environment variables -```bash -# Volume paths -META_CORE_PATH=/meta-core # Infrastructure volume -FILES_PATH=/files # Shared media volume +| Variable | Default | Description | +|----------|---------|-------------| +| `META_CORE_PATH` | `/meta-core` | Mount of META_CORE_VOLUME. Required — startup fails fast if absent. | +| `FILES_PATH` | `/files` | Mount of FILES_VOLUME. | +| `CACHE_FOLDER_PATH` | `/data/cache` | Local cache root. | +| `SERVICE_NAME` | `meta-sort` | Identifier used for service registration. | +| `SERVICE_VERSION` | `1.0.0` | Recorded in the service registry. | +| `FUSE_API_PORT` | `3000` | HTTP API port (legacy name; this is the Fastify API). | +| `FUSE_API_HOST` | `0.0.0.0` | API bind host. | +| `BASE_URL` | — | External base URL recorded in service registration. | +| `ADVERTISE_HOST` | — | Hostname to advertise to other services. Falls back to auto-detected IP. | +| `MAX_WORKER_THREADS` | `os.cpus().length` | If set, overrides all pipeline concurrencies to `(M*2, M, M)`. | +| `FAST_QUEUE_CONCURRENCY` | `32` | Container-plugin fast queue concurrency (see plugin-task-queue-architecture.md). | +| `FILE_LIGHT_SLOTS` | `16` | Max files concurrently in the light phase. | +| `FILE_BG_SLOTS` | `16` | Max files concurrently in the background phase. | +| `METADATA_FORMATS` | `meta` | Comma-separated list. `meta` = `.meta` YAML files; `jellyfin` = `.nfo` XML; empty disables sidecar generation. | +| `CONTAINER_PLUGINS_CONFIG` | `/app/plugins.yml` | Path to plugin manifest YAML. If missing, container plugins are skipped. | +| `DOCKER_SOCKET_PATH` | `/var/run/docker.sock` | Docker socket for container management. | +| `CONTAINER_CALLBACK_URL` | `http://meta-sort:8180` | URL plugins POST callbacks to. | +| `CONTAINER_META_CORE_URL` | `http://meta-sort` | URL plugins use for meta-core-compatible `/meta/*` routes (served by meta-sort itself for now). | +| `CONTAINER_NETWORK` | `meta-network` | Docker network to attach plugin containers to. | +| `PLUGIN_STACK_NAME` | — | Docker Compose project label so plugin containers group together in Docker Desktop. | +| `PLUGIN_CACHE_HOST_PATH` | — | Host path bind-mounted into each plugin for persistent caches. | -# Watch paths (comma-separated, relative to FILES_PATH or absolute) -WATCH_PATHS=/files +Source of truth: `packages/meta-sort-core/src/config/EnvConfig.ts`. -# Processing configuration -MAX_WORKER_THREADS=16 # Defaults to CPU count -LIGHT_PROCESSING_TIMEOUT_MS=120000 -HASH_PROCESSING_TIMEOUT_MS=7200000 +The Redis URL and WebDAV URLs are **not** configured via environment +variables — meta-sort discovers them from `kv-leader.info` (or its +in-memory equivalent fetched through meta-core's `/api/urls`). -# Remote mount configuration -REMOTE_STORAGE_CONFIG=/meta-core/config/remote-storage.conf +### Concurrency defaults + +Computed in `src/index.ts` at startup: + +``` +cpuCount = os.cpus().length +defaultBackgroundWorkers = max(1, floor(cpuCount / 2)) + +validationConcurrency = MAX_WORKER_THREADS ? M*2 : cpuCount * 2 +fastQueueConcurrency = MAX_WORKER_THREADS ?? cpuCount +backgroundQueueConcurrency = MAX_WORKER_THREADS ?? defaultBackgroundWorkers ``` -### Remote Storage Configuration +The pipeline logs the chosen values at startup — check container logs if +you need to know what was selected. -Create `/meta-core/config/remote-storage.conf`: +### Plugin configuration -```conf -# NFS mount (into FILES_VOLUME) -nfs://nas.local:/volume1/media /files/nfs/nas-media nfs defaults,rw 0 0 - -# SMB mount (credentials in separate file) -//server/share /files/smb/nas-share cifs credentials=/meta-core/config/smb-creds.txt,uid=1000 0 0 - -# rclone mount -gdrive: /files/gdrive rclone vfs-cache-mode=full 0 0 -``` +Container plugins are described in `dev/config/plugins.yml` (copy from +`plugins.yml.example`). Plugin runtime config (API keys, language, etc.) +is set through meta-sort's REST API, **not** in YAML. See the root +`CLAUDE.md` "Plugin Development" section for examples. --- -## API +## REST API -### REST Endpoints +Defined in `packages/meta-sort-core/src/api/UnifiedAPIServer.ts`. -| Endpoint | Method | Description | -|----------|--------|-------------| -| `/api/health` | GET | Service health check | -| `/api/scan/trigger` | POST | Trigger manual folder scan | -| `/api/processing/status` | GET | Current pipeline state | -| `/api/metrics` | GET | Performance metrics | -| `/api/mounts` | GET | List mounted remote storage | -| `/api/mounts` | POST | Add new remote mount | -| `/api/mounts/{id}` | DELETE | Remove remote mount | +### Health -### Processing Status Response +| Endpoint | Description | +|----------|-------------| +| `GET /health` | Liveness check. | +| `GET /api/health` | Liveness + KV connectivity. | +| `GET /meta-health` | meta-core-compatible health probe. | -```json -{ - "totalPending": 150, - "totalLightProcessing": 16, - "totalHashProcessing": 8, - "totalDone": 45000, - "throughput": { - "filesPerSecond": 34, - "avgLightProcessingMs": 470, - "avgHashProcessingMs": 15000 - } -} -``` +### Processing + +| Endpoint | Description | +|----------|-------------| +| `GET /api/processing/status` | Counters + state-manager snapshot. | +| `GET /api/processing/queue` | Per-queue depths (validation / fast / background). | +| `GET /api/processing/failed` | Failed-file roster. | +| `POST /api/processing/retry` | Retry a single failed file (`{ filePath }`). | +| `POST /api/processing/retry-all` | Retry every failed file. | +| `POST /api/processing/wait-empty` | Block until queues drain (with optional `timeout` query). | + +> Note: `/api/scan/trigger` and `/api/metadata/clear` were removed — +> they now live on **meta-core**. The `setupScanRoutes()` method in the +> source is intentionally a no-op left as a marker. + +### Plugins + +| Endpoint | Description | +|----------|-------------| +| `GET /api/plugins` | Discovered plugins (manifest + status + execution order). | +| `GET /api/plugins/timings` | Plugin processing-time stats. | +| `POST /api/plugins/:pluginId/activate` | Activate a plugin. | +| `POST /api/plugins/:pluginId/deactivate` | Deactivate a plugin. | +| `PUT /api/plugins/:pluginId/config` | Set runtime config (e.g. `{ apiKey: "..." }`). | +| `POST /api/plugins/:pluginId/clear-cache` | Clear a single plugin's cache. | +| `POST /api/plugins/clear-cache` | Clear every plugin's cache. | +| `POST /api/plugins/rescan` | Re-scan plugin manifests. | +| `POST /api/plugins/:pluginId/recompute` | Recompute this plugin for all files. | +| `POST /api/plugins/callback` | Container plugin task-complete callback (used by plugins, not humans). | +| `GET /api/plugins/containers` | Container-plugin status (health, instances, queue depths). | +| `GET /api/plugins/containers/:pluginId/manifest` | Plugin manifest. | +| `POST /api/plugins/containers/:pluginId/restart` | Restart instances. | +| `POST /api/plugins/containers/restart-all` | Restart all. | +| `POST /api/plugins/containers` | Add a new plugin (image, instances, resources, config, defaultQueue). | +| `PUT /api/plugins/containers/:pluginId` | Update plugin spec. | +| `DELETE /api/plugins/containers/:pluginId` | Remove plugin. | + +### Files and metadata + +| Endpoint | Description | +|----------|-------------| +| `GET /api/file/download?path=...` | Download a file via the configured WebDAV backend. | +| `POST /file/cid` | Compute the CID (midhash256) for a path (`{ path }`). | +| `GET /webdav/*` (and `PUT`, `DELETE`, `MKCOL`, `PROPFIND`) | WebDAV view of `/files`. Unauthenticated. | + +### meta-core-compatible KV (for container plugins) + +These mirror meta-core's `/meta/*` so plugin containers can talk to a +single host. They are only registered when the KV client is up. + +| Endpoint | Description | +|----------|-------------| +| `GET /meta/:hash` | All metadata for a file. | +| `GET /meta/:hash/*` | Single property. | +| `PUT /meta/:hash/*` | Set single property (`{ value }`). | +| `DELETE /meta/:hash/*` | Delete single property. | +| `PATCH /meta/:hash` | Merge partial metadata. | +| `POST /meta/:hash/_add/:key` | Append to a comma-separated set field. | + +### Other + +| Endpoint | Description | +|----------|-------------| +| `GET /api/metrics` | Performance metrics. | +| `GET /api/stats` | Redis stats summary (only when KV is up). | +| `GET /api/services` | Proxy to meta-core's `/api/services` via the LeaderClient. | --- -## Usage +## Operations -### Development +### Dev URLs + +- Behind Caddy (auth enforced, self-signed cert): + `https://metasort-dev.localhost:8180` +- Debug-direct backend (no auth, no TLS): `http://localhost:18180` +- Container name (backend): `metasort-app` +- Container name (hash-lock proxy): `metasort` + +### Logs ```bash -# Install dependencies -pnpm install - -# Build all packages -pnpm run build - -# Run in development mode -pnpm run dev +docker compose -f dev/docker-compose.yml logs -f metasort-app +docker exec metasort-app supervisorctl status ``` -### Docker +### Reload after a code change + +Use the reload script (in-container build via supervisord). Do **not** +`docker restart metasort-app`. ```bash -# Build and run with Docker Compose -docker-compose up -d - -# Check health -curl http://localhost/health -curl http://localhost/api/processing/status +cd dev +./scripts/reload-meta-sort.sh # full rebuild +./scripts/reload-meta-sort.sh --backend # backend only +./scripts/reload-meta-sort.sh --ui # UI only +./scripts/reload-meta-sort.sh --no-deps # skip dependency rebuild ``` +### Running tests + +```bash +docker exec meta-test-runner /app/test/test.sh sort +``` + +See the root `CLAUDE.md` "Running Tests" section for the full matrix. + --- +## Package layout + +This service is a nested pnpm workspace. Top-level structure: + +``` +packages/meta-sort/ +├── Dockerfile +├── docker/ # nginx, redis (legacy), supervisord configs +│ ├── nginx.conf +│ ├── redis.conf +│ └── supervisord.conf +├── docs/ # See "Further reading" below +├── package.json # Nested workspace root +├── packages/ +│ ├── meta-sort-core/ # @meta-sort/core - main service +│ │ ├── src/ +│ │ │ ├── api/ # UnifiedAPIServer.ts (Fastify) +│ │ │ ├── config/ # EnvConfig.ts, SupportedFileTypes.ts +│ │ │ ├── container-plugins/# ContainerManager, ContainerPluginScheduler, ... +│ │ │ ├── events/ # FileEventConsumer, SSEEventClient +│ │ │ ├── jellyfin/ # Jellyfin / NFO output helpers +│ │ │ ├── kv/ # KVManager, LeaderClient, RedisClient +│ │ │ ├── logic/ # StreamingPipeline, state managers, fileProcessor +│ │ │ ├── metrics/ # PerformanceMetrics +│ │ │ ├── plugin-engine/ # Vestigial in-process types + TaskScheduler glue +│ │ │ ├── types/ # Shared TypeScript types +│ │ │ ├── utils/ # Utilities +│ │ │ ├── webdav/ # WebDAV client + endpoint configuration +│ │ │ └── index.ts # Entry point +│ │ └── package.json +│ ├── meta-sort-ui/ # @meta-sort/ui - React dashboard (Vite) +│ ├── meta-sort-editor/ # Metadata editor UI (Vite) +│ ├── async-utils/ # @worph/async-utils - MultiQueue, etc. +│ └── shared/ # Symlinks to filename-tool, meta-hash, meta-interface +├── pnpm-workspace.yaml +└── pnpm-lock.yaml +``` + +> Built-in container plugins **do not live here.** They live at the +> repo root as separate submodules: +> `packages/plugins/metamesh-plugin-{file-info,ffmpeg,filename-parser,jellyfin-nfo,tmdb,anime-detector,language,subtitle,subtitle-extractor,torrent,fullhash}`. + +### Key components + +| Component | File | Purpose | +|-----------|------|---------| +| Entry point | `src/index.ts` | Wires KVManager → API server → plugin manager → ContainerManager → StreamingPipeline → FileEventConsumer. | +| Streaming pipeline | `src/logic/pipeline/StreamingPipeline.ts` | 3-queue (validation / fast / background) orchestrator. | +| State manager | `src/logic/UnifiedProcessingStateManager.ts` | Tracks `discovered → lightProcessing → hashProcessing → done`. | +| File processor | `src/logic/WatchedFileProcessor.ts` | Light + hash phase implementation. | +| SSE consumer | `src/events/FileEventConsumer.ts` | Translates meta-core events into pipeline calls. | +| SSE transport | `src/events/SSEEventClient.ts` | Long-lived HTTP SSE with cursor persistence. | +| KV manager | `src/kv/KVManager.ts` | Bootstraps RedisKVClient + ServiceRegistration from LeaderClient. | +| Leader client | `src/kv/LeaderClient.ts` | Reads `/meta-core/locks/kv-leader.info`, fetches `/api/urls`. | +| Container manager | `src/container-plugins/ContainerManager.ts` | Docker socket orchestration. | +| Container scheduler | `src/container-plugins/ContainerPluginScheduler.ts` | Plugin fast/background queues + callbacks. | +| API server | `src/api/UnifiedAPIServer.ts` | Fastify REST + WebDAV-facing meta routes. | +| Config | `src/config/EnvConfig.ts` | Environment variable parsing. | + +--- + +## Further reading + +In this directory: + +- [`docs/streaming-pipeline-architecture.md`](docs/streaming-pipeline-architecture.md) — pipeline internals. +- [`docs/containerized-plugin-architecture.md`](docs/containerized-plugin-architecture.md) — current plugin design. +- [`docs/plugin-task-queue-architecture.md`](docs/plugin-task-queue-architecture.md) — fast/background queue scheduling for plugin tasks. +- [`docs/plugin-system.md`](docs/plugin-system.md) — pointer to current docs (the in-process plugin system is gone). +- [`docs/fast-hash-global-id.md`](docs/fast-hash-global-id.md) — midhash256 / fast hash design. + +Repo-level docs of interest: + +- Root `CLAUDE.md` — authoritative ground truth on the dev stack, ports, + containers, and operational commands. +- `/METADATA_KEYS.md` — schema for the `/file/{cid}` Redis hash. +- `packages/meta-core/docs/` — leader election and Redis storage owner. + ## License -MIT License - see LICENSE file for details. +MIT — see LICENSE. diff --git a/docs/fast-hash-global-id.md b/docs/fast-hash-global-id.md index 17983e3..22de535 100644 --- a/docs/fast-hash-global-id.md +++ b/docs/fast-hash-global-id.md @@ -587,7 +587,7 @@ Deduplication logic remains identical. Fast-hash serves the same purpose as full The Stremio addon retrieves files by metadata queries (title, season, episode), not by hash. The hash is used internally for metadata storage. Changing to fast-hash is transparent to the addon. -### Meta-Orbit (Distributed Synchronization) +### meta-share (Distributed Synchronization) **Impact:** Hash serves as content identifier for P2P synchronization. @@ -716,7 +716,7 @@ MetaMesh implements a two-phase processing pipeline: **Phase 2: Hash Processing (Full Hashes - Background)** - Compute full-file hashes: SHA-256, SHA-1, MD5, CRC32 (slow, minutes) - Update etcd with additional hash metadata -- Required for Meta-Orbit distributed synchronization +- Required for meta-share distributed synchronization ### Why Both Are Needed @@ -726,7 +726,7 @@ MetaMesh implements a two-phase processing pipeline: - **Fast deduplication**: Quickly identify potential duplicates **Full Hash Purpose:** -- **Meta-Orbit integration**: Distributed P2P synchronization requires cryptographically secure hashes +- **meta-share integration**: Distributed P2P synchronization requires cryptographically secure hashes - **External system compatibility**: Torrent verification, IPFS CIDs - **Collision resolution**: If midhash256 collision detected, full SHA-256 provides definitive answer @@ -744,7 +744,7 @@ Light Processing (32 workers, parallel) Hash Processing (8 workers, background) - Compute SHA-256, SHA-1, MD5, CRC32 (15+ minutes for 50GB) - Update etcd with full hashes - - Enable Meta-Orbit synchronization ✓ + - Enable meta-share synchronization ✓ ``` ### Collision Detection and Handling @@ -778,4 +778,4 @@ The trade-offs—weaker cryptographic guarantees, potential collisions for adver This architecture aligns with Meta-Mesh's design philosophy: **optimize for the common case** (large media files, organic content, personal libraries) while maintaining **practicality and performance** over theoretical perfection. -Meta-Mesh leverages midhash256 for **immediate file availability** while computing full SHA-256 hashes **in the background** for Meta-Orbit distributed synchronization and external system compatibility. This dual-hash approach provides both speed and security. +Meta-Mesh leverages midhash256 for **immediate file availability** while computing full SHA-256 hashes **in the background** for meta-share distributed synchronization and external system compatibility. This dual-hash approach provides both speed and security. diff --git a/docs/plugin-system.md b/docs/plugin-system.md index 9984d4f..8710d58 100644 --- a/docs/plugin-system.md +++ b/docs/plugin-system.md @@ -1,762 +1,35 @@ -# Meta-Sort Plugin System Architecture +# Plugin System (Deprecated — See Container Plugins) -This document describes the plugin system architecture for meta-sort, enabling extensible metadata extraction through dynamically loaded plugins. +> **This document is a historical pointer.** The in-process plugin system it +> originally described — `plugins//{manifest.yml,index.ts}` directories +> loaded via dynamic `import()` inside meta-sort-core — has been removed. +> All plugins now run as Docker containers and communicate over HTTP. -## Overview +For the current design, read: -The plugin system provides a modular architecture for metadata extraction. Plugins are self-contained units that: +- **[containerized-plugin-architecture.md](./containerized-plugin-architecture.md)** — full container plugin design (HTTP contract, lifecycle, callbacks, KV access). +- **[plugin-task-queue-architecture.md](./plugin-task-queue-architecture.md)** — how plugin work is dispatched across the fast / background queues. +- Root **`CLAUDE.md`** (`Plugin Development` and `Runtime Plugin Configuration` sections) — operational quickstart. +- Plugin sources at **`packages/plugins/metamesh-plugin-*`** (separate submodules) — `file-info`, `ffmpeg`, `filename-parser`, `jellyfin-nfo`, `tmdb`, `anime-detector`, `language`, `subtitle`, `subtitle-extractor`, `torrent`, `fullhash`. -- Are discovered and loaded from a plugins directory at startup -- Can be activated/deactivated at runtime without restart -- Process files sequentially in dependency order -- Store metadata in a flat key-value format -- Have access to their own persistent cache folder +## What survived from the old design -## Directory Structure +The `Plugin` / `PluginContext` / `PluginLifecycleContext` / `PluginKVStore` +interfaces in `src/plugin-engine/types.ts` are still exported, but the +container-only `PluginManager` (`src/plugin-engine/PluginManager.ts`) never +constructs an in-process `Plugin` instance — manifests, lifecycle, and +execution are owned by `ContainerManager` and `ContainerPluginScheduler` +under `src/container-plugins/`. Treat the `Plugin`-interface types as +vestigial and do not write new code against them. -``` -packages/meta-sort/packages/meta-sort-core/ -├── plugins/ # All plugins (shipped + custom) -│ ├── file-info/ -│ │ ├── manifest.yml # Plugin metadata & config schema -│ │ └── index.ts # Plugin implementation -│ ├── ffmpeg/ -│ ├── filename-parser/ -│ ├── anime-detector/ -│ ├── jellyfin-nfo/ -│ ├── subtitle/ -│ ├── torrent/ -│ ├── language/ -│ ├── tmdb/ -│ └── debug/ -├── plugins.json # Runtime state (active/inactive + configs) -└── src/ - └── plugin/ - ├── types.ts # Type definitions - ├── PluginLoader.ts # Manifest parsing, dynamic import - ├── PluginContext.ts # Context factory - ├── PluginManager.ts # Lifecycle management - └── PluginExecutor.ts # Dependency-ordered execution -``` +## Why the change -Cache folders are stored separately under `CACHE_FOLDER_PATH` (default `/data/cache/plugins/`): +| Limitation of in-process plugins | Container model | +|----------------------------------|-----------------| +| Heavy native deps (FFmpeg, Whisper, ML libs) inflated the meta-sort image | Each plugin ships its own image | +| All plugins had to be TypeScript / Node | Any language with an HTTP server | +| A crashing plugin took down meta-sort | Plugin crashes are isolated; meta-sort restarts the container | +| Plugin updates required redeploying meta-sort | Plugins are versioned and pulled independently | -``` -/data/cache/plugins/ -├── ffmpeg/ -│ ├── {cid1}.json -│ └── {cid2}.json -├── tmdb/ -│ ├── {cid1}.json -│ └── {cid2}.json -└── ... -``` - -## Plugin Manifest - -Each plugin must have a `manifest.yml` file defining its metadata, dependencies, and configuration schema. - -```yaml -# manifest.yml -id: tmdb # Unique identifier (matches folder name) -name: TMDB Enricher # Human-readable name -version: 1.0.0 # Semantic version -description: Enriches metadata from The Movie Database API - -# Plugins that must execute BEFORE this one -dependencies: - - filename-parser # Needs parsed title/year - -# Configuration schema (rendered in dashboard) -config: - apiKey: - type: string - label: API Key - required: true - secret: true # Masked in UI - language: - type: string - label: Language - default: en-US - includeAdult: - type: boolean - label: Include Adult Content - default: false - -# Metadata schema for editor UI (optional) -# Describes keys this plugin writes -schema: - tmdb/id: - label: TMDB ID - type: string - readonly: true - tmdb/title: - label: Title - type: string - tmdb/overview: - label: Overview - type: text - tmdb/poster: - label: Poster URL - type: string - format: url - tmdb/genre/{n}: - label: Genre - type: string - indexed: true # Indicates indexed keys (0, 1, 2...) -``` - -### Manifest Fields - -| Field | Required | Description | -|-------|----------|-------------| -| `id` | Yes | Unique plugin identifier, must match folder name | -| `name` | Yes | Display name for dashboard | -| `version` | Yes | Semantic version string | -| `description` | No | Brief description | -| `dependencies` | No | Array of plugin IDs that must run before this one | -| `config` | No | Configuration schema for dashboard UI | -| `schema` | No | Metadata keys written by this plugin (for editor) | - -### Config Field Types - -```yaml -config: - fieldName: - type: string | number | boolean | select - label: Display Label - required: true | false # Default: false - default: # Default value - secret: true | false # Mask in UI (default: false) - options: # For type: select - - value: opt1 - label: Option 1 - - value: opt2 - label: Option 2 -``` - -## Plugin Interface - -Plugins export a default object implementing the `Plugin` interface. - -```typescript -// types.ts - -export interface Plugin { - /** - * Called once when the plugin is loaded at startup. - * Use for one-time initialization. - */ - onLoad?(ctx: PluginLifecycleContext): Promise; - - /** - * Called when the plugin is activated. - * Triggered at startup (if previously active) or from dashboard. - */ - onActivate?(ctx: PluginLifecycleContext): Promise; - - /** - * Called when the plugin is deactivated from the dashboard. - * Use for cleanup, releasing resources. - */ - onDeactivate?(ctx: PluginLifecycleContext): Promise; - - /** - * Called when the plugin is unloaded at shutdown. - */ - onUnload?(ctx: PluginLifecycleContext): Promise; - - /** - * Process a single file. Called for each file in the library. - * Must be implemented by all plugins. - */ - process(ctx: PluginContext): Promise; -} - -export interface PluginLifecycleContext { - /** Plugin configuration from plugins.json */ - config: Record; - - /** Absolute path to plugin's cache directory */ - cacheDir: string; - - /** Scoped logger */ - log: PluginLogger; -} -``` - -## Plugin Context - -The `PluginContext` is injected into the `process()` function, providing access to file metadata and utilities. - -```typescript -export interface PluginContext { - /** - * Key-value storage for the current file's metadata. - * Keys use path notation: "video/codec", "audio/0/language" - */ - kv: PluginKVStore; - - /** Absolute path to the file being processed */ - filePath: string; - - /** Plugin configuration values */ - config: Record; - - /** Absolute path to plugin's cache directory */ - cacheDir: string; - - /** Cache utilities for persistent storage */ - cache: PluginCache; - - /** Scoped logger (prefixed with plugin name) */ - log: PluginLogger; -} -``` - -### KV Store Interface - -```typescript -export interface PluginKVStore { - /** - * Get a value by key. - * @param key Path-style key, e.g., "video/codec" or "audio/0/language" - */ - get(key: string): string | undefined; - - /** - * Set a value. Only string values are supported. - * @param key Path-style key - * @param value String value - */ - set(key: string, value: string): void; - - /** - * Delete a key. - */ - delete(key: string): void; - - /** - * Get all keys, optionally filtered by prefix. - * @param prefix Optional prefix filter, e.g., "audio/" returns ["audio/0/codec", "audio/0/language", ...] - */ - keys(prefix?: string): string[]; - - /** - * Get all key-value pairs, optionally filtered by prefix. - */ - entries(prefix?: string): Array<[string, string]>; -} -``` - -### Cache Interface - -```typescript -export interface PluginCache { - /** - * Get full path for a cache file. - * @param filename Relative filename within cache directory - */ - getPath(filename: string): string; - - /** - * Read and parse a JSON cache file. - * @returns Parsed data or null if not found - */ - readJson(filename: string): Promise; - - /** - * Write data as JSON to cache file. - */ - writeJson(filename: string, data: unknown): Promise; - - /** - * Check if a cache file exists. - */ - exists(filename: string): Promise; - - /** - * Delete a cache file. - */ - delete(filename: string): Promise; - - /** - * Clear all files in the plugin's cache directory. - */ - clear(): Promise; -} -``` - -### Logger Interface - -```typescript -export interface PluginLogger { - debug(message: string, ...args: unknown[]): void; - info(message: string, ...args: unknown[]): void; - warn(message: string, ...args: unknown[]): void; - error(message: string, ...args: unknown[]): void; -} -``` - -## Plugin Implementation Example - -```typescript -// plugins/tmdb/index.ts -import type { Plugin, PluginContext, PluginLifecycleContext } from '../../src/plugin/types'; - -interface TmdbConfig { - apiKey: string; - language: string; - includeAdult: boolean; -} - -const plugin: Plugin = { - async onLoad(ctx: PluginLifecycleContext) { - ctx.log.info('TMDB plugin loaded'); - }, - - async onActivate(ctx: PluginLifecycleContext) { - const config = ctx.config as TmdbConfig; - if (!config.apiKey) { - ctx.log.warn('No API key configured, plugin will skip processing'); - } - }, - - async onDeactivate(ctx: PluginLifecycleContext) { - ctx.log.info('TMDB plugin deactivated'); - }, - - async process(ctx: PluginContext) { - const config = ctx.config as TmdbConfig; - - // Skip if not configured - if (!config.apiKey) return; - - // Get data from previous plugins - const title = ctx.kv.get('filename/title'); - const year = ctx.kv.get('filename/year'); - const videoType = ctx.kv.get('video/type'); - - if (!title) return; - - // Check cache first (keyed by title+year) - const cacheKey = `${sanitize(title)}-${year || 'unknown'}.json`; - const cached = await ctx.cache.readJson(cacheKey); - - if (cached) { - applyResult(ctx.kv, cached); - return; - } - - // Fetch from TMDB API - try { - const result = await fetchTmdb(title, year, videoType, config); - if (result) { - await ctx.cache.writeJson(cacheKey, result); - applyResult(ctx.kv, result); - } - } catch (error) { - ctx.log.error('TMDB fetch failed', error); - } - } -}; - -function applyResult(kv: PluginContext['kv'], result: TmdbResult) { - kv.set('tmdb/id', String(result.id)); - kv.set('tmdb/title', result.title); - kv.set('tmdb/overview', result.overview); - kv.set('tmdb/poster', result.poster_path); - kv.set('tmdb/backdrop', result.backdrop_path); - kv.set('tmdb/rating', String(result.vote_average)); - kv.set('tmdb/year', result.release_date?.slice(0, 4) || ''); - - // Genres as indexed keys - result.genres?.forEach((genre, i) => { - kv.set(`tmdb/genre/${i}`, genre.name); - }); -} - -export default plugin; -``` - -## Metadata Key Format - -All metadata is stored as flat key-value pairs with string values only. Keys use path notation with `/` as separator. - -### Key Naming Convention - -``` -{plugin-id}/{field} # Simple field -{plugin-id}/{group}/{field} # Grouped field -{plugin-id}/{group}/{index}/{field} # Indexed field (for lists) -``` - -### Examples - -``` -# File info plugin -file/name = "Movie.2024.1080p.mkv" -file/size = "4294967296" -file/type = "video" -file/mime = "video/x-matroska" - -# FFmpeg plugin -video/codec = "h264" -video/width = "1920" -video/height = "1080" -video/duration = "7200.5" -video/bitrate = "8000000" -audio/0/codec = "aac" -audio/0/language = "eng" -audio/0/channels = "6" -audio/1/codec = "dts" -audio/1/language = "fra" -audio/1/channels = "6" -subtitle/0/language = "eng" -subtitle/0/format = "subrip" -subtitle/1/language = "fra" -subtitle/1/format = "subrip" - -# Filename parser plugin -filename/title = "Movie Title" -filename/year = "2024" -filename/quality = "1080p" -filename/source = "bluray" - -# TMDB plugin -tmdb/id = "12345" -tmdb/title = "Movie Title" -tmdb/overview = "A movie about..." -tmdb/poster = "/path/to/poster.jpg" -tmdb/genre/0 = "Action" -tmdb/genre/1 = "Adventure" -``` - -### Key Constraints - -- Keys must be non-empty strings -- Keys use `/` as path separator -- Values must be strings (numbers converted via `String()`) -- No nested objects or arrays (use indexed keys instead) -- Keys are case-sensitive -- Recommended: use lowercase with hyphens for multi-word keys - -## Plugin Manager - -The `PluginManager` handles plugin lifecycle and orchestration. - -```typescript -export class PluginManager extends EventEmitter { - /** - * Scan plugins directory and load manifests. - * Called once at startup. - */ - async scanPlugins(): Promise; - - /** - * Load all plugins and activate those marked active in state. - * Called once at startup after scanning. - */ - async loadAll(): Promise; - - /** - * Activate a plugin (hot, no restart required). - * Calls plugin.onActivate(), rebuilds execution order, saves state. - * Emits 'plugin:activated' event. - */ - async activate(pluginId: string): Promise; - - /** - * Deactivate a plugin (hot, no restart required). - * Calls plugin.onDeactivate(), rebuilds execution order, saves state. - * Emits 'plugin:deactivated' event. - */ - async deactivate(pluginId: string): Promise; - - /** - * Update plugin configuration. - * Saves to plugins.json. - * Emits 'plugin:config-changed' event. - */ - async updateConfig(pluginId: string, config: Record): Promise; - - /** - * Clear a plugin's cache directory. - */ - async clearCache(pluginId: string): Promise; - - /** - * Get all discovered plugins. - */ - getPlugins(): PluginInfo[]; - - /** - * Get only active plugins. - */ - getActivePlugins(): PluginInfo[]; - - /** - * Get execution order (dependency-sorted active plugin IDs). - */ - getExecutionOrder(): string[]; - - /** - * Process a file through all active plugins. - */ - async processFile(filePath: string, kv: KVStore): Promise; - - /** - * Shutdown: deactivate and unload all plugins. - */ - async shutdown(): Promise; -} -``` - -### Events - -| Event | Payload | Description | -|-------|---------|-------------| -| `plugin:activated` | `{ pluginId: string }` | Plugin was activated | -| `plugin:deactivated` | `{ pluginId: string }` | Plugin was deactivated | -| `plugin:config-changed` | `{ pluginId: string, config: object }` | Plugin config updated | -| `plugin:error` | `{ pluginId: string, error: Error }` | Plugin error occurred | - -### Plugin Info - -```typescript -export interface PluginInfo { - id: string; - name: string; - version: string; - description?: string; - dependencies: string[]; - active: boolean; - config: Record; - configSchema?: ConfigSchema; - metadataSchema?: MetadataSchema; - status: 'loaded' | 'error' | 'unloaded'; - error?: string; -} -``` - -## State Management - -Plugin state is persisted in `plugins.json` located in the meta-sort-core package root. - -```json -{ - "plugins": { - "file-info": { - "active": true, - "config": {} - }, - "ffmpeg": { - "active": true, - "config": {} - }, - "tmdb": { - "active": true, - "config": { - "apiKey": "your-api-key", - "language": "en-US", - "includeAdult": false - } - }, - "custom-plugin": { - "active": false, - "config": {} - } - } -} -``` - -### State Behavior - -- **New plugins**: Discovered plugins not in state file are added with `active: false` -- **Removed plugins**: Plugins in state but not on disk are kept (preserves config for reinstall) -- **Config merge**: Default values from manifest are merged with saved config -- **Persistence**: State is saved immediately on any change - -## Execution Pipeline - -### Startup Sequence - -``` -1. PluginManager.scanPlugins() - └── Scan plugins/ directory - └── Parse each manifest.yml - └── Validate manifest structure - -2. PluginManager.loadAll() - └── Load plugins.json state - └── For each discovered plugin: - ├── Dynamic import(plugin/index.js) - ├── Validate Plugin interface - ├── Call plugin.onLoad() - └── If active in state: - └── Call plugin.onActivate() - └── Resolve execution order (topological sort) -``` - -### File Processing Sequence - -``` -1. Create KVStore for file - -2. For each active plugin (in dependency order): - ├── Create PluginContext - ├── Start timing - ├── try: - │ └── await plugin.process(ctx) - └── catch: - └── Log error, continue to next plugin - -3. Persist KV to Redis -``` - -### Activation Sequence (Hot) - -``` -1. PluginManager.activate(pluginId) - ├── Validate plugin exists and is loaded - ├── Check dependencies are active - ├── Call plugin.onActivate() - ├── Mark as active in memory - ├── Rebuild execution order - ├── Save plugins.json - └── Emit 'plugin:activated' event - -2. Caller receives event - └── Trigger library rescan -``` - -### Deactivation Sequence (Hot) - -``` -1. PluginManager.deactivate(pluginId) - ├── Check no active plugins depend on this one - ├── Call plugin.onDeactivate() - ├── Mark as inactive in memory - ├── Rebuild execution order - ├── Save plugins.json - └── Emit 'plugin:deactivated' event - -2. Caller receives event - └── Trigger library rescan - -Note: Metadata written by the plugin remains in Redis. - A separate "clean metadata" feature can remove orphaned keys. -``` - -## Dependency Resolution - -Plugins declare dependencies in their manifest. The system performs topological sort to determine execution order. - -### Rules - -1. A plugin cannot be activated if any of its dependencies are inactive -2. A plugin cannot be deactivated if any active plugin depends on it -3. Circular dependencies are detected and rejected at load time -4. Missing dependencies are logged as warnings; plugin can still load but may fail - -### Example - -```yaml -# filename-parser depends on file-info -# tmdb depends on filename-parser -# anime-detector depends on filename-parser - -Resolved order: file-info → filename-parser → [tmdb, anime-detector] -``` - -Plugins at the same dependency level may execute in any order (or parallel in future). - -## REST API - -The plugin system exposes REST endpoints for dashboard integration. - -### Endpoints - -``` -GET /api/plugins # List all plugins -GET /api/plugins/:id # Get plugin details -POST /api/plugins/:id/activate # Activate plugin -POST /api/plugins/:id/deactivate # Deactivate plugin -PUT /api/plugins/:id/config # Update plugin config -DELETE /api/plugins/:id/cache # Clear plugin cache -GET /api/plugins/:id/schema # Get metadata schema (for editor) -``` - -### Response Format - -```typescript -// GET /api/plugins -interface PluginsResponse { - plugins: PluginInfo[]; - executionOrder: string[]; -} - -// POST /api/plugins/:id/activate -interface ActivateResponse { - success: boolean; - error?: string; # e.g., "Dependency 'x' is not active" -} - -// PUT /api/plugins/:id/config -interface ConfigUpdateRequest { - config: Record; -} -``` - -## Error Handling - -### Plugin Load Errors - -- Invalid manifest: Plugin marked as `status: 'error'`, logged, skipped -- Import failure: Plugin marked as `status: 'error'`, logged, skipped -- Missing dependencies: Warning logged, plugin can load but activation may fail - -### Plugin Execution Errors - -- Errors in `process()` are caught, logged, and execution continues to next plugin -- The file's metadata from previous plugins is preserved -- Error count is tracked per plugin for monitoring - -### Lifecycle Hook Errors - -- Errors in `onActivate()` prevent activation, error returned to caller -- Errors in `onDeactivate()` are logged but deactivation proceeds -- Errors in `onLoad()`/`onUnload()` are logged, plugin marked as error state - -## Future Considerations - -### Metadata Cleanup - -A future feature will allow cleaning metadata keys written by a specific plugin: - -```typescript -// Future API -await pluginManager.cleanMetadata(pluginId); -// Removes all keys matching "{pluginId}/*" from Redis -``` - -### Parallel Execution - -Plugins at the same dependency level could execute in parallel: - -```typescript -// Future: parallel execution of independent plugins -const level1 = ['file-info']; -const level2 = ['ffmpeg', 'filename-parser']; // parallel -const level3 = ['tmdb', 'anime-detector']; // parallel -``` - -### Plugin Hot Reload - -Reloading a plugin's code without full restart (for development): - -```typescript -// Future API -await pluginManager.reload(pluginId); -// Re-imports plugin code, preserves state -``` +See `containerized-plugin-architecture.md` for the migration rationale and +the HTTP contract container plugins must implement. diff --git a/docs/streaming-pipeline-architecture.md b/docs/streaming-pipeline-architecture.md index fdbbe41..3db08a2 100644 --- a/docs/streaming-pipeline-architecture.md +++ b/docs/streaming-pipeline-architecture.md @@ -1,863 +1,302 @@ # Streaming Pipeline Architecture -**Author**: Meta Mesh Development Team -**Last Updated**: 2025-11-03 -**Status**: Current Implementation +The meta-sort streaming pipeline processes media files as they arrive from +meta-core (the file watcher), classifies them by extension, extracts light +metadata + a permanent identifier, and dispatches heavier work to container +plugins. It is implemented in +`packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts`. --- -## Table of Contents +## Contents 1. [Overview](#overview) -2. [Architecture Principles](#architecture-principles) -3. [Pipeline Stages](#pipeline-stages) -4. [Component Design](#component-design) -5. [Data Flow](#data-flow) -6. [State Management](#state-management) -7. [Concurrency Model](#concurrency-model) -8. [Performance Characteristics](#performance-characteristics) -9. [Architecture Decisions](#architecture-decisions) -10. [Future Considerations](#future-considerations) +2. [Event source: meta-core SSE](#event-source-meta-core-sse) +3. [Pipeline stages](#pipeline-stages) +4. [State manager](#state-manager) +5. [Concurrency model](#concurrency-model) +6. [Reset semantics (generation counter)](#reset-semantics-generation-counter) +7. [Stats and observability](#stats-and-observability) +8. [Key code locations](#key-code-locations) --- ## Overview -The Meta Mesh streaming pipeline is a **multi-stage asynchronous processing system** designed to efficiently process hundreds of thousands of media files with minimal memory overhead and optimal CPU utilization. +``` + meta-core /api/events/files (SSE) + │ + ▼ + FileEventConsumer + │ add / change / delete / rename / reset + ▼ + handleFileAdded ─┐ + handleFileChanged┤───────► validationQueue (PQueue) + handleFileDeleted┘ │ extension filter + ▼ + fastQueue (PQueue) + processLightPhase + - filename parse, FFprobe-light, stats + - midhash256 (permanent ID) computed or + adopted from meta-core's pre-computed value + - KV write → file appears in VFS + - dispatch all enabled container plugins + │ + ▼ + backgroundQueue (PQueue) + processHashPhase + - SHA-256, additional heavy hashing + │ + ▼ + await ContainerPluginScheduler `file:complete` + (or complete immediately if no container plugins) + │ + ▼ + UnifiedProcessingStateManager → DONE +``` -### Key Characteristics +Two things are worth highlighting because they differ from older +documentation: -- **Streaming Architecture**: Files are processed as they're discovered, not batched -- **Multi-Stage Pipeline**: Four independent stages with different concurrency levels -- **Non-Blocking**: Uses async generators to prevent event loop starvation -- **Scalable**: Handles 900k+ files and 100TB+ datasets -- **Separation of Concerns**: Generic file discovery separated from business logic - -### Performance Goals - -| Metric | Target | Current | -|--------|--------|---------| -| Discovery Speed | Immediate streaming | ✅ Achieved | -| Time to First Result | < 5 seconds | ✅ Achieved | -| Event Loop Blocking | Zero | ✅ Achieved | -| Memory Overhead | O(n) for queues only | ✅ Achieved | -| CPU Utilization | 100% during processing | ✅ Achieved | +1. **midhash256 is the permanent file ID.** It is computed during the fast + queue (or supplied pre-computed by meta-core in the SSE payload) and the + file becomes accessible in the VFS as soon as the fast queue commits its + KV write. There is no `tempId` and no rename when SHA-256 finishes. +2. **There is no in-process plugin pipeline.** Filename parsing, ffprobe, + TMDB, etc. are container plugins running as separate Docker containers; + `StreamingPipeline` dispatches to them via + `ContainerPluginScheduler.dispatchAllPlugins` and waits for the + scheduler's `file:complete` event before marking the file DONE. --- -## Architecture Principles +## Event source: meta-core SSE -### 1. Streaming Over Batching +meta-sort no longer reads the filesystem directly. File discovery and +change watching live in meta-core, which streams events over HTTP SSE. -**Problem**: Original implementation waited for complete file discovery before processing, causing 15+ minute delays. +- **Subscriber:** `src/events/FileEventConsumer.ts` — converts each event + to an absolute path (`filesPath` + relative path from meta-core) and + forwards it to the pipeline. +- **Transport:** `src/events/SSEEventClient.ts` — long-lived HTTP + connection to `${metaCoreApiUrl}/api/events/files`, with cursor + persistence at `/meta-core/cursors/meta-sort-files.cursor`, automatic + reconnect, and `Last-Event-ID` resume semantics. +- **Cursor gap (`gap` event):** the cursor was trimmed out of retention. + The default response is to log and wait for the next `reset` event from + meta-core; the pipeline rebuilds from there. -**Solution**: Stream files through the pipeline as they're discovered. +Event types and their effect on the pipeline: -```typescript -// ❌ Batch Processing (Old) -const allFiles = await discoverAllFiles(); // Wait for all files -await processFiles(allFiles); // Then process - -// ✅ Streaming Processing (New) -for await (const file of discoverFiles()) { // Process as discovered - pipeline.validate(file); // Immediate processing -} -``` - -### 2. Separation of Concerns - -**Problem**: Business logic (extension filtering) was mixed with generic utilities (file discovery). - -**Solution**: Clean separation between packages: - -- **meta-hash** (Generic Library): Platform-agnostic file discovery -- **meta-mesh** (Application): Business-specific processing logic - -This allows `meta-hash` to be reused in other contexts without Meta Mesh-specific dependencies. - -### 3. Event Loop Yielding - -**Problem**: Tight synchronous loops starved the Node.js event loop, preventing PQueue from dispatching work. - -**Solution**: Use async generators that naturally yield to the event loop: - -```typescript -async *walkDirectory(directory: string): AsyncGenerator { - const entries = await readdir(directory); // Async I/O - yields to event loop - - for (const entry of entries) { - yield fullPath; // Generator yield - allows event loop to process other tasks - } -} -``` - -### 4. Independent Stage Queues - -**Problem**: Single queue with uniform concurrency doesn't match workload characteristics. - -**Solution**: Each stage has its own queue with optimized concurrency: - -- **Validation**: High concurrency (fast I/O checks) -- **Metadata**: Medium concurrency (balanced CPU/I/O) -- **Hash**: Medium concurrency (CPU-bound operations) +| SSE event | Pipeline call | Notes | +|-----------|---------------|-------| +| `add` | `handleFileAdded(path, midhash256?)` | Standard new-file path. | +| `change` | `handleFileChanged(path, midhash256?)` | Removes any prior state, re-validates. | +| `delete` | `handleFileDeleted(path)` | Drops state + calls `fileProcessor.deleteFile`. | +| `rename` | delete(oldPath) + add(newPath) | Old path is removed; new path is re-ingested. | +| `reset` | `pipeline.reset()` | Increments the reset generation, clears queues. | --- -## Pipeline Stages +## Pipeline stages -The pipeline consists of four stages, each optimized for its specific workload: +### Stage 1: Validation (`validateFile`) -``` -┌─────────────┐ ┌────────────┐ ┌──────────┐ ┌──────┐ ┌──────┐ -│ Discovery │───▶│ Validation │───▶│ Metadata │───▶│ Hash │───▶│ Done │ -│ (Stream) │ │ (32 queue) │ │(16 queue)│ │(16 q)│ │ │ -└─────────────┘ └────────────┘ └──────────┘ └──────┘ └──────┘ - Async Gen Extension Quick Meta Content Final - Yields files Filtering Extraction Hashing Storage -``` +`StreamingPipeline.validateFile(filePath, midhash256?)` -### Stage 0: Discovery (Streaming) +- Extension check against `PipelineConfig.supportedExtensions` + (`src/config/SupportedFileTypes.ts`). No I/O. +- Files with unsupported extensions are removed from the state manager and + counted in `filteredCount`. They do not progress. +- Strict MIME validation is intentionally **not** implemented — the comment + in code reserves the option but it is off by default and there is no + external library wired in. +- A generation check protects against in-flight tasks completing after a + `reset()` (see [Reset semantics](#reset-semantics-generation-counter)). +- Valid files are enqueued onto the fast queue via + `this.fastQueue.add(() => this.processLightPhase(filePath, midhash256))`. -**Location**: `packages/meta-hash/src/lib/folder-watcher/FolderWatcher.ts` +### Stage 2: Fast queue (`processLightPhase`) -**Purpose**: Discover all files in watched directories and emit paths as async generator. +`StreamingPipeline.processLightPhase(filePath, midhash256?)` calls +`fileProcessor.processLightPhase(...)` which: -**Implementation**: -```typescript -async *discoverFiles(directories: string[]): AsyncGenerator { - for (const dir of directories) { - yield* this.walkDirectory(dir); - } -} -``` +1. Computes or adopts `midhash256` (the permanent file ID, written as + `cid_midhash256` on the metadata record). +2. Extracts light metadata (filename parsing via `@metazla/filename-tools`, + basic stat/ffprobe info — see `src/logic/fileProcessor/` for the + adapter, and the file-info / filename-parser / ffmpeg container + plugins for the actual work). +3. Writes metadata to Redis through the KV client. The file is now visible + to meta-fuse, meta-stremio, etc. -**Characteristics**: -- **Concurrency**: N/A (single-threaded recursive walk) -- **I/O Pattern**: Sequential directory reads with async/await -- **Output**: File path strings -- **Filtering**: None (emits all files regardless of type) -- **Duration**: Depends on filesystem speed and directory depth +After `processLightPhase` returns, `StreamingPipeline`: -**Key Design Choice**: No filtering logic at this layer keeps the component generic and reusable. All business logic happens in downstream stages. +- Sends a fire-and-forget `/add` POST to the Stremio addon on + `http://localhost:7000` (silent fail if it isn't running). +- Calls `dispatchContainerPluginTasks(filePath, hashId, existingMeta)` — + this iterates every enabled container plugin via + `ContainerPluginScheduler.dispatchAllPlugins`. Tasks are routed to the + scheduler's own fast / background queues based on each plugin manifest's + `defaultQueue`. Plugins write results back through meta-core's `/meta/:hash` + API; `StreamingPipeline` does not aggregate them itself. +- Enqueues the file on the background queue + (`processHashPhase`) for the in-process hashing pass. + +### Stage 3: Background queue (`processHashPhase`) + +`StreamingPipeline.processHashPhase(filePath)` calls +`fileProcessor.processHashPhase(...)` for the slow in-process hashing +(SHA-256 and any other heavy hashes the file processor computes). + +Completion path depends on whether container plugins are configured: + +- **No container plugins:** the file is moved to DONE immediately via + `stateManager.completeHashProcessing(filePath, hashId, virtualPath)`. +- **Container plugins configured:** the background queue completes, but + the file stays in `hashProcessing` until the `ContainerPluginScheduler` + emits `file:complete` for that `(fileHash, filePath)` pair. The + scheduler-side listener (`setContainerPluginScheduler` in + `StreamingPipeline.ts`) re-reads metadata from Redis, computes the + virtual path via `MetaDataToFolderStruct.renamingRule(...)`, and then + transitions the file to DONE. + +The scheduler listener also defends against three races: a different reset +generation, a missing `hashProcessing` entry, and a hash-mismatch where the +callback belongs to a previous attempt for the same path. --- -### Stage 1: Validation (32 Workers) +## State manager -**Location**: `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts::validateFile()` +`src/logic/UnifiedProcessingStateManager.ts` -**Purpose**: Fast extension-based filtering to eliminate unsupported file types. - -**Implementation**: -```typescript -private async validateFile(filePath: string): Promise { - // Extension check (instant, no I/O) - const ext = path.extname(filePath).toLowerCase(); - if (!this.config.supportedExtensions.has(ext)) { - return; // Skip unsupported files - } - - // Mark as pending - this.config.stateManager.addPending(filePath); - this.validatedCount++; - - // Stage 2: Metadata (fire and forget) - this.metadataQueue.add(() => this.extractMetadata(filePath)); -} -``` - -**Characteristics**: -- **Concurrency**: 32 workers (CPU count × 2) -- **I/O Pattern**: Zero disk I/O (string operations only) -- **Processing Time**: < 1ms per file -- **Success Rate**: ~65% (video/subtitle/torrent files only) -- **Queue Behavior**: Typically empty (keeps up with discovery) - -**Key Design Choice**: High concurrency (32 workers) is acceptable because validation is pure CPU-bound string operations with no I/O. This ensures validation never becomes a bottleneck. - -**MIME Validation (Skipped)**: The original implementation included optional MIME type validation (reading file headers). This was removed because: -- Requires disk I/O (read first 4KB of each file) -- Adds ~25% overhead to discovery time -- Extension checking is sufficient for 99.9% of cases -- Edge cases are handled gracefully in later stages - ---- - -### Stage 2: Metadata Extraction (16 Workers) - -**Location**: `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts::extractMetadata()` - -**Purpose**: Extract lightweight metadata without computing content hash. - -**Implementation**: -```typescript -private async extractMetadata(filePath: string): Promise { - this.config.stateManager.startLightProcessing(filePath); - - // Call existing light processing logic (quick metadata extraction) - const quickMeta = await this.config.fileProcessor.processFileQuick(filePath); - - // Mark light processing as complete and move to hash queue - this.config.stateManager.completeLightProcessing(filePath); - this.metadataCount++; - - // Stage 3: Hash (fire and forget) - this.hashQueue.add(() => this.computeHash(filePath)); -} -``` - -**What Metadata is Extracted**: -1. **Filename Parsing** (via `filename-tool`): - - Series title, season, episode number - - Movie title, year - - Quality indicators (1080p, 4K, etc.) - - Language codes - - Release group - -2. **File System Stats**: - - File size - - Modification time - - Permissions - -3. **Basic Media Info** (via FFmpeg): - - Duration - - Container format - - Codec information (video/audio) - - Resolution (width × height) - - Bitrate - -**Characteristics**: -- **Concurrency**: 16 workers (CPU count) -- **I/O Pattern**: Light disk reads + FFmpeg process spawning -- **Processing Time**: ~470ms average per file -- **Storage**: Temporary metadata stored in etcd with `tempId` -- **Virtual FS**: Files appear in virtual filesystem immediately after this stage - -**Key Design Choice**: This stage extracts "just enough" metadata to populate the virtual filesystem and make files available to users, while deferring expensive hash computation to the next stage. This provides immediate value while processing continues in the background. - ---- - -### Stage 3: Hash Computation (16 Workers) - -**Location**: `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts::computeHash()` - -**Purpose**: Compute content hashes and perform deep media analysis. - -**Implementation**: -```typescript -private async computeHash(filePath: string): Promise { - const current = this.hashCount + 1; - const queueSize = this.discoveredCount; - - await this.config.fileProcessor.processFile(filePath, current, queueSize); - - this.hashCount++; - this.config.stateManager.completeHashProcessing(filePath); -} -``` - -**What is Computed**: -1. **Content Hashes** (via `meta-hash`): - - SHA-256 (primary hash, used for deduplication) - - SHA-1 (compatibility with legacy systems) - - MD5 (fast duplicate detection) - - CRC32 (checksum verification) - -2. **Deep Media Analysis** (via FFmpeg): - - Audio track details (language, codec, channels) - - Subtitle track details (language, format) - - Chapter information - - Embedded metadata (title, artist, etc.) - -**Characteristics**: -- **Concurrency**: 16 workers (CPU count) -- **I/O Pattern**: Full file read (streaming hash computation) -- **Processing Time**: Varies by file size (50MB/s typical throughput) -- **Storage**: Final metadata stored in etcd indexed by content hash -- **Memory**: Streaming hashing (fixed memory footprint regardless of file size) - -**Key Design Choice**: Hash computation is CPU and I/O intensive, so we limit concurrency to CPU count to avoid thrashing. The hash is computed in streaming fashion (read chunks → update hash → repeat) to maintain constant memory usage even for 100GB+ files. - -**Duplicate Detection**: Once a hash is computed, the system can: -- Detect exact duplicates (same hash) -- Detect version duplicates (same title/episode but different quality) -- Build virtual folder structure organized by content - ---- - -### Stage 4: Completion - -**Location**: `packages/meta-sort/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts::completeHashProcessing()` - -**Purpose**: Mark file as fully processed and finalize metadata. - -**Actions**: -1. Delete temporary metadata (`/tmp/{tempId}` key in etcd) -2. Store final metadata (`/file/{hash}` key in etcd) -3. Update virtual filesystem structure -4. Record performance metrics -5. Trigger duplicate detection (if enabled) - -**Final Metadata Structure**: -```typescript -{ - hash: "sha256:abc123...", // Primary content hash - filePath: "/data/watch/...", // Original file location - virtualPath: "/Anime/Series/S01/", // Virtual filesystem path - size: 1234567890, // File size in bytes - duration: 1440, // Duration in seconds - title: "Series Name", // Parsed title - season: 1, // Season number - episode: 5, // Episode number - resolution: "1920x1080", // Video resolution - codec: "h264", // Video codec - // ... many more fields -} -``` - ---- - -## Component Design - -### FolderWatcher (meta-hash) - -**File**: `packages/meta-hash/src/lib/folder-watcher/FolderWatcher.ts` - -**Responsibility**: Generic file discovery with no business logic. - -**Public API**: -```typescript -class FolderWatcher { - // Discover all files in directories (returns async generator) - async *discoverFiles(directories: string[]): AsyncGenerator - - // Watch directories for changes (using chokidar) - watch( - directories: string[], - callbacks: { - onAdd?: (filePath: string) => void | Promise; - onChange?: (filePath: string) => void | Promise; - onUnlink?: (filePath: string) => void | Promise; - }, - options?: WatchOptions - ): chokidar.FSWatcher -} -``` - -**Design Rationale**: -- **Generic**: No knowledge of video files, extensions, or Meta Mesh concepts -- **Reusable**: Can be used in other projects (e.g., backup tools, file indexers) -- **Minimal Dependencies**: Only depends on Node.js built-ins and chokidar -- **Error Handling**: Gracefully handles permission errors, missing directories, etc. - ---- - -### StreamingPipeline (meta-mesh) - -**File**: `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts` - -**Responsibility**: Orchestrate multi-stage file processing with application-specific logic. - -**Public API**: -```typescript -class StreamingPipeline { - // Start processing files from discovery stream - async start(discoveryStream: AsyncGenerator): Promise - - // Handle file change events - async handleFileAdded(filePath: string): Promise - async handleFileChanged(filePath: string): Promise - async handleFileDeleted(filePath: string): Promise - - // Get pipeline statistics - getStats(): PipelineStats -} -``` - -**Design Rationale**: -- **Application-Specific**: Knows about video files, subtitles, torrents -- **Configurable**: Concurrency levels, supported extensions, validation rules -- **Observable**: Provides real-time statistics via `getStats()` -- **Fire-and-Forget**: Stages don't block each other (errors logged but don't halt pipeline) - ---- - -### UnifiedProcessingStateManager (meta-mesh) - -**File**: `packages/meta-sort/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts` - -**Responsibility**: Track files through the 4-state processing lifecycle. - -**States**: -```typescript -type FileState = - | 'pending' // Discovered, waiting for validation - | 'lightProcessing' // Extracting metadata - | 'hashProcessing' // Computing content hash - | 'done' // Fully processed -``` - -**Public API**: -```typescript -class UnifiedProcessingStateManager { - addPending(filePath: string): void - startLightProcessing(filePath: string): void - completeLightProcessing(filePath: string, tempId?: string): void - completeHashProcessing(filePath: string, hash?: string): void - removeFile(filePath: string): void - getSnapshot(): UnifiedProcessingSnapshot -} -``` - -**Design Rationale**: -- **Centralized State**: Single source of truth for file processing status -- **Performance Metrics**: Records timestamps for each stage transition -- **Observability**: Provides snapshot API for monitoring UIs -- **Memory Efficient**: Limits history to last 100 completed files - ---- - -## Data Flow - -### Initial Scan (Cold Start) +States: `'discovered' | 'lightProcessing' | 'hashProcessing' | 'done'`. ``` -┌─────────────────────────────────────────────────────────────────────┐ -│ 1. FolderWatcher starts async generator │ -│ for await (const file of folderWatcher.discoverFiles([...])) │ -└────────────────────────────┬────────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────────┐ -│ 2. Discovery yields file paths one-by-one │ -│ "/data/watch/Anime/Series/S01E01.mkv" ────────┐ │ -│ "/data/watch/Anime/Series/S01E02.mkv" ────────┤ │ -│ "/data/watch/Movies/Film.mp4" ────────┤ │ -│ ... │ │ -└────────────────────────────────────────────────────┼────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────────┐ -│ 3. StreamingPipeline.start() consumes generator │ -│ - Validates each file (extension check) │ -│ - Adds to metadata queue (fire-and-forget) │ -│ - Logs progress every 1000 files │ -└────────────────────────────┬────────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────────┐ -│ 4. Validation Queue (32 workers) processes files │ -│ - Filter by extension (.mkv, .mp4, .srt, etc.) │ -│ - State: pending → ready for metadata │ -│ - Queue typically empty (keeps up with discovery) │ -└────────────────────────────┬────────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────────┐ -│ 5. Metadata Queue (16 workers) processes files │ -│ - Extract filename metadata (title, season, episode) │ -│ - Run FFmpeg for basic media info │ -│ - Store temp metadata in etcd (/tmp/{tempId}) │ -│ - State: lightProcessing │ -│ - Duration: ~470ms average │ -└────────────────────────────┬────────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────────┐ -│ 6. Hash Queue (16 workers) processes files │ -│ - Compute content hashes (SHA-256, SHA-1, MD5, CRC32) │ -│ - Deep FFmpeg analysis (all tracks, chapters) │ -│ - Delete temp metadata, store final metadata (/file/{hash}) │ -│ - Update virtual filesystem structure │ -│ - State: hashProcessing → done │ -│ - Duration: Varies by file size │ -└─────────────────────────────────────────────────────────────────────┘ -``` - -### File Change Event (Hot Path) - -``` -┌─────────────────────────────────────────────────────────────────────┐ -│ 1. chokidar detects file system change │ -│ - add: New file created │ -│ - change: Existing file modified │ -│ - unlink: File deleted │ -└────────────────────────────┬────────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────────┐ -│ 2. FolderWatcher invokes callback │ -│ onAdd(filePath) ──▶ pipeline.handleFileAdded() │ -│ onChange(filePath) ──▶ pipeline.handleFileChanged() │ -│ onUnlink(filePath) ──▶ pipeline.handleFileDeleted() │ -└────────────────────────────┬────────────────────────────────────────┘ - │ - ▼ -┌─────────────────────────────────────────────────────────────────────┐ -│ 3. Pipeline handles event │ -│ - Added: Process new file through validation queue │ -│ - Changed: Remove old metadata, reprocess through validation │ -│ - Deleted: Remove from state manager and etcd │ -└─────────────────────────────────────────────────────────────────────┘ -``` - ---- - -## State Management - -### File State Transitions - -``` -┌─────────┐ addPending() ┌────────────────┐ -│ PENDING ├────────────────────────▶│ LIGHT │ -└─────────┘ │ PROCESSING │ - ▲ └────────┬───────┘ - │ │ - │ removeFile() │ completeLightProcessing() - │ (on error) │ - │ ▼ - │ ┌────────────────┐ - │ │ HASH │ - └────────────────────────────────┤ PROCESSING │ - └────────┬───────┘ - │ - │ completeHashProcessing() +discovered ── startLightProcessing ──► lightProcessing │ + │ completeLightProcessing ▼ - ┌────────────────┐ - │ DONE │ - └────────────────┘ + hashProcessing ── completeHashProcessing ──► done ``` -### State Storage Locations +Notable details: -| State | In-Memory Map | etcd Key | Notes | -|-------|--------------|----------|-------| -| pending | ✅ `Map` | ❌ | Lightweight, no disk I/O | -| lightProcessing | ✅ `Map` | ✅ `/tmp/{tempId}` | Temp metadata for virtual FS | -| hashProcessing | ✅ `Map` | ✅ `/tmp/{tempId}` | Same as light processing | -| done | ✅ `Array` (last 100) | ✅ `/file/{hash}` | Final metadata, temp deleted | - -### State Metadata - -Each state tracks detailed timing information: - -```typescript -interface UnifiedFileState { - filePath: string; - state: FileState; - tempId?: string; // Temporary etcd key - hash?: string; // Final content hash (SHA-256) - virtualPath?: string; // Path in virtual filesystem - error?: string; // Error message if processing failed - - // Timestamps - discoveredAt?: number; - lightProcessingStartedAt?: number; - lightProcessingCompletedAt?: number; - hashProcessingStartedAt?: number; - hashProcessingCompletedAt?: number; - - // Processing times (milliseconds) - lightProcessingTime?: number; - hashProcessingTime?: number; - totalProcessingTime?: number; -} -``` - -This enables detailed performance analysis and bottleneck identification. +- `done` is kept as a bounded array (`maxDoneHistory = 100`); a running + `totalProcessedCount` survives that truncation. +- `seenHashes: Set` + `duplicateCount` track per-content duplicate + ingestion. +- `UnifiedFileState` tracks per-stage `*StartedAt` / `*CompletedAt` + timestamps and derived `*ProcessingTime` values; this is what the + `/api/processing/status` and `/api/processing/queue` endpoints expose. --- -## Concurrency Model +## Concurrency model -### Queue Configuration +Three independent `p-queue` instances inside `StreamingPipeline`: -The pipeline uses **PQueue** (p-queue) for concurrency control with stage-specific limits: +| Queue | Field | Configured by | +|-------|-------|---------------| +| Validation | `validationQueue` | `PipelineConfig.validationConcurrency` | +| Fast (light phase) | `fastQueue` | `PipelineConfig.fastQueueConcurrency` | +| Background (hash phase) | `backgroundQueue` | `PipelineConfig.backgroundQueueConcurrency` | -```typescript -// packages/meta-sort/packages/meta-sort-core/src/index.ts +Defaults are derived in `src/index.ts`: + +```ts const cpuCount = os.cpus().length; +const defaultBackgroundWorkers = Math.max(1, Math.floor(cpuCount / 2)); const validationConcurrency = config.MAX_WORKER_THREADS - ? config.MAX_WORKER_THREADS * 2 - : cpuCount * 2; // 32 on 16-core system - -const metadataConcurrency = config.MAX_WORKER_THREADS - || cpuCount; // 16 on 16-core system - -const hashConcurrency = config.MAX_WORKER_THREADS - || cpuCount; // 16 on 16-core system + ? config.MAX_WORKER_THREADS * 2 : cpuCount * 2; +const fastQueueConcurrency = config.MAX_WORKER_THREADS + ?? cpuCount; +const backgroundQueueConcurrency = config.MAX_WORKER_THREADS + ?? defaultBackgroundWorkers; ``` -### Concurrency Rationale +i.e. on a host with `os.cpus().length = N`: -| Stage | Concurrency | Rationale | -|-------|-------------|-----------| -| **Validation** | CPU × 2 | Pure CPU-bound string operations, no I/O. High concurrency ensures validation never blocks discovery. | -| **Metadata** | CPU × 1 | Mixed CPU/I/O workload. Spawns FFmpeg processes. One per core prevents context switching overhead. | -| **Hash** | CPU × 1 | CPU and I/O intensive (reads entire file). One per core maximizes throughput without thrashing. | +- `validationConcurrency = N * 2` +- `fastQueueConcurrency = N` +- `backgroundQueueConcurrency = floor(N / 2)` (minimum 1) -### Backpressure Handling +Setting `MAX_WORKER_THREADS=M` overrides all three to `(M * 2, M, M)`. -PQueue provides natural backpressure: +Note: container-plugin tasks do **not** share these queues — they are +scheduled inside `ContainerPluginScheduler` (`src/container-plugins/`), +whose own concurrencies come from `FAST_QUEUE_CONCURRENCY` (default 32) +and the same `backgroundQueueConcurrency` value as above. See +`plugin-task-queue-architecture.md`. -```typescript -// Fire-and-forget pattern with error handling -this.metadataQueue.add(() => this.extractMetadata(filePath)) - .catch(err => console.error(`[Pipeline] Metadata error:`, err.message)); -``` +### Pause / resume -- **Queue Size**: Number of pending tasks -- **Pending**: Number of tasks currently executing -- **Behavior**: When queue fills up, new tasks wait (no memory explosion) - -### Memory Characteristics - -For 900k files with 16 workers: - -``` -Memory Usage ≈ Queue Overhead + Worker Memory - -Queue Overhead: - - Validation queue: ~0 (keeps up with discovery) - - Metadata queue: ~100-500 tasks = ~50KB - - Hash queue: ~1000-5000 tasks = ~500KB - -Worker Memory: - - Metadata workers: 16 × ~50MB = 800MB - - Hash workers: 16 × ~100MB = 1.6GB - -Total: ~2.5GB (well within 16GB Node.js heap limit) -``` +`pipeline.pause()` / `pipeline.resume()` pause all three internal queues. +The API server wires these to operator controls (see +`UnifiedAPIServer.setStreamingPipeline`). In-flight tasks finish; no new +ones start until resume. --- -## Performance Characteristics +## Reset semantics (generation counter) -### Discovery Phase +`StreamingPipeline` keeps a `resetGeneration` counter and a per-file +`fileGenerations: Map`. Every time `reset()` is called +(triggered by a `reset` SSE event from meta-core, or by an operator action +that fans out to one), the counter increments and the queues are cleared. -**Throughput**: ~1000-5000 files/second (depends on filesystem) +Each long-running phase checks `fileGenerations.get(filePath) === +resetGeneration` at every yield point and bails out silently if the file +belongs to an older generation. This is the primary defense against: -**Bottleneck**: Filesystem I/O (directory reads) +- Old tasks completing into a cleared state manager. +- Container plugin callbacks arriving for files that were since + rescheduled (callback handler additionally verifies the hash matches the + current `hashProcessing` entry). -**Optimization**: Async generator yields to event loop, preventing blocking - -**Metrics** (from logs): -``` -[Pipeline] Progress: discovered=13000, validated=11937, metadata=1604, hash=0 -``` - -- 13,000 files discovered -- 11,937 validated (91% match extensions) -- 1,604 metadata extracted (12% complete) -- 0 hashes computed (still in queue) - -### Metadata Extraction Phase - -**Throughput**: ~34 files/second (16 workers ÷ 470ms average) - -**Bottleneck**: FFmpeg process spawning + basic media analysis - -**Optimization**: Limited concurrency prevents process thrashing - -**Average Times**: -- Filename parsing: ~5ms -- File stats: ~1ms -- FFmpeg basic info: ~460ms -- etcd write: ~4ms -- **Total**: ~470ms - -### Hash Computation Phase - -**Throughput**: Varies by file size - -| File Size | Hash Time | Throughput | -|-----------|-----------|------------| -| 100MB | ~2 seconds | 50MB/s | -| 1GB | ~20 seconds | 50MB/s | -| 10GB | ~200 seconds | 50MB/s | -| 50GB | ~1000 seconds | 50MB/s | - -**Bottleneck**: Disk read speed + SHA-256 computation - -**Optimization**: Streaming hash (constant memory), parallel workers - -### End-to-End Performance - -For a typical dataset (46,000 files, ~5TB total): - -| Metric | Time | -|--------|------| -| Discovery | ~30 seconds | -| First result | ~5 seconds | -| 50% complete | ~20 minutes | -| 100% complete | ~40 minutes | - -**Key Improvement**: Users see results in 5 seconds (vs. 15+ minutes in old batch system) +The reset also resets `discoveredCount`, `validatedCount`, `filteredCount`, +`fastProcessedCount`, `backgroundProcessedCount`, clears the +`ContainerPluginScheduler`, and resets `performanceMetrics`. --- -## Architecture Decisions +## Stats and observability -### Decision 1: Streaming vs. Batching +`StreamingPipeline.getStats()` returns: -**Problem**: Original batch processing had 15-minute discovery time before any processing started. +- counters: `discovered`, `validated`, `filtered`, `fastProcessed`, + `backgroundProcessed` +- per-queue `{ pending, size }` for validation, fast, background — if a + `ContainerPluginScheduler` is attached, the fast / background numbers + come from the scheduler's queues (those are the user-meaningful + numbers because that's where the plugin work actually runs) +- `state`: the full `UnifiedProcessingStateManager.getSnapshot()` -**Options Considered**: -1. **Optimize batch processing** (parallel directory traversal, caching) -2. **Hybrid approach** (batch discovery with parallel processing) -3. **Full streaming** (process as discovered) +These are exposed via the API endpoints registered in +`src/api/UnifiedAPIServer.ts`: -**Decision**: Full streaming (Option 3) - -**Rationale**: -- **User Experience**: Results appear within seconds -- **Memory Efficiency**: Bounded queue size prevents memory explosion -- **Simplicity**: Async generators are easier to reason about than batching logic -- **Scalability**: Works equally well for 100 files or 1 million files +- `GET /api/processing/status` — counts + state snapshot +- `GET /api/processing/queue` — per-queue depths +- `GET /api/processing/failed` — failed-file roster (also recorded into + `performanceMetrics`) +- `POST /api/processing/retry` and `POST /api/processing/retry-all` +- `POST /api/processing/wait-empty` — pauseless wait for the queues to + drain (useful in tests) --- -### Decision 2: Separate Generic and Application Logic +## Key code locations -**Problem**: Original `FolderWatcher` had hardcoded video file extensions, making it non-reusable. - -**Options Considered**: -1. **Keep extension filtering in FolderWatcher** (simple, but not reusable) -2. **Pass extension filter to FolderWatcher** (more generic, but clutters API) -3. **Remove all filtering from FolderWatcher** (cleanest separation) - -**Decision**: Remove all filtering (Option 3) - -**Rationale**: -- **Reusability**: `meta-hash` can be used for non-media file projects -- **Single Responsibility**: FolderWatcher does one thing (discover files) -- **Flexibility**: Application can implement arbitrary filtering logic -- **Performance**: Minimal overhead (extension check is < 1ms) - ---- - -### Decision 3: Three Separate Queues - -**Problem**: Single queue with uniform concurrency doesn't match workload characteristics. - -**Options Considered**: -1. **Single queue** (simple, but suboptimal) -2. **Two queues** (light + hash) -3. **Three queues** (validation + metadata + hash) - -**Decision**: Three queues (Option 3) - -**Rationale**: -- **Validation**: High concurrency (32) keeps up with discovery -- **Metadata**: Medium concurrency (16) balances FFmpeg overhead -- **Hash**: Medium concurrency (16) maximizes disk throughput -- **Independence**: Each stage can progress at its own pace -- **Observability**: Clear metrics per stage - ---- - -### Decision 4: Skip MIME Validation - -**Problem**: Extension checking might allow incorrect file types (e.g., `.mkv` file that's actually a text file). - -**Options Considered**: -1. **Always validate MIME type** (accurate, but slow) -2. **Optional MIME validation** (configurable, but complex) -3. **Skip MIME validation** (fast, handle errors in later stages) - -**Decision**: Skip MIME validation (Option 3) - -**Rationale**: -- **Performance**: MIME validation requires reading 4KB of each file (~25% overhead) -- **Accuracy**: Extension checking is 99.9% accurate in practice -- **Error Handling**: FFmpeg will fail gracefully on invalid files -- **User Control**: Users control watched directories (low risk of wrong extensions) - ---- - -### Decision 5: Fire-and-Forget Stage Progression - -**Problem**: Should errors in one stage halt processing or be logged and continue? - -**Options Considered**: -1. **Halt on error** (safe, but brittle) -2. **Retry with backoff** (robust, but complex) -3. **Log and continue** (fire-and-forget) - -**Decision**: Log and continue (Option 3) - -**Rationale**: -- **Resilience**: One corrupt file doesn't stop processing 899,999 good files -- **Observability**: Errors are logged with full context -- **Metrics**: Error count tracked in state manager -- **Manual Recovery**: Failed files can be manually reprocessed via API - ---- - -### Decision 6: Async Generator vs. Event Emitter - -**Problem**: How should FolderWatcher emit discovered files? - -**Options Considered**: -1. **Event Emitter** (traditional Node.js pattern) -2. **Callback** (simple, but nested callbacks) -3. **Async Generator** (modern, composable) - -**Decision**: Async Generator (Option 3) - -**Rationale**: -- **Backpressure**: Consumer controls consumption rate via `for await` -- **Composability**: Can be chained, filtered, transformed with async iterators -- **Simplicity**: No callback hell, no event listener management -- **Performance**: Natural event loop yielding prevents blocking - -```typescript -// ✅ Async Generator (Clean) -for await (const file of folderWatcher.discoverFiles([...])) { - await processFile(file); -} - -// ❌ Event Emitter (More boilerplate) -folderWatcher.on('file', async (file) => { - await processFile(file); -}); -folderWatcher.on('end', () => { - console.log('Discovery complete'); -}); -folderWatcher.start([...]); -``` - ---- - -## Appendix: Key Code Locations - -| Component | File Path | Lines | Description | -|-----------|-----------|-------|-------------| -| FolderWatcher | `packages/meta-hash/src/lib/folder-watcher/FolderWatcher.ts` | 170 | Generic file discovery | -| StreamingPipeline | `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts` | 237 | Multi-stage orchestrator | -| PipelineConfig | `packages/meta-sort/packages/meta-sort-core/src/logic/pipeline/PipelineConfig.ts` | 62 | Configuration interface | -| StateManager | `packages/meta-sort/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts` | 214 | State tracking | -| SupportedFileTypes | `packages/meta-sort/packages/meta-sort-core/src/config/SupportedFileTypes.ts` | 57 | Extension configuration | -| Main Entry | `packages/meta-sort/packages/meta-sort-core/src/index.ts` | 109 | Pipeline initialization | - ---- - -## Glossary - -- **Async Generator**: JavaScript function that returns an async iterable (uses `async function*` and `yield`) -- **Backpressure**: Flow control mechanism that prevents fast producers from overwhelming slow consumers -- **Event Loop**: Node.js single-threaded event processing loop (must not be blocked by long synchronous operations) -- **Fire-and-Forget**: Pattern where a function is called without awaiting its result (errors handled separately) -- **PQueue**: Third-party library for promise-based concurrency control (limits simultaneous async operations) -- **Streaming**: Processing data incrementally as it arrives (vs. batching, which waits for all data) - ---- - -## References - -- [Node.js Async Iterators](https://nodejs.org/api/esm.html#esm_async_iterators) -- [p-queue Documentation](https://github.com/sindresorhus/p-queue) -- [chokidar File Watching](https://github.com/paulmillr/chokidar) -- [FFmpeg Documentation](https://ffmpeg.org/documentation.html) -- [etcd Key-Value Store](https://etcd.io/docs/) +| Component | File | +|-----------|------| +| Pipeline orchestrator | `src/logic/pipeline/StreamingPipeline.ts` | +| Pipeline config | `src/logic/pipeline/PipelineConfig.ts` | +| State manager | `src/logic/UnifiedProcessingStateManager.ts` | +| File processor (light/hash phases) | `src/logic/WatchedFileProcessor.ts`, `src/logic/fileProcessor/` | +| SSE subscriber | `src/events/FileEventConsumer.ts` | +| SSE transport + cursor | `src/events/SSEEventClient.ts` | +| Container plugin dispatch | `src/container-plugins/ContainerPluginScheduler.ts` | +| Extension allow-list | `src/config/SupportedFileTypes.ts` | +| Concurrency defaults / startup wiring | `src/index.ts` | +| API endpoints | `src/api/UnifiedAPIServer.ts` | diff --git a/packages/async-utils/README.md b/packages/async-utils/README.md index 50a7d51..483ef0b 100644 --- a/packages/async-utils/README.md +++ b/packages/async-utils/README.md @@ -35,7 +35,7 @@ lazyValue().then(value => console.log(value)); ### Listener Cleaner ```typescript -import { ListenerCleaner } from "worph/async-utils"; +import { ListenerCleaner } from "@worph/async-utils"; const cleaner = new ListenerCleaner(); cleaner.add(() => console.log("Cleanup action"));