# Streaming Pipeline Architecture

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`.

---

## Contents

1. [Overview](#overview)
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

```
                  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
```

Two things are worth highlighting because they differ from older
documentation:

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.

---

## Event source: meta-core SSE

meta-sort no longer reads the filesystem directly. File discovery and
change watching live in meta-core, which streams events over HTTP SSE.

- **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.

Event types and their effect on the pipeline:

| 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

### Stage 1: Validation (`validateFile`)

`StreamingPipeline.validateFile(filePath, midhash256?)`

- 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))`.

### Stage 2: Fast queue (`processLightPhase`)

`StreamingPipeline.processLightPhase(filePath, midhash256?)` calls
`fileProcessor.processLightPhase(...)` which:

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.

After `processLightPhase` returns, `StreamingPipeline`:

- 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.

---

## State manager

`src/logic/UnifiedProcessingStateManager.ts`

States: `'discovered' | 'lightProcessing' | 'hashProcessing' | 'done'`.

```
discovered ── startLightProcessing ──► lightProcessing
                                              │
                                              │ completeLightProcessing
                                              ▼
                                       hashProcessing ── completeHashProcessing ──► done
```

Notable details:

- `done` is kept as a bounded array (`maxDoneHistory = 100`); a running
  `totalProcessedCount` survives that truncation.
- `seenHashes: Set<string>` + `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

Three independent `p-queue` instances inside `StreamingPipeline`:

| Queue | Field | Configured by |
|-------|-------|---------------|
| Validation | `validationQueue` | `PipelineConfig.validationConcurrency` |
| Fast (light phase) | `fastQueue` | `PipelineConfig.fastQueueConcurrency` |
| Background (hash phase) | `backgroundQueue` | `PipelineConfig.backgroundQueueConcurrency` |

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;
const fastQueueConcurrency = config.MAX_WORKER_THREADS
    ?? cpuCount;
const backgroundQueueConcurrency = config.MAX_WORKER_THREADS
    ?? defaultBackgroundWorkers;
```

i.e. on a host with `os.cpus().length = N`:

- `validationConcurrency = N * 2`
- `fastQueueConcurrency = N`
- `backgroundQueueConcurrency = floor(N / 2)` (minimum 1)

Setting `MAX_WORKER_THREADS=M` overrides all three to `(M * 2, M, M)`.

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`.

### Pause / resume

`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.

---

## Reset semantics (generation counter)

`StreamingPipeline` keeps a `resetGeneration` counter and a per-file
`fileGenerations: Map<string, number>`. 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.

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:

- 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).

The reset also resets `discoveredCount`, `validatedCount`, `filteredCount`,
`fastProcessedCount`, `backgroundProcessedCount`, clears the
`ContainerPluginScheduler`, and resets `performanceMetrics`.

---

## Stats and observability

`StreamingPipeline.getStats()` returns:

- 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()`

These are exposed via the API endpoints registered in
`src/api/UnifiedAPIServer.ts`:

- `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)

---

## Key code locations

| 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` |