8 Commits
Author SHA1 Message Date
Worph 33f576c92a beacon v2 discovery: meshdisco.ts on 239.255.99.1:9099, core located by metamesh.core cap, /api/neighbors ?cap=, nav lists services only
Build and Push Docker Image / build (push) Failing after 14s
2026-10-04 11:51:06 +00:00
Worph b4528a099a README: sync with current code 2026-10-03 11:52:00 +00:00
Worph 9280e23e83 improve search and fix languages
Build and Push Docker Image / build (push) Failing after 13s
2026-09-14 22:30:13 +02:00
Worph f5cafa77b2 replace file discovery by udp discovery
Build and Push Docker Image / build (push) Failing after 14s
2026-09-14 11:25:34 +02:00
Worph af51bbf943 change service discovery 2026-09-13 19:49:45 +02:00
Worph 985b000c24 align with metadata standard
Build and Push Docker Image / build (push) Failing after 14s
2026-09-13 15:56:54 +02:00
Worph 021991b00d improve stability
Build and Push Docker Image / build (push) Failing after 14s
2026-09-13 12:02:27 +02:00
Worph 862f73d26d cid rank remove
Build and Push Docker Image / build (push) Failing after 13s
2026-09-09 17:58:15 +02:00
28 changed files with 1707 additions and 1033 deletions
+99 -63
View File
@@ -3,17 +3,18 @@
File-watcher and metadata-extraction service for the MetaMesh ecosystem. File-watcher and metadata-extraction service for the MetaMesh ecosystem.
meta-sort consumes file events from meta-core (over SSE), drives metadata meta-sort consumes file events from meta-core (over SSE), drives metadata
through a streaming pipeline + container plugins, and writes results back 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 to Redis through meta-core's HTTP API. It exposes a Fastify REST API and
the `/files` volume over WebDAV so that other services and plugin reads media exclusively over **meta-core's WebDAV** — it mounts neither
containers can read media without mounting the same volume. `/meta-core` nor `/files`.
> **Service role:** meta-sort is **Redis-consuming**, not Redis-managing. > **Service role:** meta-sort is **Redis-consuming**, not Redis-managing.
> It does not run leader election, it does not spawn Redis, and it does > It does not run leader election, it does not spawn Redis, and it has
> not own `kv-leader.lock`. Those responsibilities belong to **meta-core** > no direct Redis connection: every read and write goes through
> (a separate Go service). meta-sort reads > **meta-core** (a separate Go service) over HTTP. meta-sort locates
> `/meta-core/locks/kv-leader.info` via `LeaderClient` to discover the > meta-core over UDP multicast (**beacon v2**,
> Redis URL, API URL, and WebDAV URLs, and registers its own URLs into > `239.255.99.1:9099`) — meta-core's announce carries its API and WebDAV
> `/meta-core/services/<hostname>.json`. > URLs — and announces itself on the same group so sibling dashboards can
> list it. `META_CORE_URL` pins meta-core and disables wire selection.
--- ---
@@ -29,14 +30,18 @@ What this service does:
3. **Manages container plugins**: spawns plugin containers via the Docker 3. **Manages container plugins**: spawns plugin containers via the Docker
socket, exposes a callback endpoint, lets plugins write metadata back socket, exposes a callback endpoint, lets plugins write metadata back
into Redis via the meta-core-compatible `/meta/*` API on this service. into Redis via the meta-core-compatible `/meta/*` API on this service.
4. **Serves files over WebDAV** at `/webdav/*` so container plugins and 4. **Computes file identity**: `midhash256` in the light phase, full-file
sibling services (meta-fuse, meta-stremio) can read media without digests (sha1, sha256, md5, sha3, crc32, btih v2) in the background
needing the underlying bind/SMB/rclone mounts. phase. Digests are persisted only as the bare-CID key-set
(`cids/<cid>`), never as per-algorithm `cid_*` fields
(`src/config/TargetHash.ts`, `src/kv/MetaCoreApiWriter.ts`).
What this service does **not** do (anymore): What this service does **not** do (anymore):
- It does not run leader election. - It does not run leader election.
- It does not spawn or supervise Redis. - It does not spawn or supervise Redis, nor talk to it directly.
- It does not serve WebDAV — files are read from meta-core's WebDAV, and
plugin containers get that same URL (`WEBDAV_URL`).
- It does not watch the filesystem directly (chokidar lives in meta-core). - 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 manage remote mounts (mounts are now in meta-core / rclone-smb).
- It does not have an in-process plugin runtime — all plugins are - It does not have an in-process plugin runtime — all plugins are
@@ -48,43 +53,46 @@ What this service does **not** do (anymore):
## Architecture ## Architecture
### Two-volume layout ### Volumes
meta-sort consumes two host-side volumes via standard bind mounts: meta-sort needs **no shared volume**. Media lives on meta-core's
`FILES_VOLUME` and is read over meta-core's WebDAV; file paths stored in
Redis are **relative to `/files`**, which keeps storage portable across
hosts. The only mounts are its own cache (`/data/cache`), the Docker
socket (for plugin containers) and `plugins.yml`.
| Volume | Container path | Owner | Purpose | ### Locating meta-core (beacon v2)
|--------|----------------|-------|---------|
| `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 |
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 (separate container) meta-core (separate container)
├── runs Redis ├── runs Redis, the file watcher, and the WebDAV server for /files
├── runs file watcher (chokidar) and publishes file:events ├── HTTP API: /urls, /api/events/files (SSE), /meta/*, ...
├── runs HTTP API on /api/urls, /api/events/files, /meta/*, /api/services, ... └── advertises metamesh.core on 239.255.99.1:9099 (UDP multicast),
└── writes /meta-core/locks/kv-leader.info carrying its /urls payload (apiUrl, webdavUrl, webdavUrlInternal)
│ │
▼ ▼
meta-sort (this service) meta-sort (this service)
├── LeaderClient reads kv-leader.info → Redis URL, API URL, ├── LeaderClient → MetaCoreLocator (src/discovery/meshdisco.ts)
│ WebDAV URLs, etc. │ picks up the core announce (or GET {META_CORE_URL}/urls
├── KVManager creates RedisKVClient (writes are routed │ when pinned) and announces meta-sort on the same group
│ through meta-core's HTTP API for the api-mediated-access ├── KVManager builds an HTTP-only RedisKVClient against
│ model) │ meta-core's API and configures the WebDAV client
├── ServiceRegistration writes ├── GET /api/neighbors serves the UDP neighbour map (nav menu)
│ /meta-core/services/meta-sort-<hostname>.json
└── FileEventConsumer subscribes to └── FileEventConsumer subscribes to
${apiUrl}/api/events/files (SSE) ${apiUrl}/api/events/files (SSE)
``` ```
Source: `packages/meta-sort-core/src/kv/{KVManager,LeaderClient,RedisClient,ServiceRegistration}.ts`, Source: `packages/meta-sort-core/src/discovery/meshdisco.ts`,
`packages/meta-sort-core/src/kv/{KVManager,LeaderClient,RedisClient,MetaCoreApiWriter}.ts`,
`packages/meta-sort-core/src/events/{FileEventConsumer,SSEEventClient}.ts`. `packages/meta-sort-core/src/events/{FileEventConsumer,SSEEventClient}.ts`.
The protocol spec is `docs/project-architecture/beacon-v2.md` in
the meta-root. The `meshdisco.ts` port is mirrored across services and
guarded by the meta-root's `scripts/check-mirrors.sh` — change it there in
lockstep.
The SSE cursor is persisted to `/meta-core/cursors/meta-sort-files.cursor`
(`FileEventConsumer.ts` default); since `/meta-core` is no longer mounted,
that path is container-local and does not survive a container recreate.
### Processing pipeline ### Processing pipeline
@@ -119,7 +127,9 @@ them via the Docker socket and a callback endpoint:
- **Built-in plugins** (each is a separate git submodule under - **Built-in plugins** (each is a separate git submodule under
`packages/plugins/metamesh-plugin-*` at the repo root): `file-info`, `packages/plugins/metamesh-plugin-*` at the repo root): `file-info`,
`ffmpeg`, `filename-parser`, `jellyfin-nfo`, `tmdb`, `anime-detector`, `ffmpeg`, `filename-parser`, `jellyfin-nfo`, `tmdb`, `anime-detector`,
`language`, `subtitle`, `subtitle-extractor`, `torrent`, `fullhash`. `language`, `subtitle`, `subtitle-extractor`, `still-extractor`,
`opensubtitle`, `torrent`, `fullhash`. Which ones actually run is
decided by `plugins.yml` — the image ships none.
- **Dispatch:** `ContainerPluginScheduler.dispatchAllPlugins(hashId, - **Dispatch:** `ContainerPluginScheduler.dispatchAllPlugins(hashId,
filePath, metaFlat)` — fire-and-forget. The scheduler keeps its own filePath, metaFlat)` — fire-and-forget. The scheduler keeps its own
fast / background queues based on each plugin manifest's fast / background queues based on each plugin manifest's
@@ -128,8 +138,10 @@ them via the Docker socket and a callback endpoint:
`/api/plugins/callback`, and they can read/write Redis metadata `/api/plugins/callback`, and they can read/write Redis metadata
through the meta-core-compatible `/meta/:hash` routes that meta-sort through the meta-core-compatible `/meta/:hash` routes that meta-sort
also exposes (so plugins only need one HTTP target). also exposes (so plugins only need one HTTP target).
- **Plugins access files** by hitting meta-sort's WebDAV - **Plugins access files** over meta-core's WebDAV: each plugin
(`http://metasort-app/webdav/...` inside the docker network). container is started with `WEBDAV_URL` (meta-core's
`webdavUrlInternal`), `META_CORE_URL` (= `CONTAINER_META_CORE_URL`),
`CALLBACK_URL` and `PLUGIN_ID` (`ContainerManager.ts`).
For the plugin HTTP contract, see For the plugin HTTP contract, see
[`docs/containerized-plugin-architecture.md`](docs/containerized-plugin-architecture.md) [`docs/containerized-plugin-architecture.md`](docs/containerized-plugin-architecture.md)
@@ -143,15 +155,16 @@ and [`docs/plugin-task-queue-architecture.md`](docs/plugin-task-queue-architectu
| Variable | Default | Description | | Variable | Default | Description |
|----------|---------|-------------| |----------|---------|-------------|
| `META_CORE_PATH` | `/meta-core` | Mount of META_CORE_VOLUME. Required — startup fails fast if absent. | | `META_CORE_URL` | — | Pins meta-core's API URL; UDP discovery then never overrides it. |
| `FILES_PATH` | `/files` | Mount of FILES_VOLUME. | | `ENABLE_UDP_DISCOVERY` | on | `false`/`0` disables the beacon v2 announce/listen. |
| `ALLOW_LEGACY_REDIS_URL` | — | `1` downgrades the "meta-core still publishes `redisUrl`" startup error to a warning. |
| `FILES_PATH` | `/files` | Virtual root that relative paths resolve under (on meta-core's WebDAV). |
| `CACHE_FOLDER_PATH` | `/data/cache` | Local cache root. | | `CACHE_FOLDER_PATH` | `/data/cache` | Local cache root. |
| `SERVICE_NAME` | `meta-sort` | Identifier used for service registration. | | `SERVICE_NAME` | `meta-sort` | Name announced on the discovery group. |
| `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_PORT` | `3000` | HTTP API port (legacy name; this is the Fastify API). |
| `FUSE_API_HOST` | `0.0.0.0` | API bind host. | | `FUSE_API_HOST` | `0.0.0.0` | API bind host. |
| `BASE_URL` | — | External base URL recorded in service registration. | | `PUBLIC_URL` | — | Browser-facing URL announced for neighbours' nav menus. Wins over `BASE_URL` (use it for a debug-direct port when no Caddy is in front). |
| `ADVERTISE_HOST` | — | Hostname to advertise to other services. Falls back to auto-detected IP. | | `BASE_URL` | — | Perimeter URL announced when `PUBLIC_URL` is unset; falls back to `http://<hostname>:<FUSE_API_PORT>`. |
| `MAX_WORKER_THREADS` | `os.cpus().length` | If set, overrides all pipeline concurrencies to `(M*2, M, M)`. | | `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). | | `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_LIGHT_SLOTS` | `16` | Max files concurrently in the light phase. |
@@ -160,16 +173,22 @@ and [`docs/plugin-task-queue-architecture.md`](docs/plugin-task-queue-architectu
| `CONTAINER_PLUGINS_CONFIG` | `/app/plugins.yml` | Path to plugin manifest YAML. If missing, container plugins are skipped. | | `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. | | `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_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_META_CORE_URL` | `http://meta-sort` | URL plugins use for meta-core-compatible `/meta/*` routes (served by meta-sort itself). The default host doesn't resolve on current stacks — dev sets `http://metasort-app` (port 80 dev nginx proxies `/meta/`), the store app `http://metasort-app:3000` (the stock nginx has no `/meta/` location). |
| `CONTAINER_NETWORK` | `meta-network` | Docker network to attach plugin containers to. | | `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_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. | | `PLUGIN_CACHE_HOST_PATH` | — | Host path bind-mounted into each plugin for persistent caches. |
Source of truth: `packages/meta-sort-core/src/config/EnvConfig.ts`. Source of truth: `packages/meta-sort-core/src/config/EnvConfig.ts` (plus
`PUBLIC_URL` in `kv/KVManager.ts` and `ENABLE_UDP_DISCOVERY` /
`ALLOW_LEGACY_REDIS_URL` in `kv/LeaderClient.ts`). `META_CORE_PATH`,
`REDIS_URL`, `ADVERTISE_HOST` and `SERVICE_VERSION` are still parsed by
`EnvConfig.ts` but nothing acts on them any more. The discovery group
defaults to the protocol endpoint (`239.255.99.1:9099`, 10 s interval);
`BEACON_GROUP` / `BEACON_PORT` / `BEACON_INTERVAL_MS` override it.
The Redis URL and WebDAV URLs are **not** configured via environment The WebDAV URLs are **not** configured via environment variables —
variables — meta-sort discovers them from `kv-leader.info` (or its meta-sort takes them from meta-core's announce (or `GET /urls` when
in-memory equivalent fetched through meta-core's `/api/urls`). pinned).
### Concurrency defaults ### Concurrency defaults
@@ -239,7 +258,9 @@ Defined in `packages/meta-sort-core/src/api/UnifiedAPIServer.ts`.
| `POST /api/plugins/callback` | Container plugin task-complete callback (used by plugins, not humans). | | `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` | Container-plugin status (health, instances, queue depths). |
| `GET /api/plugins/containers/:pluginId/manifest` | Plugin manifest. | | `GET /api/plugins/containers/:pluginId/manifest` | Plugin manifest. |
| `GET /api/plugins/containers/:pluginId/logs` | Container logs (`?instance=`, `?tail=`). |
| `POST /api/plugins/containers/:pluginId/restart` | Restart instances. | | `POST /api/plugins/containers/:pluginId/restart` | Restart instances. |
| `POST /api/plugins/containers/:pluginId/stop` / `start` | Stop / start instances. |
| `POST /api/plugins/containers/restart-all` | Restart all. | | `POST /api/plugins/containers/restart-all` | Restart all. |
| `POST /api/plugins/containers` | Add a new plugin (image, instances, resources, config, defaultQueue). | | `POST /api/plugins/containers` | Add a new plugin (image, instances, resources, config, defaultQueue). |
| `PUT /api/plugins/containers/:pluginId` | Update plugin spec. | | `PUT /api/plugins/containers/:pluginId` | Update plugin spec. |
@@ -249,9 +270,8 @@ Defined in `packages/meta-sort-core/src/api/UnifiedAPIServer.ts`.
| Endpoint | Description | | Endpoint | Description |
|----------|-------------| |----------|-------------|
| `GET /api/file/download?path=...` | Download a file via the configured WebDAV backend. | | `GET /api/file/download?path=...` | Download a file via meta-core's WebDAV. |
| `POST /file/cid` | Compute the CID (midhash256) for a path (`{ path }`). | | `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) ### meta-core-compatible KV (for container plugins)
@@ -272,8 +292,8 @@ single host. They are only registered when the KV client is up.
| Endpoint | Description | | Endpoint | Description |
|----------|-------------| |----------|-------------|
| `GET /api/metrics` | Performance metrics. | | `GET /api/metrics` | Performance metrics. |
| `GET /api/stats` | Redis stats summary (only when KV is up). | | `GET /api/stats` | File count + total size from meta-core's `/api/files/tuples` summary, cached stale-while-revalidate. |
| `GET /api/services` | Proxy to meta-core's `/api/services` via the LeaderClient. | | `GET /api/neighbors` | beacon v2 neighbour map (nav menu). Served locally — works while meta-core is down. Replaces the removed `/api/services`. |
--- ---
@@ -286,6 +306,18 @@ single host. They are only registered when the KV client is up.
- Debug-direct backend (no auth, no TLS): `http://localhost:18180` - Debug-direct backend (no auth, no TLS): `http://localhost:18180`
- Container name (backend): `metasort-app` - Container name (backend): `metasort-app`
- Container name (hash-lock proxy): `metasort` - Container name (hash-lock proxy): `metasort`
- CasaOS store app: `MetaSort` in the meta-root's `packages/MetaAppStore`
(`ghcr.io/worph/meta-sort`, fronted by an `appshield` perimeter).
### Docker image
The repo `Dockerfile` (published to `ghcr.io/worph/meta-sort` by
`.github/workflows/docker-publish.yml`) is an all-in-one image: nginx on
`:80` → Fastify on `:3000`, rclone, and a bundled `meta-core` binary
copied from `ghcr.io/worph/meta-core` and started by
`docker/supervisord.conf`. Deployments run meta-core as its own app, so
both the dev stack and the store app mount their own supervisord/nginx
configs that drop the bundled meta-core.
### Logs ### Logs
@@ -310,7 +342,8 @@ cd dev
### Running tests ### Running tests
```bash ```bash
docker exec meta-test-runner /app/test/test.sh sort docker exec meta-test-runner /app/test/test.sh sort # E2E (meta-root dev stack)
pnpm test # unit tests (@meta-sort/core, mocha)
``` ```
See the root `CLAUDE.md` "Running Tests" section for the full matrix. See the root `CLAUDE.md` "Running Tests" section for the full matrix.
@@ -336,9 +369,10 @@ packages/meta-sort/
│ │ │ ├── api/ # UnifiedAPIServer.ts (Fastify) │ │ │ ├── api/ # UnifiedAPIServer.ts (Fastify)
│ │ │ ├── config/ # EnvConfig.ts, SupportedFileTypes.ts │ │ │ ├── config/ # EnvConfig.ts, SupportedFileTypes.ts
│ │ │ ├── container-plugins/# ContainerManager, ContainerPluginScheduler, ... │ │ │ ├── container-plugins/# ContainerManager, ContainerPluginScheduler, ...
│ │ │ ├── discovery/ # meshdisco.ts (beacon v2, mirrored port)
│ │ │ ├── events/ # FileEventConsumer, SSEEventClient │ │ │ ├── events/ # FileEventConsumer, SSEEventClient
│ │ │ ├── jellyfin/ # Jellyfin / NFO output helpers │ │ │ ├── jellyfin/ # Jellyfin / NFO output helpers
│ │ │ ├── kv/ # KVManager, LeaderClient, RedisClient │ │ │ ├── kv/ # KVManager, LeaderClient, RedisClient, MetaCoreApiWriter
│ │ │ ├── logic/ # StreamingPipeline, state managers, fileProcessor │ │ │ ├── logic/ # StreamingPipeline, state managers, fileProcessor
│ │ │ ├── metrics/ # PerformanceMetrics │ │ │ ├── metrics/ # PerformanceMetrics
│ │ │ ├── plugin-engine/ # Vestigial in-process types + TaskScheduler glue │ │ │ ├── plugin-engine/ # Vestigial in-process types + TaskScheduler glue
@@ -350,27 +384,28 @@ packages/meta-sort/
│ ├── meta-sort-ui/ # @meta-sort/ui - React dashboard (Vite) │ ├── meta-sort-ui/ # @meta-sort/ui - React dashboard (Vite)
│ ├── meta-sort-editor/ # Metadata editor UI (Vite) │ ├── meta-sort-editor/ # Metadata editor UI (Vite)
│ ├── async-utils/ # @worph/async-utils - MultiQueue, etc. │ ├── async-utils/ # @worph/async-utils - MultiQueue, etc.
│ └── shared/ # Symlinks to filename-tool, meta-hash, meta-interface │ └── shared/ # Vendored copies of filename-tool, meta-hash, meta-interface
├── pnpm-workspace.yaml ├── pnpm-workspace.yaml
└── pnpm-lock.yaml └── pnpm-lock.yaml
``` ```
> Built-in container plugins **do not live here.** They live at the > Built-in container plugins **do not live here.** They live at the
> repo root as separate submodules: > 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}`. > `packages/plugins/metamesh-plugin-{file-info,ffmpeg,filename-parser,jellyfin-nfo,tmdb,anime-detector,language,subtitle,subtitle-extractor,still-extractor,opensubtitle,torrent,fullhash}`.
### Key components ### Key components
| Component | File | Purpose | | Component | File | Purpose |
|-----------|------|---------| |-----------|------|---------|
| Entry point | `src/index.ts` | Wires KVManager → API server → plugin manager → ContainerManager → StreamingPipeline → FileEventConsumer. | | Entry point | `src/index.ts` | Wires KVManager → API server → plugin manager → ContainerManager → StreamingPipeline → FileEventConsumer. |
| Discovery | `src/discovery/meshdisco.ts` | beacon v2 node + `MetaCoreLocator`. |
| Streaming pipeline | `src/logic/pipeline/StreamingPipeline.ts` | 3-queue (validation / fast / background) orchestrator. | | Streaming pipeline | `src/logic/pipeline/StreamingPipeline.ts` | 3-queue (validation / fast / background) orchestrator. |
| State manager | `src/logic/UnifiedProcessingStateManager.ts` | Tracks `discovered → lightProcessing → hashProcessing → done`. | | State manager | `src/logic/UnifiedProcessingStateManager.ts` | Tracks `discovered → lightProcessing → hashProcessing → done`. |
| File processor | `src/logic/WatchedFileProcessor.ts` | Light + hash phase implementation. | | File processor | `src/logic/WatchedFileProcessor.ts` | Light + hash phase implementation. |
| SSE consumer | `src/events/FileEventConsumer.ts` | Translates meta-core events into pipeline calls. | | 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. | | SSE transport | `src/events/SSEEventClient.ts` | Long-lived HTTP SSE with cursor persistence. |
| KV manager | `src/kv/KVManager.ts` | Bootstraps RedisKVClient + ServiceRegistration from LeaderClient. | | KV manager | `src/kv/KVManager.ts` | Bootstraps the HTTP-only RedisKVClient + WebDAV client from LeaderClient. |
| Leader client | `src/kv/LeaderClient.ts` | Reads `/meta-core/locks/kv-leader.info`, fetches `/api/urls`. | | Leader client | `src/kv/LeaderClient.ts` | Legacy name: thin adapter over `MetaCoreLocator` (UDP), or `GET /urls` when `META_CORE_URL` is pinned. |
| Container manager | `src/container-plugins/ContainerManager.ts` | Docker socket orchestration. | | Container manager | `src/container-plugins/ContainerManager.ts` | Docker socket orchestration. |
| Container scheduler | `src/container-plugins/ContainerPluginScheduler.ts` | Plugin fast/background queues + callbacks. | | Container scheduler | `src/container-plugins/ContainerPluginScheduler.ts` | Plugin fast/background queues + callbacks. |
| API server | `src/api/UnifiedAPIServer.ts` | Fastify REST + WebDAV-facing meta routes. | | API server | `src/api/UnifiedAPIServer.ts` | Fastify REST + WebDAV-facing meta routes. |
@@ -393,8 +428,9 @@ Repo-level docs of interest:
- Root `CLAUDE.md` — authoritative ground truth on the dev stack, ports, - Root `CLAUDE.md` — authoritative ground truth on the dev stack, ports,
containers, and operational commands. containers, and operational commands.
- `/METADATA_KEYS.md` — schema for the `/file/{cid}` Redis hash. - `/METADATA_KEYS.md` — schema for the `/file/{cid}` Redis hash.
- `packages/meta-core/docs/` — leader election and Redis storage owner. - `packages/meta-core/docs/` — Redis storage owner, api-mediated access.
- `docs/project-architecture/beacon-v2.md` — beacon v2 spec.
## License ## License
MIT — see LICENSE. MIT (see `package.json`).
@@ -11,8 +11,9 @@ import { performanceMetrics } from '../metrics/PerformanceMetrics.js';
import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js'; import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js';
import type { IKVClient } from '../kv/IKVClient.js'; import type { IKVClient } from '../kv/IKVClient.js';
import type { KVManager } from '../kv/KVManager.js'; import type { KVManager } from '../kv/KVManager.js';
import type { ExtendedProcessingSnapshot, IKVClientWithPubSub } from '../types/ExtendedInterfaces.js'; import type { ExtendedProcessingSnapshot } from '../types/ExtendedInterfaces.js';
import { getErrorMessage } from '../types/ExtendedInterfaces.js'; import { getErrorMessage, hasFileTuples } from '../types/ExtendedInterfaces.js';
import type { IKVClientWithFileTuples } from '../types/ExtendedInterfaces.js';
import { config } from '../config/EnvConfig.js'; import { config } from '../config/EnvConfig.js';
import * as webdav from '../webdav/WebdavClient.js'; import * as webdav from '../webdav/WebdavClient.js';
import type { PluginManager } from '../plugin-engine/PluginManager.js'; import type { PluginManager } from '../plugin-engine/PluginManager.js';
@@ -29,6 +30,45 @@ export interface UnifiedAPIServerConfig {
enableCors?: boolean; enableCors?: boolean;
} }
/** File totals served by /api/stats. Null when meta-core cannot supply them. */
interface FileStats {
fileCount: number | null;
totalSize: number | null;
source: 'meta-core' | 'unavailable';
}
/** Metadata reads in flight at once while a recompute builds its task list. */
const RECOMPUTE_READ_CONCURRENCY = 8;
/** Run `worker` over `items` with at most `limit` calls in flight. */
async function forEachWithConcurrency<T>(
items: readonly T[],
limit: number,
worker: (item: T) => Promise<void>
): Promise<void> {
let next = 0;
const runners = Array.from({ length: Math.min(limit, items.length) }, async () => {
while (next < items.length) {
const item = items[next++];
await worker(item);
}
});
await Promise.all(runners);
}
/** Flatten nested metadata into the `a/b/c` keys the task scheduler expects. */
function flattenMetadata(obj: Record<string, unknown>, out: Record<string, string> = {}, prefix = ''): Record<string, string> {
for (const [key, value] of Object.entries(obj)) {
const fullKey = prefix ? `${prefix}/${key}` : key;
if (value !== null && typeof value === 'object' && !Array.isArray(value)) {
flattenMetadata(value as Record<string, unknown>, out, fullKey);
} else if (value !== undefined && value !== null) {
out[fullKey] = String(value);
}
}
return out;
}
export class UnifiedAPIServer { export class UnifiedAPIServer {
private app: FastifyInstance; private app: FastifyInstance;
private unifiedStateManager: UnifiedProcessingStateManager | null = null; private unifiedStateManager: UnifiedProcessingStateManager | null = null;
@@ -44,9 +84,15 @@ export class UnifiedAPIServer {
private containerPluginScheduler: ContainerPluginScheduler | null = null; private containerPluginScheduler: ContainerPluginScheduler | null = null;
private streamingPipeline: StreamingPipeline | null = null; private streamingPipeline: StreamingPipeline | null = null;
// Cache for total file size (avoid running du on every request) // /api/stats file totals: stale-while-revalidate, one refresh at a time.
private totalSizeCache: { value: number; timestamp: number } | null = null; // Never walk hashIds for these — on a shared meta-core that is every gateway
private static TOTAL_SIZE_CACHE_TTL_MS = 30000; // 30 seconds // record too (82k on watch.nsl.sh), and a sweep per poll stacked unbounded.
private fileStats: { value: FileStats; fetchedAt: number } | null = null;
private fileStatsInFlight: Promise<FileStats> | null = null;
private static FILE_STATS_TTL_MS = 30000; // 30 seconds
// Plugins whose recompute is currently being queued (single-flight per plugin)
private recomputeInFlight = new Set<string>();
constructor( constructor(
config: UnifiedAPIServerConfig = {}, config: UnifiedAPIServerConfig = {},
@@ -97,48 +143,46 @@ export class UnifiedAPIServer {
} }
/** /**
* Get total size of processed files from Redis sizeByte field (cached) * File count + total size of the files meta-core holds on disk, for the
* This is "eventually correct" - accumulates as files are processed by the file-info plugin * monitor's "Total Size" card: one summary call to meta-core's
* GET /api/files/tuples (batched server-side), cached for FILE_STATS_TTL_MS.
*
* Stale-while-revalidate with a single in-flight refresh — a poll never waits
* on meta-core once a value exists, and concurrent polls share one request.
*/ */
private async getTotalFileSizeFromRedis(): Promise<number> { private async getFileStats(): Promise<FileStats> {
// Check cache const cached = this.fileStats;
const now = Date.now(); if (cached && Date.now() - cached.fetchedAt < UnifiedAPIServer.FILE_STATS_TTL_MS) {
if (this.totalSizeCache && (now - this.totalSizeCache.timestamp) < UnifiedAPIServer.TOTAL_SIZE_CACHE_TTL_MS) { return cached.value;
return this.totalSizeCache.value;
} }
// Sum sizeByte from all file entries in Redis if (!this.fileStatsInFlight) {
let totalSize = 0; this.fileStatsInFlight = this.fetchFileStats()
.catch((error: unknown): FileStats => {
if (!this.kvClient) { console.warn('[API] /api/stats: meta-core file totals unavailable:', getErrorMessage(error));
return 0; return cached?.value ?? { fileCount: null, totalSize: null, source: 'unavailable' };
})
.then((value) => {
this.fileStats = { value, fetchedAt: Date.now() };
return value;
})
.finally(() => {
this.fileStatsInFlight = null;
});
} }
try { return cached ? cached.value : this.fileStatsInFlight;
// Get all file hash IDs
const hashIds = await this.kvClient.getAllHashIds();
// Sum up sizeByte values (batch process for efficiency)
for (const hashId of hashIds) {
try {
const sizeByte = await this.kvClient.getMetadata(hashId, 'sizeByte');
if (sizeByte !== null && sizeByte !== undefined) {
const size = typeof sizeByte === 'number' ? sizeByte : parseInt(String(sizeByte), 10);
if (!isNaN(size) && size > 0) {
totalSize += size;
}
}
} catch {
// Skip entries without sizeByte
}
}
} catch (error) {
console.warn('[API] Failed to calculate total size from Redis:', error);
} }
// Update cache private async fetchFileStats(): Promise<FileStats> {
this.totalSizeCache = { value: totalSize, timestamp: now }; if (!this.kvClient || !hasFileTuples(this.kvClient)) {
return totalSize; return { fileCount: null, totalSize: null, source: 'unavailable' };
}
const summary = await this.kvClient.getFileTuples({ summary: true });
if (!summary) {
return { fileCount: null, totalSize: null, source: 'unavailable' };
}
return { fileCount: summary.count, totalSize: summary.totalSize, source: 'meta-core' };
} }
/** /**
@@ -193,7 +237,7 @@ export class UnifiedAPIServer {
this.setupStatsRoutes(); this.setupStatsRoutes();
} }
// Service discovery — proxies to meta-core's /api/services using the // Service discovery — beacon v2 neighbour list, served from the
// LeaderClient to resolve meta-core's current URL dynamically. // LeaderClient to resolve meta-core's current URL dynamically.
this.setupServiceDiscoveryRoutes(); this.setupServiceDiscoveryRoutes();
@@ -524,53 +568,17 @@ export class UnifiedAPIServer {
/** /**
* Setup Stats API routes (Redis stats for monitor UI) * Setup Stats API routes (file totals for the monitor UI)
*/ */
private setupStatsRoutes(): void { private setupStatsRoutes(): void {
if (!this.kvClient) { if (!this.kvClient) {
return; return;
} }
// Get Redis/KV stats (for monitor UI) // At most one meta-core call per FILE_STATS_TTL_MS however often the
this.app.get('/api/stats', async (request, reply) => { // monitor polls — see getFileStats.
try { this.app.get('/api/stats', async () => {
const hashIds = await this.kvClient!.getAllHashIds(); return this.getFileStats();
const redis = (this.kvClient as IKVClientWithPubSub).getRedisClient?.() as { info: (section: string) => Promise<string> } | undefined;
let memoryUsage = 'N/A';
let memoryUsageBytes = 0;
if (redis) {
try {
const info = await redis.info('memory');
const match = info.match(/used_memory_human:(\S+)/);
const bytesMatch = info.match(/used_memory:(\d+)/);
if (match) {
memoryUsage = match[1];
}
if (bytesMatch) {
memoryUsageBytes = parseInt(bytesMatch[1], 10);
}
} catch {
// Ignore memory info errors
}
}
// Get total size from Redis sizeByte (eventually correct as files are processed)
const totalSize = await this.getTotalFileSizeFromRedis();
return {
fileCount: hashIds.length,
keyCount: hashIds.length,
totalSize,
memoryUsage,
memoryUsageBytes
};
} catch (error: any) {
return reply.status(500).send({
error: 'Failed to get stats',
details: error.message
});
}
}); });
} }
@@ -584,33 +592,31 @@ export class UnifiedAPIServer {
} }
/** /**
* Service discovery — proxies to meta-core's /api/services, resolving the * Service discovery — beacon v2 (docs/project-architecture/beacon-v2.md).
* current leader's URL via LeaderClient (no hardcoded hostname). * Served from this service's own beacon view; no meta-core round trip, so
* the nav survives a core outage.
*/ */
private setupServiceDiscoveryRoutes(): void { private setupServiceDiscoveryRoutes(): void {
this.app.get('/api/services', async (_request, reply) => { // beacon v2: neighbours heard over UDP, served from this service's own
try { // view. Unlike /api/services below it needs no meta-core, so the nav still
// renders when the core is down. `?all=1` returns every instance,
// `?cap=<pattern>` filters by capability. `services` is kept as an alias
// while any older dashboard build is still in circulation.
this.app.get('/api/neighbors', async (request, reply) => {
const leaderClient = this.kvManager?.getLeaderClient(); const leaderClient = this.kvManager?.getLeaderClient();
if (!leaderClient) { if (!leaderClient) {
return reply.status(503).send({ error: 'KVManager not initialized' }); return reply.send({ current: 'meta-sort', enabled: false, count: 0, neighbors: [] });
} }
const apiUrl = await leaderClient.getApiUrl(); const q = (request.query ?? {}) as { all?: string; cap?: string };
if (!apiUrl) { const neighbors = leaderClient.getNeighbors({ all: !!q.all, cap: q.cap || undefined });
return reply.status(503).send({ error: 'Leader not available yet' }); return reply.send({
} current: 'meta-sort',
const response = await fetch(`${apiUrl}/api/services`, { enabled: true,
signal: AbortSignal.timeout(5000), count: neighbors.length,
neighbors,
services: neighbors,
self: leaderClient.self(),
}); });
if (!response.ok) {
return reply.status(response.status).send({
error: `Upstream ${apiUrl} returned ${response.status}`,
});
}
const body = await response.json();
return reply.send(body);
} catch (err) {
return reply.status(502).send({ error: getErrorMessage(err) });
}
}); });
} }
@@ -886,38 +892,43 @@ export class UnifiedAPIServer {
return reply.status(400).send({ error: `Plugin '${pluginId}' is not active` }); return reply.status(400).send({ error: `Plugin '${pluginId}' is not active` });
} }
if (!hasFileTuples(this.kvClient)) {
return reply.status(501).send({ error: 'Recompute needs meta-core GET /api/files/tuples' });
}
const kvClient = this.kvClient as IKVClientWithFileTuples;
if (this.recomputeInFlight.has(pluginId)) {
return reply.status(409).send({ error: `Recompute for plugin '${pluginId}' is already being queued` });
}
this.recomputeInFlight.add(pluginId);
try { try {
// Get all files from KV // Only file-backed records can be recomputed, and meta-core lists exactly
const hashIds = await this.kvClient.getAllHashIds(); // those in one batched call. Walking getAllHashIds() instead reads every
console.log(`[Recompute] Starting recompute for plugin '${pluginId}' on ${hashIds.length} files`); // gateway record too (82k on watch.nsl.sh) to keep a few thousand.
const tuples = await kvClient.getFileTuples();
if (!tuples || !tuples.files) {
return reply.status(501).send({
error: 'Recompute needs meta-core GET /api/files/tuples',
details: 'meta-core did not return a file list'
});
}
const candidates = tuples.files;
console.log(`[Recompute] Starting recompute for plugin '${pluginId}' on ${candidates.length} files`);
// Prepare file data with metadata for TaskScheduler // Prepare file data with metadata for TaskScheduler
const files: Array<{ filePath: string; fileHash: string; kvData?: Record<string, string> }> = []; const files: Array<{ filePath: string; fileHash: string; kvData?: Record<string, string> }> = [];
await forEachWithConcurrency(candidates, RECOMPUTE_READ_CONCURRENCY, async (tuple) => {
for (const hashId of hashIds) { const metadata = await kvClient.getMetadataFlat(tuple.hashId);
const metadata = await this.kvClient.getMetadataFlat(hashId); if (!metadata) {
if (metadata && metadata.filePath) { return;
// Convert nested metadata to flat key-value pairs for KVStore
const kvData: Record<string, string> = {};
const flattenObj = (obj: any, prefix = '') => {
for (const [key, value] of Object.entries(obj)) {
const fullKey = prefix ? `${prefix}/${key}` : key;
if (value !== null && typeof value === 'object' && !Array.isArray(value)) {
flattenObj(value, fullKey);
} else if (value !== undefined && value !== null) {
kvData[fullKey] = String(value);
} }
}
};
flattenObj(metadata);
files.push({ files.push({
fileHash: hashId, fileHash: tuple.hashId,
filePath: metadata.filePath, filePath: metadata.filePath || tuple.filePath,
kvData kvData: flattenMetadata(metadata)
});
}); });
}
}
// Create and enqueue tasks with forceRecompute option // Create and enqueue tasks with forceRecompute option
const tasks = taskScheduler.createTasksForPluginOnFiles( const tasks = taskScheduler.createTasksForPluginOnFiles(
@@ -935,7 +946,7 @@ export class UnifiedAPIServer {
status: 'ok', status: 'ok',
message: `Recompute triggered for plugin '${pluginId}'`, message: `Recompute triggered for plugin '${pluginId}'`,
filesQueued: tasks.length, filesQueued: tasks.length,
totalFiles: hashIds.length totalFiles: candidates.length
}; };
} catch (error: any) { } catch (error: any) {
console.error(`[Recompute] Error triggering recompute for plugin '${pluginId}':`, error); console.error(`[Recompute] Error triggering recompute for plugin '${pluginId}':`, error);
@@ -943,6 +954,8 @@ export class UnifiedAPIServer {
error: 'Failed to trigger recompute', error: 'Failed to trigger recompute',
details: error.message details: error.message
}); });
} finally {
this.recomputeInFlight.delete(pluginId);
} }
}); });
} }
+176
View File
@@ -8,6 +8,7 @@
import { expect } from 'chai'; import { expect } from 'chai';
import { UnifiedAPIServer } from './UnifiedAPIServer.js'; import { UnifiedAPIServer } from './UnifiedAPIServer.js';
import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js'; import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js';
import { performanceMetrics } from '../metrics/PerformanceMetrics.js';
import type { IKVClient } from '../kv/IKVClient.js'; import type { IKVClient } from '../kv/IKVClient.js';
import type { FastifyInstance } from 'fastify'; import type { FastifyInstance } from 'fastify';
@@ -23,6 +24,25 @@ class MockKVClient implements IKVClient {
private data: Map<string, Map<string, any>> = new Map(); private data: Map<string, Map<string, any>> = new Map();
private rawStore: Map<string, string> = new Map(); private rawStore: Map<string, string> = new Map();
// Stand-in for meta-core's GET /api/files/tuples. null = a meta-core that
// predates the endpoint.
fileTuples: Array<{ hashId: string; filePath: string; sizeByte: number; mtimeNano: number }> | null = null;
fileTuplesDelayMs = 0;
calls = { getFileTuples: 0, getAllHashIds: 0, getMetadataFlat: 0 };
async getFileTuples(opts: { summary?: boolean } = {}) {
this.calls.getFileTuples++;
if (this.fileTuplesDelayMs) {
await new Promise(resolve => setTimeout(resolve, this.fileTuplesDelayMs));
}
if (!this.fileTuples) return null;
const files = this.fileTuples;
const totalSize = files.reduce((sum, f) => sum + f.sizeByte, 0);
return opts.summary
? { count: files.length, totalSize }
: { count: files.length, totalSize, files };
}
// Basic operations // Basic operations
async set(key: string, value: any): Promise<void> { async set(key: string, value: any): Promise<void> {
this.rawStore.set(key, JSON.stringify(value)); this.rawStore.set(key, JSON.stringify(value));
@@ -99,6 +119,7 @@ class MockKVClient implements IKVClient {
} }
async getMetadataFlat(hashId: string): Promise<any | null> { async getMetadataFlat(hashId: string): Promise<any | null> {
this.calls.getMetadataFlat++;
const entry = this.data.get(hashId); const entry = this.data.get(hashId);
if (!entry) return null; if (!entry) return null;
return Object.fromEntries(entry); return Object.fromEntries(entry);
@@ -125,6 +146,7 @@ class MockKVClient implements IKVClient {
} }
async getAllHashIds(): Promise<string[]> { async getAllHashIds(): Promise<string[]> {
this.calls.getAllHashIds++;
return Array.from(this.data.keys()); return Array.from(this.data.keys());
} }
@@ -736,3 +758,157 @@ describe('UnifiedAPIServer', function() {
}); });
}); });
}); });
// =============================================================================
// File totals + recompute against meta-core's GET /api/files/tuples
// =============================================================================
//
// Each test builds its own server: /api/stats caches for 30 s and recompute is
// single-flight, so shared state between tests would hide both behaviours.
describe('UnifiedAPIServer file totals and recompute', function() {
const tuple = (hashId: string, sizeByte: number) =>
({ hashId, filePath: `/files/watch/${hashId}.mkv`, sizeByte, mtimeNano: 1 });
let app: FastifyInstance;
let kv: MockKVClient;
let queued: Array<{ pluginId: string; fileHash: string }>;
beforeEach(async function() {
kv = new MockKVClient();
queued = [];
const pluginManager = { getPlugins: () => [{ id: 'ffmpeg', active: true }] };
const taskScheduler = {
createTasksForPluginOnFiles: (pluginId: string, files: Array<{ fileHash: string }>) =>
files.map(f => ({ pluginId, fileHash: f.fileHash })),
enqueueTasks: (tasks: Array<{ pluginId: string; fileHash: string }>) => { queued.push(...tasks); },
};
const server = new UnifiedAPIServer(
{ port: 0, host: 'localhost', enableCors: false },
new MockProcessingStateManager() as any,
kv as any,
4,
16,
() => ({ fast: { pending: 0, running: 0 }, background: { pending: 0, running: 0 } }),
undefined,
() => pluginManager as any,
() => taskScheduler as any
);
app = server.getApp();
await app.ready();
});
afterEach(async function() {
await app.close();
});
it('GET /api/stats sums meta-core file tuples without walking hash ids', async function() {
kv.fileTuples = [tuple('a', 100), tuple('b', 250)];
const response = await app.inject({ method: 'GET', url: '/api/stats' });
expect(response.statusCode).to.equal(200);
expect(JSON.parse(response.payload)).to.deep.equal({ fileCount: 2, totalSize: 350, source: 'meta-core' });
expect(kv.calls.getAllHashIds).to.equal(0);
expect(kv.calls.getMetadataFlat).to.equal(0);
});
it('GET /api/stats degrades to nulls on a meta-core without the endpoint', async function() {
kv.fileTuples = null;
const response = await app.inject({ method: 'GET', url: '/api/stats' });
expect(response.statusCode).to.equal(200);
expect(JSON.parse(response.payload)).to.deep.equal({ fileCount: null, totalSize: null, source: 'unavailable' });
expect(kv.calls.getAllHashIds).to.equal(0);
});
it('GET /api/stats shares one meta-core call between concurrent polls', async function() {
kv.fileTuples = [tuple('a', 100)];
kv.fileTuplesDelayMs = 50;
const responses = await Promise.all(
Array.from({ length: 10 }, () => app.inject({ method: 'GET', url: '/api/stats' }))
);
for (const response of responses) {
expect(response.statusCode).to.equal(200);
expect(JSON.parse(response.payload).totalSize).to.equal(100);
}
expect(kv.calls.getFileTuples).to.equal(1);
});
it('POST recompute returns 501 on a meta-core without the endpoint', async function() {
kv.fileTuples = null;
const response = await app.inject({ method: 'POST', url: '/api/plugins/ffmpeg/recompute' });
expect(response.statusCode).to.equal(501);
expect(kv.calls.getAllHashIds).to.equal(0);
expect(queued).to.have.length(0);
});
it('POST recompute queues only file-backed records', async function() {
kv.seedData('a', { filePath: '/files/watch/a.mkv', fileType: 'video' });
kv.seedData('b', { filePath: '/files/watch/b.mkv', fileType: 'video' });
kv.seedData('gateway-record', { title: 'Not a file' });
kv.fileTuples = [tuple('a', 100), tuple('b', 250)];
const response = await app.inject({ method: 'POST', url: '/api/plugins/ffmpeg/recompute' });
expect(response.statusCode).to.equal(200);
const body = JSON.parse(response.payload);
expect(body.filesQueued).to.equal(2);
expect(body.totalFiles).to.equal(2);
expect(queued.map(q => q.fileHash).sort()).to.deep.equal(['a', 'b']);
expect(kv.calls.getMetadataFlat).to.equal(2);
expect(kv.calls.getAllHashIds).to.equal(0);
});
it('POST recompute refuses a second run for the same plugin while one is queuing', async function() {
kv.seedData('a', { filePath: '/files/watch/a.mkv' });
kv.fileTuples = [tuple('a', 100)];
kv.fileTuplesDelayMs = 50;
const [first, second] = await Promise.all([
app.inject({ method: 'POST', url: '/api/plugins/ffmpeg/recompute' }),
app.inject({ method: 'POST', url: '/api/plugins/ffmpeg/recompute' }),
]);
expect([first.statusCode, second.statusCode].sort()).to.deep.equal([200, 409]);
expect(queued).to.have.length(1);
});
});
// =============================================================================
// File-level processing metric
// =============================================================================
describe('UnifiedProcessingStateManager file metric', function() {
// Files announced over SSE go straight to light processing; they are never
// marked discovered, which is the path that used to leave "Processed" at 0.
const runFile = (manager: UnifiedProcessingStateManager, filePath: string, error?: string) => {
manager.startLightProcessing(filePath);
manager.completeLightProcessing(filePath, 'bagacbabaeexamplehash');
manager.startHashProcessing(filePath);
manager.completeHashProcessing(filePath, 'bagacbabaeexamplehash', undefined, error);
};
it('counts a file as processed when its hash phase completes cleanly', function() {
const manager = new UnifiedProcessingStateManager();
const before = performanceMetrics.getMetrics().totalFilesProcessed;
runFile(manager, '/files/watch/metric-ok.mkv');
expect(performanceMetrics.getMetrics().totalFilesProcessed).to.equal(before + 1);
});
it('does not count a file whose hash phase failed', function() {
const manager = new UnifiedProcessingStateManager();
const before = performanceMetrics.getMetrics().totalFilesProcessed;
runFile(manager, '/files/watch/metric-failed.mkv', 'plugin failed');
expect(performanceMetrics.getMetrics().totalFilesProcessed).to.equal(before);
});
});
@@ -26,7 +26,8 @@ export function renamingRule(metamesh: Local, filepath?: string): string | null
throw new Error(`No extension found for file: ${filepath}`); throw new Error(`No extension found for file: ${filepath}`);
} }
const title = metamesh?.titles?.eng || metamesh.originalTitle; // The display name — never the `titles/*` name set, which has no preferred member (METADATA_KEYS.md §3).
const title = metamesh.title || metamesh.originalTitle;
if (!title) { if (!title) {
if (metamesh.fileType !== "torrent") { if (metamesh.fileType !== "torrent") {
//don't warn for torrent files because they don't always have a title (eg representing a folder of files) //don't warn for torrent files because they don't always have a title (eg representing a folder of files)
@@ -142,53 +142,18 @@ export class ContainerManager extends EventEmitter {
return; return;
} }
// Fall back to service discovery for any missing URLs // Anything still missing is a configuration problem, not something to
const serviceFile = join(config.META_CORE_PATH, 'services', 'meta-sort.json'); // guess at. The old fallback here read
// `${META_CORE_PATH}/services/meta-sort.json` — a filename that stopped
try { // existing when registration files became hostname-suffixed, using a
const content = await fs.readFile(serviceFile, 'utf-8'); // schema (`.api` / `.endpoints`) that no longer matched either. It had
const serviceInfo: ServiceInfo = JSON.parse(content); // been falling into its own catch block for a long time, and the
// volume it read is gone since meta-discovery v1.
// Extract URLs from service info
const api = serviceInfo.api;
const endpoints = serviceInfo.endpoints || {};
// Use service discovery for callback URL if not set
if (!this.callbackUrl) { if (!this.callbackUrl) {
if (endpoints.callback) { console.warn('[ContainerManager] No callback URL — set CONTAINER_CALLBACK_URL');
this.callbackUrl = endpoints.callback;
console.log(`[ContainerManager] Discovered callback URL: ${this.callbackUrl}`);
} else if (api) {
// Fallback: construct from api base
this.callbackUrl = `${api}/api/plugins/callback`;
console.log(`[ContainerManager] Constructed callback URL: ${this.callbackUrl}`);
} }
}
// Use service discovery for WebDAV URL if not set (fallback)
if (!this.webdavUrl) { if (!this.webdavUrl) {
if (endpoints.webdav) { console.warn('[ContainerManager] No WebDAV URL — meta-core did not supply one');
this.webdavUrl = endpoints.webdav;
console.log(`[ContainerManager] Discovered WebDAV URL: ${this.webdavUrl}`);
} else if (api) {
// Fallback: construct from api base
this.webdavUrl = `${api}/webdav`;
console.log(`[ContainerManager] Constructed WebDAV URL: ${this.webdavUrl}`);
}
}
if (endpoints.health) {
// Extract meta-core URL from health endpoint
const healthUrl = new URL(endpoints.health);
this.metaCoreUrl = `${healthUrl.protocol}//${healthUrl.host}`;
console.log(`[ContainerManager] Discovered meta-core URL: ${this.metaCoreUrl}`);
}
console.log('[ContainerManager] Service discovery successful');
} catch (error) {
console.warn(`[ContainerManager] Service discovery failed: ${error}`);
console.log(`[ContainerManager] Using callback URL: ${this.callbackUrl}`);
console.log(`[ContainerManager] Using WebDAV URL: ${this.webdavUrl || 'not set'}`);
} }
} }
@@ -0,0 +1,569 @@
/**
* Beacon v2 — local resource advertise / discover over UDP multicast
* (239.255.99.1:9099).
*
* Every node advertises resources tagged with capability strings
* (`metamesh.core`, `metamesh.service/meta-sort`, `metamesh.transport/nzb@1`)
* and keeps a live view of everyone else's. This service uses it to locate
* meta-core (the `metamesh.core` resource carries its /urls block) and to feed
* the nav menu (`metamesh.service/*`), with no shared volume.
*
* The normative spec is docs/project-architecture/beacon-v2.md.
*
* ⚠ MIRRORED FILE. Byte-identical copies live in meta-sort-core, meta-fuse-core
* and meta-dup-core. They cannot share a package: meta-fuse-core and
* meta-dup-core do not depend on @metazla/meta-interface at all, and each
* submodule is its own Docker build context, so a workspace dep resolves
* outside the build. Edit one copy → edit all three, and
* `scripts/check-mirrors.sh` will tell you if you forgot.
*/
import dgram from 'dgram';
import { hostname, networkInterfaces } from 'os';
export const PROTO = 'beacon';
export const VERSION = 2;
export const DEFAULT_GROUP = '239.255.99.1';
export const DEFAULT_PORT = 9099;
export const DEFAULT_INTERVAL_MS = 10_000;
/** A node unheard for interval × this is dropped (evaluated at read time). */
export const LIVENESS_FACTOR = 3;
/** Senders stay under this so nothing fragments. */
export const MAX_DATAGRAM = 1400;
export const TYPE_PROBE = 'probe';
export const TYPE_ADVERTISE = 'advertise';
export const TYPE_BYE = 'bye';
export const CAP_CORE = 'metamesh.core';
export const CAP_SERVICE_PREFIX = 'metamesh.service/';
export const CAP_ANY_SERVICE = 'metamesh.service/*';
/** Mirrors meta-core's GET /urls response minus redisUrl (retired). */
export interface MeshUrls {
hostname: string;
baseUrl: string;
apiUrl: string;
webdavUrl: string;
webdavUrlInternal: string;
}
export interface BeaconNodeInfo {
name: string;
instance: string;
version?: string;
/** 'starting' | 'running'; absent reads as running. */
status?: string;
}
export interface BeaconResource {
id: string;
caps: string[];
/** Well-known: http, ui, manifest, mcp. A '/…' value is relative to http. */
endpoints?: Record<string, string>;
rev?: string;
/** The one consumer instance this resource belongs to. */
binds?: string;
data?: Record<string, unknown>;
}
export interface BeaconMessage {
proto: string;
v: number;
type: string;
node?: BeaconNodeInfo;
resources?: BeaconResource[];
from?: string;
want?: string[];
}
/** A node as served by /api/neighbors. */
export interface MeshNeighbor extends BeaconNodeInfo {
resources: BeaconResource[];
/** Every resource's caps, flattened. */
caps: string[];
/** The metamesh.service/* resource's endpoints.ui (v1-compatible field). */
baseUrl?: string;
/** Source address taken from the packet, never from the payload. */
addr: string;
/** Unix seconds. */
lastSeen: number;
}
export interface MeshNodeConfig {
name: string;
instance?: string;
version?: string;
group?: string;
port?: number;
intervalMs?: number;
enabled?: boolean;
/** Rebuilt on every advertise so the payload never goes stale. */
payload?: () => { status?: string; resources: BeaconResource[] };
}
/**
* Does capability `cap` satisfy `pattern`? `*` matches everything; an `@N` on
* the pattern must equal the cap's contract (none ignores it); `x/*` matches
* any variant of `x`; otherwise the bases must be equal.
*/
export function capMatches(pattern: string, cap: string): boolean {
if (pattern === '*') return true;
const split = (s: string): [string, string | null] => {
const i = s.lastIndexOf('@');
return i >= 0 ? [s.slice(0, i), s.slice(i + 1)] : [s, null];
};
const [pbase, pc] = split(pattern);
const [cbase, cc] = split(cap);
if (pc !== null && cc !== pc) return false;
if (pbase.endsWith('/*')) {
const prefix = pbase.slice(0, -2);
if (!cbase.startsWith(prefix)) return false;
const rest = cbase.slice(prefix.length);
return rest.startsWith('/') && rest.length > 1;
}
return pbase === cbase;
}
export function resourceMatches(r: BeaconResource, pattern: string): boolean {
return (r.caps ?? []).some((c) => capMatches(pattern, c));
}
/** The named endpoint as an absolute URL ('/…' joins onto endpoints.http). */
export function resourceEndpoint(r: BeaconResource, name: string): string | undefined {
const v = r.endpoints?.[name];
if (!v) return undefined;
if (!v.startsWith('/')) return v;
const base = r.endpoints?.http;
return base ? base.replace(/\/+$/, '') + v : undefined;
}
/**
* Decode a datagram; null for anything that is not a well-formed beacon v2
* message (beacon v1, meta-discovery v1, other versions, junk).
*/
export function parseMessage(buf: Buffer): BeaconMessage | null {
let m: BeaconMessage;
try {
m = JSON.parse(buf.toString('utf8')) as BeaconMessage;
} catch {
return null;
}
if (!m || m.proto !== PROTO || m.v !== VERSION) return null;
if (m.type === TYPE_PROBE) return m;
if (m.type === TYPE_ADVERTISE || m.type === TYPE_BYE) {
if (!m.node?.name || !m.node?.instance) return null;
return m;
}
return null;
}
/** IPv4 addresses of every up, non-internal interface. */
function localInterfaceAddresses(): string[] {
const out: string[] = [];
const nets = networkInterfaces();
for (const name of Object.keys(nets)) {
for (const ni of nets[name] ?? []) {
// Node has flip-flopped between 'IPv4' and 4 for `family`.
const isV4 = (ni as { family: string | number }).family === 'IPv4' ||
(ni as { family: string | number }).family === 4;
if (isV4 && !ni.internal && ni.address) out.push(ni.address);
}
}
return out;
}
/** First non-internal IPv4 address, or the hostname. Matches the Go helper. */
export function localIPv4(): string {
return localInterfaceAddresses()[0] ?? hostname();
}
function envNum(key: string): number | undefined {
const n = Number(process.env[key]);
return Number.isFinite(n) && n > 0 ? n : undefined;
}
interface Seen {
node: BeaconNodeInfo;
resources: BeaconResource[];
addr: string;
lastSeen: number; // ms
}
function toNeighbor(s: Seen): MeshNeighbor {
const caps: string[] = [];
let baseUrl: string | undefined;
for (const r of s.resources) {
for (const c of r.caps ?? []) if (!caps.includes(c)) caps.push(c);
if (!baseUrl && resourceMatches(r, CAP_ANY_SERVICE)) baseUrl = resourceEndpoint(r, 'ui');
}
return {
...s.node,
resources: s.resources,
caps,
...(baseUrl ? { baseUrl } : {}),
addr: s.addr,
lastSeen: s.lastSeen / 1000,
};
}
/**
* One beacon v2 participant: advertises its resources, answers probes, and
* keeps a TTL map of every other node.
*/
export class MeshNode {
private cfg: Required<Omit<MeshNodeConfig, 'payload'>> & Pick<MeshNodeConfig, 'payload'>;
private socket: dgram.Socket | null = null;
private timer: NodeJS.Timeout | null = null;
private nodes = new Map<string, Seen>();
private ifaceAddrs: string[] = [];
private listeners: ((n: MeshNeighbor) => void)[] = [];
private started = false;
constructor(config: MeshNodeConfig) {
this.cfg = {
name: config.name,
instance: config.instance ?? hostname(),
version: config.version ?? '',
group: config.group ?? process.env.BEACON_GROUP ?? DEFAULT_GROUP,
port: config.port ?? envNum('BEACON_PORT') ?? DEFAULT_PORT,
intervalMs: config.intervalMs ?? envNum('BEACON_INTERVAL_MS') ?? DEFAULT_INTERVAL_MS,
enabled: config.enabled ?? true,
payload: config.payload,
};
}
async start(): Promise<void> {
if (this.started || !this.cfg.enabled) {
if (!this.cfg.enabled) console.log('[beacon] Disabled by configuration');
return;
}
this.started = true;
await new Promise<void>((resolve) => {
// reuseAddr sets SO_REUSEADDR (+ SO_REUSEPORT on Linux for udp4),
// so several listeners can share 9099 on one host.
const socket = dgram.createSocket({ type: 'udp4', reuseAddr: true });
this.socket = socket;
socket.on('message', (buf, rinfo) => this.onMessage(buf, rinfo));
socket.on('error', (err) => {
// Never fatal: a host that blocks multicast must degrade, not
// take the service down.
console.error('[beacon] Socket error:', err.message);
});
socket.bind(this.cfg.port, '0.0.0.0', () => {
try {
socket.setMulticastTTL(1); // link-local only
} catch (e) {
console.warn('[beacon] Could not set multicast TTL:', e);
}
// Join on EVERY interface. A container on two docker networks
// that joins only the default route silently sees one network.
this.ifaceAddrs = localInterfaceAddresses();
let joined = 0;
for (const addr of this.ifaceAddrs) {
try {
socket.addMembership(this.cfg.group, addr);
joined++;
} catch (e) {
console.warn(`[beacon] Join ${this.cfg.group} on ${addr} failed:`, e);
}
}
console.log(
`[beacon] Listening on ${this.cfg.group}:${this.cfg.port} as ` +
`${this.cfg.name}/${this.cfg.instance} (joined ${joined} of ${this.ifaceAddrs.length} interfaces)`
);
resolve();
});
});
this.advertise();
this.probe();
this.timer = setInterval(() => this.advertise(), this.cfg.intervalMs);
// Don't hold the event loop open on shutdown.
this.timer.unref?.();
}
async stop(): Promise<void> {
if (!this.started) return;
this.started = false;
if (this.timer) { clearInterval(this.timer); this.timer = null; }
// Bye, so neighbours drop us immediately rather than waiting out the
// staleness window.
this.multicast({ proto: PROTO, v: VERSION, type: TYPE_BYE, node: this.nodeInfo() });
await new Promise<void>((resolve) => {
if (!this.socket) return resolve();
try { this.socket.close(() => resolve()); } catch { resolve(); }
});
this.socket = null;
this.listeners = [];
}
/**
* Multicast a probe. Every node owning a resource matching one of `want`
* (every node, when empty) replies immediately.
*/
probe(want: string[] = []): void {
const msg: BeaconMessage = { proto: PROTO, v: VERSION, type: TYPE_PROBE, from: this.cfg.instance };
if (want.length) msg.want = want;
this.multicast(msg);
}
/** Every live node, sorted by (name, instance). */
getNeighbors(): MeshNeighbor[] {
const cutoff = Date.now() - this.cfg.intervalMs * LIVENESS_FACTOR;
const out: MeshNeighbor[] = [];
for (const [key, s] of this.nodes) {
if (s.lastSeen < cutoff) { this.nodes.delete(key); continue; }
out.push(toNeighbor(s));
}
return out.sort((a, b) => a.name.localeCompare(b.name) || a.instance.localeCompare(b.instance));
}
/** One row per node name, most recently seen instance wins. */
getNeighborsByName(): MeshNeighbor[] {
const best = new Map<string, MeshNeighbor>();
for (const nb of this.getNeighbors()) {
const cur = best.get(nb.name);
if (!cur || nb.lastSeen > cur.lastSeen) best.set(nb.name, nb);
}
return [...best.values()].sort((a, b) => a.name.localeCompare(b.name));
}
/** This node as a row, so a UI can render itself without waiting. */
self(): MeshNeighbor {
const { status, resources } = this.payload();
return toNeighbor({ node: this.nodeInfo(status), resources, addr: '', lastSeen: Date.now() });
}
/** Fires for every advertise received (after the self-echo filter). */
onAdvertise(cb: (n: MeshNeighbor) => void): void {
this.listeners.push(cb);
}
private payload(): { status?: string; resources: BeaconResource[] } {
return this.cfg.payload ? this.cfg.payload() : { resources: [] };
}
private nodeInfo(status?: string): BeaconNodeInfo {
const n: BeaconNodeInfo = { name: this.cfg.name, instance: this.cfg.instance };
if (this.cfg.version) n.version = this.cfg.version;
if (status) n.status = status;
return n;
}
private advertiseMsg(): BeaconMessage {
const { status, resources } = this.payload();
return { proto: PROTO, v: VERSION, type: TYPE_ADVERTISE, node: this.nodeInfo(status), resources };
}
private onMessage(buf: Buffer, rinfo: dgram.RemoteInfo): void {
const msg = parseMessage(buf);
if (!msg) return; // v1 or foreign traffic on the shared group
if (msg.type === TYPE_PROBE) {
if (msg.from === this.cfg.instance) return;
const ad = this.advertiseMsg();
const want = msg.want ?? [];
const wanted = want.length === 0 ||
(ad.resources ?? []).some((r) => want.some((p) => resourceMatches(r, p)));
if (wanted) this.replyTo(ad, rinfo);
return;
}
const node = msg.node!;
if (node.instance === this.cfg.instance) return; // our own echo
if (msg.type === TYPE_BYE) { this.nodes.delete(node.instance); return; }
const seen: Seen = { node, resources: msg.resources ?? [], addr: rinfo.address, lastSeen: Date.now() };
this.nodes.set(node.instance, seen);
const nb = toNeighbor(seen);
for (const cb of this.listeners) {
try { cb(nb); } catch (e) { console.error('[beacon] Listener threw:', e); }
}
}
private encode(msg: BeaconMessage): Buffer {
const body = Buffer.from(JSON.stringify(msg));
if (body.length > MAX_DATAGRAM) {
console.warn(`[beacon] ${body.length}-byte datagram exceeds the fragmentation-safe ${MAX_DATAGRAM}`);
}
return body;
}
private advertise(): void {
this.multicast(this.advertiseMsg());
}
private replyTo(msg: BeaconMessage, rinfo: dgram.RemoteInfo): void {
try {
this.socket?.send(this.encode(msg), rinfo.port, rinfo.address);
} catch (e) {
console.warn('[beacon] Reply failed:', e);
}
}
/** Writes once per interface — the default route alone reaches one network. */
private multicast(msg: BeaconMessage): void {
const socket = this.socket;
if (!socket) return;
const body = this.encode(msg);
for (const addr of this.ifaceAddrs) {
try {
socket.setMulticastInterface(addr);
socket.send(body, this.cfg.port, this.cfg.group);
} catch (e) {
console.warn(`[beacon] Send on ${addr} failed:`, e);
}
}
}
}
export interface MetaCoreLocatorConfig {
/** This service's name, for its own advertise. */
serviceName: string;
/** Browser-facing URL for the nav menu. */
baseUrl?: string;
version?: string;
/**
* Explicit meta-core API URL. When set this ALWAYS wins and the wire is
* never consulted for core selection — see the spec's "the
* metamesh.core pin". This is what stops a client box from latching onto a
* gateway box's core when both are reachable on a shared network.
*/
metaCoreUrl?: string;
group?: string;
port?: number;
intervalMs?: number;
enabled?: boolean;
}
/**
* Locates meta-core over beacon v2 (the first `metamesh.core` resource that
* carries `data.urls`), and advertises this service as
* `metamesh.service/<serviceName>`.
*/
export class MetaCoreLocator {
private node: MeshNode;
private cfg: MetaCoreLocatorConfig;
private current: MeshUrls | null = null;
private changeCallbacks: (() => void)[] = [];
constructor(config: MetaCoreLocatorConfig) {
this.cfg = config;
this.node = new MeshNode({
name: config.serviceName,
version: config.version,
group: config.group,
port: config.port,
intervalMs: config.intervalMs,
enabled: config.enabled,
payload: () => ({
resources: [{
id: 'service',
caps: [CAP_SERVICE_PREFIX + config.serviceName],
endpoints: { ui: config.baseUrl ?? `http://${localIPv4()}` },
}],
}),
});
this.node.onAdvertise((nb) => {
if (this.cfg.metaCoreUrl) return; // pinned: the wire cannot move us
const core = nb.resources.find((r) => resourceMatches(r, CAP_CORE));
const urls = core?.data?.urls as MeshUrls | undefined;
if (!urls?.apiUrl) return;
const prev = this.current;
if (!prev) {
this.current = urls;
console.log(`[beacon] meta-core discovered at ${urls.apiUrl} (${nb.addr})`);
this.notifyChange();
return;
}
if (prev.apiUrl === urls.apiUrl) {
this.current = urls; // refresh the rest of the fields
return;
}
// A second, different core is advertising. Do not flap between
// them — keep the first and say so loudly, because on a PCS box
// this means a client container can see the gateway's core.
console.warn(
`[beacon] Ignoring a second meta-core at ${urls.apiUrl} (${nb.addr}); ` +
`staying with ${prev.apiUrl}. Set META_CORE_URL to pin this explicitly.`
);
});
}
async start(): Promise<void> {
if (this.cfg.metaCoreUrl) {
console.log(`[beacon] meta-core pinned to ${this.cfg.metaCoreUrl}; discovery is advisory`);
}
await this.node.start();
}
async stop(): Promise<void> {
await this.node.stop();
this.changeCallbacks = [];
}
/** The pin if set, else whatever discovery elected. */
getApiUrl(): string | null {
if (this.cfg.metaCoreUrl) return this.cfg.metaCoreUrl;
return this.current?.apiUrl ?? null;
}
/** Full URLs block. null when pinned and /urls has not been fetched yet. */
getUrls(): MeshUrls | null {
return this.current;
}
/**
* Block until meta-core is located. Probes immediately and re-probes every
* 500ms — the same loop shape as the old LeaderClient.waitForLeader, so the
* 30s boot timeout semantics are preserved exactly.
*/
async waitForCore(timeoutMs = 30_000): Promise<string> {
const deadline = Date.now() + timeoutMs;
let logged = false;
while (Date.now() < deadline) {
const url = this.getApiUrl();
if (url) return url;
if (!logged) { console.log('[beacon] Waiting for meta-core...'); logged = true; }
this.node.probe([CAP_CORE]);
await new Promise((r) => setTimeout(r, 500));
}
throw new Error(`[beacon] No meta-core found within ${timeoutMs}ms`);
}
/** Fires when the elected core's apiUrl changes. Replaces the fs.watch. */
onChange(cb: () => void): void {
this.changeCallbacks.push(cb);
}
/**
* Neighbours for /api/neighbors: one row per name (or every instance with
* `all`), optionally only nodes with a resource matching `cap`.
*/
getNeighbors(opts: { all?: boolean; cap?: string } = {}): MeshNeighbor[] {
const list = opts.all ? this.node.getNeighbors() : this.node.getNeighborsByName();
const cap = opts.cap;
return cap ? list.filter((n) => n.resources.some((r) => resourceMatches(r, cap))) : list;
}
self(): MeshNeighbor {
return this.node.self();
}
probe(want: string[] = []): void {
this.node.probe(want);
}
private notifyChange(): void {
for (const cb of this.changeCallbacks) {
try { cb(); } catch (e) { console.error('[beacon] Change callback threw:', e); }
}
}
}
+6 -10
View File
@@ -125,22 +125,18 @@ process.on('SIGINT', async () => {
// Start initial discovery and processing (streaming pipeline) // Start initial discovery and processing (streaming pipeline)
(async () => { (async () => {
try { try {
// Validate META_CORE_PATH is configured (required for leader discovery) // meta-core is located over UDP (beacon v2) — no /meta-core
if (!config.META_CORE_PATH) { // volume required. META_CORE_URL, when set, pins it and discovery
console.error('[Startup] ERROR: META_CORE_PATH is required. Set META_CORE_PATH environment variable.'); // never overrides; see docs/project-architecture/service-discovery.md.
console.error('[Startup] Example: META_CORE_PATH=/meta-core'); {
process.exit(1); console.log('[Startup] Initializing KV Manager (meta-core discovery)...');
}
// Initialize KV Manager (Redis via leader discovery)
if (config.META_CORE_PATH) {
console.log('[Startup] Initializing KV Manager (leader discovery)...');
kvManager = new KVManager({ kvManager = new KVManager({
metaCorePath: config.META_CORE_PATH, metaCorePath: config.META_CORE_PATH,
filesPath: config.FILES_PATH, filesPath: config.FILES_PATH,
serviceName: config.SERVICE_NAME, serviceName: config.SERVICE_NAME,
apiPort: config.FUSE_API_PORT, apiPort: config.FUSE_API_PORT,
baseUrl: config.BASE_URL, baseUrl: config.BASE_URL,
metaCoreUrl: config.META_CORE_URL,
}); });
// Wait for KV to be ready // Wait for KV to be ready
+1 -1
View File
@@ -139,7 +139,7 @@ export interface KVClientConfig {
} }
/** /**
* Leader lock file content format (kv-leader.info) * meta-core's identity and URL set (from its UDP announce, or /urls)
* Written by meta-core leader, read by other services * Written by meta-core leader, read by other services
*/ */
export interface LeaderLockInfo { export interface LeaderLockInfo {
+14 -24
View File
@@ -2,13 +2,12 @@
* KV Manager - Unified manager for KV storage and service discovery * KV Manager - Unified manager for KV storage and service discovery
* *
* This is the main entry point for the KV subsystem. It: * This is the main entry point for the KV subsystem. It:
* 1. Uses LeaderClient to read leader info from meta-core * 1. Uses LeaderClient to locate meta-core over UDP (beacon v2)
* 2. Creates and manages the KV client (Redis) * 2. Creates and manages the KV client (Redis)
* 3. Handles service discovery registration
* 4. Manages reconnection on leader changes * 4. Manages reconnection on leader changes
* *
* Note: Leader election is now handled by meta-core (Go sidecar). * Note: there is no leader election. meta-core announces itself on the
* This service only reads the leader info and connects to Redis. * discovery group; this service listens and connects to its HTTP API.
* *
* Usage: * Usage:
* ```typescript * ```typescript
@@ -28,7 +27,6 @@
import { hostname } from 'os'; import { hostname } from 'os';
import { LeaderClient } from './LeaderClient.js'; import { LeaderClient } from './LeaderClient.js';
import { ServiceRegistration } from './ServiceRegistration.js';
import { RedisKVClient } from './RedisClient.js'; import { RedisKVClient } from './RedisClient.js';
import type { IKVClient, LeaderLockInfo } from './IKVClient.js'; import type { IKVClient, LeaderLockInfo } from './IKVClient.js';
import { configure as configureWebdav } from '../webdav/WebdavClient.js'; import { configure as configureWebdav } from '../webdav/WebdavClient.js';
@@ -59,7 +57,6 @@ interface KVManagerConfig {
export class KVManager { export class KVManager {
private config: KVManagerConfig; private config: KVManagerConfig;
private leaderClient: LeaderClient | null = null; private leaderClient: LeaderClient | null = null;
private serviceRegistration: ServiceRegistration | null = null;
private kvClient: IKVClient | null = null; private kvClient: IKVClient | null = null;
private isStarted = false; private isStarted = false;
private isShuttingDown = false; private isShuttingDown = false;
@@ -112,8 +109,14 @@ export class KVManager {
// Initialize leader client // Initialize leader client
this.leaderClient = new LeaderClient({ this.leaderClient = new LeaderClient({
metaCorePath: this.config.metaCorePath, metaCoreUrl: this.config.metaCoreUrl,
metaCoreUrl: this.config.metaCoreUrl serviceName: this.config.serviceName,
// Browser-facing URL for the nav menu, in the order the spec requires:
// PUBLIC_URL (reachable when there is no Caddy perimeter in front — a
// debug-direct port) -> BASE_URL (the Caddy URL) -> container host.
// Announcing a Caddy URL on a stack with no Caddy running is what puts
// dead links in every neighbour's menu.
baseUrl: process.env.PUBLIC_URL || this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`,
}); });
// Watch for leader changes // Watch for leader changes
@@ -149,15 +152,9 @@ export class KVManager {
// Start watching for leader changes // Start watching for leader changes
this.leaderClient.startWatching(); this.leaderClient.startWatching();
// Initialize and start service registration for heartbeat // No separate service registration step any more: the LeaderClient's
const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`; // discovery node announces this service on the same UDP group it
this.serviceRegistration = new ServiceRegistration({ // listens on, so being discoverable and discovering are one thing.
metaCorePath: this.config.metaCorePath,
serviceName: this.config.serviceName,
version: '1.0.0',
apiUrl
});
await this.serviceRegistration.start();
this.isStarted = true; this.isStarted = true;
this.notifyReady(); this.notifyReady();
@@ -224,10 +221,6 @@ export class KVManager {
} }
// Service registration stop (if it was started) // Service registration stop (if it was started)
if (this.serviceRegistration) {
await this.serviceRegistration.stop();
this.serviceRegistration = null;
}
this.isStarted = false; this.isStarted = false;
console.log('[KVManager] Stopped'); console.log('[KVManager] Stopped');
@@ -324,9 +317,6 @@ export class KVManager {
/** /**
* Get service registration instance * Get service registration instance
*/ */
getServiceRegistration(): ServiceRegistration | null {
return this.serviceRegistration;
}
/** /**
* Check if the KV store is healthy * Check if the KV store is healthy
+156 -175
View File
@@ -1,26 +1,49 @@
/** /**
* LeaderClient - Client for reading leader info from meta-core * LeaderClient - locates meta-core and exposes its URLs.
* *
* This replaces the old LeaderElection.ts by delegating leader election * Since UDP discovery (meta-discovery v1, now beacon v2) this is a thin adapter
* to meta-core (Go sidecar) and just reading the results. * over MetaCoreLocator
* (../discovery/meshdisco.js): meta-core is found by UDP announce instead of by
* reading /meta-core/locks/kv-leader.info, which means this service no longer
* needs the /meta-core volume mounted at all.
* *
* Features: * The public surface is deliberately unchanged so KVManager and its callers did
* - Reads kv-leader.info file for leader URLs * not have to move:
* - Calls meta-core /urls API for current leader URLs * - getLeaderInfo() / getUrls() / getApiUrl() / getWebdavUrl*()
* - Watches for leader changes via file system watcher * - waitForLeader(timeoutMs) — same 30s default, same "poll until it answers"
* - startWatching() / onChange() — the fs.watch on the lock dir became "an
* announce arrived carrying a different apiUrl", feeding the same callback.
*
* Precedence is the pin: when metaCoreUrl (META_CORE_URL) is set it always
* wins and the wire is never consulted for core selection. See
* docs/project-architecture/service-discovery.md.
*
* The name "leader" is legacy. meta-core's flock election is vestigial — the Go
* binary only ever runs as leader — and redisUrl has been empty since the
* api-mediated-access lockdown; everything routes over HTTP.
*/ */
import { promises as fs } from 'fs'; import { MetaCoreLocator, type MeshNeighbor } from '../discovery/meshdisco.js';
import { watch, FSWatcher } from 'fs';
import { dirname } from 'path';
import type { LeaderLockInfo } from './IKVClient.js'; import type { LeaderLockInfo } from './IKVClient.js';
export interface LeaderClientConfig { export interface LeaderClientConfig {
/** Path to META_CORE_VOLUME (e.g., /meta-core) */ /**
metaCorePath: string; * @deprecated Unused since meta-discovery v1. Kept so existing callers
* compile unchanged; nothing reads the volume any more.
*/
metaCorePath?: string;
/** meta-core API URL (e.g., http://meta-core:9000 or http://localhost:9000) */ /** meta-core API URL. When set, discovery never overrides it. */
metaCoreUrl?: string; metaCoreUrl?: string;
/** This service's name, used for its own announce. */
serviceName?: string;
/** Browser-facing URL announced for the nav menu. */
baseUrl?: string;
/** Service version, display only. */
version?: string;
} }
export interface URLsResponse { export interface URLsResponse {
@@ -35,76 +58,79 @@ export interface URLsResponse {
export class LeaderClient { export class LeaderClient {
private config: LeaderClientConfig; private config: LeaderClientConfig;
private infoFilePath: string; private locator: MetaCoreLocator;
private leaderInfo: LeaderLockInfo | null = null; private leaderInfo: LeaderLockInfo | null = null;
private watcher: FSWatcher | null = null;
private onChangeCallbacks: (() => void)[] = []; private onChangeCallbacks: (() => void)[] = [];
private started = false;
// URL caching // URL caching — only used on the pinned path, where the URLs still come
// from an HTTP GET /urls. A discovered core carries them in the announce.
private cachedUrls: URLsResponse | null = null; private cachedUrls: URLsResponse | null = null;
private urlsCacheTime: number = 0; private urlsCacheTime = 0;
private readonly urlsCacheTTL: number = 5000; // 5 seconds private readonly urlsCacheTTL = 5000;
constructor(config: LeaderClientConfig) { constructor(config: LeaderClientConfig) {
this.config = config; this.config = config;
this.infoFilePath = `${config.metaCorePath}/locks/kv-leader.info`; this.locator = new MetaCoreLocator({
} serviceName: config.serviceName ?? 'meta-sort',
baseUrl: config.baseUrl,
/** version: config.version,
* Read API URL from file (plain text format) metaCoreUrl: config.metaCoreUrl,
*/ enabled: process.env.ENABLE_UDP_DISCOVERY !== 'false' &&
private async getApiUrlFromFile(): Promise<string | null> { process.env.ENABLE_UDP_DISCOVERY !== '0',
try { });
const content = await fs.readFile(this.infoFilePath, 'utf-8'); this.locator.onChange(() => {
return content.trim() || null; this.cachedUrls = null;
} catch (error) { this.urlsCacheTime = 0;
console.error('[LeaderClient] Failed to read API URL from file:', error); this.notifyChange();
return null;
}
}
/**
* Fetch URLs from meta-core /urls API with caching
*/
private async fetchUrls(apiUrl: string): Promise<URLsResponse | null> {
// Check cache
const now = Date.now();
if (this.cachedUrls && (now - this.urlsCacheTime) < this.urlsCacheTTL) {
return this.cachedUrls;
}
try {
const response = await fetch(`${apiUrl}/urls`, {
method: 'GET',
headers: { 'Accept': 'application/json' },
signal: AbortSignal.timeout(5000)
}); });
if (!response.ok) {
console.error('[LeaderClient] Failed to fetch URLs:', response.status, response.statusText);
return null;
} }
const parsed = await response.json() as URLsResponse; /** Idempotent; safe to call from any entry point. */
// api-mediated-access PR D: redisUrl is no longer published by private async ensureStarted(): Promise<void> {
// meta-core. If it shows up, an older build snuck back in — if (this.started) return;
// fail loudly so the rollback can't go unnoticed. Set this.started = true;
// ALLOW_LEGACY_REDIS_URL=1 to downgrade to a warning during a await this.locator.start();
// deliberate temporary rollback. }
if (parsed.redisUrl) {
/**
* Guard against a rollback that reintroduces direct Redis exposure.
* api-mediated-access PR D removed redisUrl; if it comes back, fail loudly
* so it can't go unnoticed. ALLOW_LEGACY_REDIS_URL=1 downgrades to a warn.
*/
private assertNoRedisUrl(redisUrl: string | undefined): void {
if (!redisUrl) return;
const msg = const msg =
'[LeaderClient] meta-core still publishes redisUrl; ' + '[LeaderClient] meta-core still publishes redisUrl; direct Redis access ' +
'direct Redis access was retired by the api-mediated-access ' + 'was retired by the api-mediated-access lockdown. Verify meta-core version.';
'lockdown. Verify meta-core version.';
if (process.env.ALLOW_LEGACY_REDIS_URL === '1') { if (process.env.ALLOW_LEGACY_REDIS_URL === '1') {
console.warn('WARNING: ' + msg); console.warn('WARNING: ' + msg);
} else { } else {
throw new Error(msg); throw new Error(msg);
} }
} }
/** HTTP GET {apiUrl}/urls — still needed when meta-core is pinned. */
private async fetchUrls(apiUrl: string): Promise<URLsResponse | null> {
const now = Date.now();
if (this.cachedUrls && now - this.urlsCacheTime < this.urlsCacheTTL) {
return this.cachedUrls;
}
try {
const response = await fetch(`${apiUrl}/urls`, {
method: 'GET',
headers: { Accept: 'application/json' },
signal: AbortSignal.timeout(5000),
});
if (!response.ok) {
console.error('[LeaderClient] Failed to fetch URLs:', response.status, response.statusText);
return null;
}
const parsed = (await response.json()) as URLsResponse;
this.assertNoRedisUrl(parsed.redisUrl);
this.cachedUrls = parsed; this.cachedUrls = parsed;
this.urlsCacheTime = now; this.urlsCacheTime = now;
return this.cachedUrls; return parsed;
} catch (error) { } catch (error) {
console.error('[LeaderClient] Error calling /urls API:', error); console.error('[LeaderClient] Error calling /urls API:', error);
return null; return null;
@@ -112,23 +138,37 @@ export class LeaderClient {
} }
/** /**
* Read leader info from file and /urls API * Resolve meta-core's URLs — from the announce when discovered (no HTTP
* hop needed, the payload already carries them), or over HTTP when pinned.
*/ */
async getUrls(): Promise<URLsResponse | null> {
await this.ensureStarted();
const wire = this.locator.getUrls();
if (wire && !this.config.metaCoreUrl) {
return {
hostname: wire.hostname,
baseUrl: wire.baseUrl,
apiUrl: wire.apiUrl,
redisUrl: '',
webdavUrl: wire.webdavUrl,
webdavUrlInternal: wire.webdavUrlInternal,
isLeader: true,
};
}
const apiUrl = this.locator.getApiUrl();
if (!apiUrl) {
console.error('[LeaderClient] No meta-core API URL available');
return null;
}
return this.fetchUrls(apiUrl);
}
async getLeaderInfo(): Promise<LeaderLockInfo | null> { async getLeaderInfo(): Promise<LeaderLockInfo | null> {
try { try {
// Read API URL from file (plain text) const urls = await this.getUrls();
const apiUrl = await this.getApiUrlFromFile(); if (!urls) return null;
if (!apiUrl) {
return null;
}
// Fetch full info from /urls API
const urls = await this.fetchUrls(apiUrl);
if (!urls) {
return null;
}
// Convert URLsResponse to LeaderLockInfo
this.leaderInfo = { this.leaderInfo = {
hostname: urls.hostname, hostname: urls.hostname,
baseUrl: urls.baseUrl, baseUrl: urls.baseUrl,
@@ -137,7 +177,7 @@ export class LeaderClient {
webdavUrl: urls.webdavUrl, webdavUrl: urls.webdavUrl,
webdavUrlInternal: urls.webdavUrlInternal, webdavUrlInternal: urls.webdavUrlInternal,
timestamp: Date.now(), timestamp: Date.now(),
pid: 0 // Unknown for remote leader pid: 0,
}; };
return this.leaderInfo; return this.leaderInfo;
} catch (error) { } catch (error) {
@@ -146,134 +186,67 @@ export class LeaderClient {
} }
} }
/**
* Get Redis URL from leader info
*/
async getRedisUrl(): Promise<string | null> { async getRedisUrl(): Promise<string | null> {
const info = await this.getLeaderInfo(); return (await this.getLeaderInfo())?.redisUrl ?? null;
return info?.redisUrl ?? null;
} }
/**
* Get WebDAV URL from leader info (external, via nginx/HTTPS)
*/
async getWebdavUrl(): Promise<string | null> { async getWebdavUrl(): Promise<string | null> {
const info = await this.getLeaderInfo(); return (await this.getLeaderInfo())?.webdavUrl ?? null;
return info?.webdavUrl ?? null;
} }
/** /** Internal WebDAV URL — use for container-to-container access. */
* Get internal WebDAV URL from leader info (direct to port 9000)
* Use this for container-to-container communication
*/
async getWebdavUrlInternal(): Promise<string | null> { async getWebdavUrlInternal(): Promise<string | null> {
const info = await this.getLeaderInfo(); return (await this.getLeaderInfo())?.webdavUrlInternal ?? null;
return info?.webdavUrlInternal ?? null;
} }
/**
* Get meta-core API URL from leader info
*/
async getApiUrl(): Promise<string | null> { async getApiUrl(): Promise<string | null> {
const info = await this.getLeaderInfo(); return (await this.getLeaderInfo())?.apiUrl ?? null;
return info?.apiUrl ?? null;
} }
/** /**
* Call meta-core /urls API to get current URLs * Block until meta-core is reachable. Probes immediately then every 500ms,
* Useful for getting URLs when you don't want to read the file directly * matching the previous file-polling loop's timing exactly.
*/
async getUrls(): Promise<URLsResponse | null> {
// First try using configured metaCoreUrl
let apiUrl = this.config.metaCoreUrl;
// Fall back to reading from file
if (!apiUrl) {
apiUrl = await this.getApiUrlFromFile();
}
if (!apiUrl) {
console.error('[LeaderClient] No meta-core API URL available');
return null;
}
return this.fetchUrls(apiUrl);
}
/**
* Wait for leader info to be available
*/ */
async waitForLeader(timeoutMs: number = 30000): Promise<LeaderLockInfo> { async waitForLeader(timeoutMs: number = 30000): Promise<LeaderLockInfo> {
const startTime = Date.now(); await this.ensureStarted();
const pollInterval = 500; const deadline = Date.now() + timeoutMs;
while (Date.now() - startTime < timeoutMs) { // Locate the core (pin returns instantly; discovery probes).
await this.locator.waitForCore(timeoutMs);
// Then wait for a usable URLs payload — on the pinned path this is the
// first successful GET /urls, so meta-core still has to be up.
while (Date.now() < deadline) {
const info = await this.getLeaderInfo(); const info = await this.getLeaderInfo();
if (info) { if (info) {
console.log(`[LeaderClient] Leader found: ${info.hostname} at ${info.redisUrl}`); console.log(`[LeaderClient] meta-core found: ${info.hostname} at ${info.apiUrl}`);
return info; return info;
} }
console.log('[LeaderClient] Waiting for meta-core...');
console.log('[LeaderClient] Waiting for leader...'); await new Promise((resolve) => setTimeout(resolve, 500));
await new Promise(resolve => setTimeout(resolve, pollInterval));
} }
throw new Error(`No meta-core found within ${timeoutMs}ms`);
throw new Error(`No leader found within ${timeoutMs}ms`);
} }
/** /**
* Start watching for leader changes * Previously an fs.watch on the lock directory. Discovery is always
* listening, so this only has to make sure the node is running.
*/ */
startWatching(): void { startWatching(): void {
if (this.watcher) { void this.ensureStarted();
return; // Already watching
} }
const lockDir = dirname(this.infoFilePath);
try {
this.watcher = watch(lockDir, (eventType, filename) => {
if (filename === 'kv-leader.info') {
console.log('[LeaderClient] Leader info changed, invalidating cache...');
// Invalidate cache to force fresh API call
this.cachedUrls = null;
this.urlsCacheTime = 0;
this.getLeaderInfo().then(() => {
this.notifyChange();
}).catch(console.error);
}
});
console.log(`[LeaderClient] Watching for leader changes in ${lockDir}`);
} catch (error) {
console.error('[LeaderClient] Failed to start watching:', error);
}
}
/**
* Stop watching for leader changes
*/
stopWatching(): void { stopWatching(): void {
if (this.watcher) { // No-op: the locator keeps listening until close().
this.watcher.close();
this.watcher = null;
console.log('[LeaderClient] Stopped watching for leader changes');
}
} }
/**
* Register callback for leader changes
*/
onChange(callback: () => void): this { onChange(callback: () => void): this {
this.onChangeCallbacks.push(callback); this.onChangeCallbacks.push(callback);
return this; return this;
} }
/**
* Notify all change callbacks
*/
private notifyChange(): void { private notifyChange(): void {
console.log('[LeaderClient] meta-core changed, invalidating cache...');
for (const callback of this.onChangeCallbacks) { for (const callback of this.onChangeCallbacks) {
try { try {
callback(); callback();
@@ -283,18 +256,26 @@ export class LeaderClient {
} }
} }
/**
* Get cached leader info (without re-reading file)
*/
getCachedLeaderInfo(): LeaderLockInfo | null { getCachedLeaderInfo(): LeaderLockInfo | null {
return this.leaderInfo; return this.leaderInfo;
} }
/** /**
* Clean up resources * Neighbours for /api/neighbors: one row per name (every instance with
* `all`), optionally only nodes with a capability matching `cap`.
*/ */
getNeighbors(opts: { all?: boolean; cap?: string } = {}): MeshNeighbor[] {
return this.locator.getNeighbors(opts);
}
/** This service's own row, so the menu can show itself. */
self(): MeshNeighbor {
return this.locator.self();
}
close(): void { close(): void {
this.stopWatching();
this.onChangeCallbacks = []; this.onChangeCallbacks = [];
void this.locator.stop();
this.started = false;
} }
} }
@@ -19,6 +19,24 @@ import {
buildFilePrefix, buildFilePrefix,
} from './MetadataUtils.js'; } from './MetadataUtils.js';
/** One file-backed root as listed by meta-core's GET /api/files/tuples. */
export interface FileTuple {
hashId: string;
filePath: string;
sizeByte: number;
mtimeNano: number;
}
/**
* GET /api/files/tuples body. `files` is present unless the summary form was
* asked for.
*/
export interface FileTuplesResult {
count: number;
totalSize: number;
files?: FileTuple[];
}
export interface MetaCoreApiWriterConfig { export interface MetaCoreApiWriterConfig {
/** Base URL for meta-core's HTTP API (e.g. http://metacore-app:9000) */ /** Base URL for meta-core's HTTP API (e.g. http://metacore-app:9000) */
apiUrl: string; apiUrl: string;
@@ -139,6 +157,27 @@ export class MetaCoreApiWriter {
return Array.isArray(ids) ? ids : []; return Array.isArray(ids) ? ids : [];
} }
/**
* GET /api/files/tuples. meta-core's FILE-BACKED roots (records carrying
* filePath, sizeByte and mtimeNano) with sizes, read server-side in batched
* MGETs — the cheap alternative to walking getAllHashIds(), which on a
* shared meta-core is every gateway record too. `summary` omits the rows.
*
* Returns null when meta-core predates the endpoint (404, or a body that is
* not the expected JSON).
*/
async getFileTuples(opts: { summary?: boolean } = {}): Promise<FileTuplesResult | null> {
const path = opts.summary ? '/api/files/tuples?summary=1' : '/api/files/tuples';
const body = await this.fetchJsonAllowing404('GET', path);
if (!body || typeof body !== 'object') return null;
const b = body as { count?: unknown; totalSize?: unknown; files?: unknown };
return {
count: Number(b.count) || 0,
totalSize: Number(b.totalSize) || 0,
files: Array.isArray(b.files) ? (b.files as FileTuple[]) : undefined,
};
}
/** /**
* GET /health on meta-core. Used as a write-path health probe. * GET /health on meta-core. Used as a write-path health probe.
*/ */
+26 -2
View File
@@ -21,7 +21,7 @@ import {
buildFilePrefix, buildFilePrefix,
buildPropertyKey, buildPropertyKey,
} from './MetadataUtils.js'; } from './MetadataUtils.js';
import { MetaCoreApiWriter } from './MetaCoreApiWriter.js'; import { MetaCoreApiWriter, type FileTuplesResult } from './MetaCoreApiWriter.js';
// ESM/CJS interop for ioredis // ESM/CJS interop for ioredis
const Redis = (IORedis as any).default ?? IORedis; const Redis = (IORedis as any).default ?? IORedis;
@@ -102,7 +102,22 @@ export class RedisKVClient implements IKVClient {
// automatic connect. Any caller that tries to use the Redis // automatic connect. Any caller that tries to use the Redis
// path (e.g. unmigrated stream code) will fail loudly rather // path (e.g. unmigrated stream code) will fail loudly rather
// than silently misbehaving. // than silently misbehaving.
this.redis = new Redis({ host: '127.0.0.1', port: 0, lazyConnect: true, enableOfflineQueue: false }); this.redis = new Redis({
host: '127.0.0.1',
port: 0,
lazyConnect: true,
enableOfflineQueue: false,
// Never reconnect. A stub that retries turns one stray call into
// an endless "Unhandled error event: ECONNREFUSED" loop.
retryStrategy: () => null,
maxRetriesPerRequest: 0,
});
let stubWarned = false;
this.redis.on('error', (err: Error) => {
if (stubWarned) return;
stubWarned = true;
console.warn(`[Redis] HTTP-only stub was used and refused (${err.message}) — a caller still reaches for direct Redis`);
});
this.isConnected = false; this.isConnected = false;
console.log('[Redis] HTTP-only mode — no direct Redis connection'); console.log('[Redis] HTTP-only mode — no direct Redis connection');
return; return;
@@ -474,6 +489,15 @@ export class RedisKVClient implements IKVClient {
return await this.redis.smembers(indexKey); return await this.redis.smembers(indexKey);
} }
/**
* File-backed roots from meta-core (GET /api/files/tuples). Only the HTTP
* writer can serve this; direct-Redis mode returns null so callers take
* their degraded path instead of sweeping every record.
*/
async getFileTuples(opts: { summary?: boolean } = {}): Promise<FileTuplesResult | null> {
return this.writer ? this.writer.getFileTuples(opts) : null;
}
// ======================================================================== // ========================================================================
// Pub/Sub Operations // Pub/Sub Operations
// ======================================================================== // ========================================================================
@@ -1,199 +0,0 @@
/**
* Service Registration using shared filesystem
*
* Each service registers itself in /meta-core/services/{service-name}-{hostname}.json
* with its base URL, status, and heartbeat timestamp.
*
* Service discovery is centralized in meta-core - this class only handles registration.
* For service discovery, use the /api/services endpoint (proxied to meta-core via nginx).
*
* Each service must call start() to register itself and begin heartbeat updates.
* meta-core runs a cleanup process that removes services with stale heartbeats.
*/
import { promises as fs } from 'fs';
import { hostname } from 'os';
import type { ServiceInfo } from './IKVClient.js';
import { isNodeError } from '../types/ExtendedInterfaces.js';
interface ServiceRegistrationConfig {
/** Path to META_CORE_VOLUME (e.g., /meta-core) */
metaCorePath: string;
/** Service name */
serviceName: string;
/** Service version (optional, not used in simplified format) */
version?: string;
/** Base URL for the service */
apiUrl: string;
/** Heartbeat interval in ms (default: 30000) */
heartbeatInterval?: number;
/** Stale threshold in ms (default: 60000) */
staleThreshold?: number;
}
export class ServiceRegistration {
private config: ServiceRegistrationConfig;
private servicesDir: string;
private serviceFilePath: string;
private heartbeatTimer: NodeJS.Timeout | null = null;
private isShuttingDown = false;
private currentHostname: string;
constructor(config: ServiceRegistrationConfig) {
this.config = {
heartbeatInterval: 30000,
staleThreshold: 60000,
...config
};
this.currentHostname = hostname();
this.servicesDir = `${this.config.metaCorePath}/services`;
// Use hostname-based file naming
this.serviceFilePath = `${this.servicesDir}/${this.config.serviceName}-${this.currentHostname}.json`;
}
/**
* Ensure services directory exists
*/
private async ensureServicesDir(): Promise<void> {
await fs.mkdir(this.servicesDir, { recursive: true });
}
/**
* Build service info for this service
*/
private buildServiceInfo(status: ServiceInfo['status']): ServiceInfo {
return {
name: this.config.serviceName,
hostname: this.currentHostname,
baseUrl: this.config.apiUrl,
status,
lastHeartbeat: new Date().toISOString()
};
}
/**
* Register this service by writing service info to the shared filesystem.
* meta-core's cleanup process monitors these files and removes stale entries.
*/
async register(): Promise<void> {
await this.ensureServicesDir();
const info = this.buildServiceInfo('running');
await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2));
console.log(`[ServiceRegistration] Registered ${this.config.serviceName}-${this.currentHostname}`);
}
/**
* Update service status
*/
async updateStatus(status: ServiceInfo['status']): Promise<void> {
try {
const content = await fs.readFile(this.serviceFilePath, 'utf-8');
const info = JSON.parse(content) as ServiceInfo;
info.status = status;
info.lastHeartbeat = new Date().toISOString();
await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2));
} catch (error) {
console.error('[ServiceRegistration] Failed to update status:', error);
}
}
/**
* Send heartbeat (update lastHeartbeat timestamp)
*/
async heartbeat(): Promise<void> {
try {
const content = await fs.readFile(this.serviceFilePath, 'utf-8');
const info = JSON.parse(content) as ServiceInfo;
info.lastHeartbeat = new Date().toISOString();
await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2));
} catch (error) {
// If file was deleted, re-register
if (isNodeError(error) && error.code === 'ENOENT') {
await this.register();
} else {
console.error('[ServiceRegistration] Heartbeat failed:', error);
}
}
}
/**
* Start heartbeat loop
*/
startHeartbeat(): void {
this.heartbeatTimer = setInterval(async () => {
if (this.isShuttingDown) return;
await this.heartbeat();
}, this.config.heartbeatInterval);
}
/**
* Stop heartbeat loop
*/
stopHeartbeat(): void {
if (this.heartbeatTimer) {
clearInterval(this.heartbeatTimer);
this.heartbeatTimer = null;
}
}
/**
* Unregister this service (on shutdown)
*/
async unregister(): Promise<void> {
this.isShuttingDown = true;
this.stopHeartbeat();
try {
// Remove the service file
await fs.unlink(this.serviceFilePath);
console.log(`[ServiceRegistration] Unregistered ${this.config.serviceName}-${this.currentHostname}`);
} catch (error) {
if (!isNodeError(error) || error.code !== 'ENOENT') {
console.error('[ServiceRegistration] Failed to unregister:', error);
}
}
}
// ========================================================================
// Lifecycle
// ========================================================================
/**
* Full startup sequence: register service and start heartbeat loop.
*/
async start(): Promise<void> {
await this.register();
this.startHeartbeat();
}
/**
* Full shutdown sequence
*/
async stop(): Promise<void> {
await this.unregister();
}
// ========================================================================
// Getters
// ========================================================================
getServiceFilePath(): string {
return this.serviceFilePath;
}
getServicesDir(): string {
return this.servicesDir;
}
}
+2 -2
View File
@@ -5,7 +5,8 @@
* - IKVClient interface for KV storage abstraction * - IKVClient interface for KV storage abstraction
* - RedisKVClient for Redis-based storage * - RedisKVClient for Redis-based storage
* - LeaderClient for reading leader info from meta-core * - LeaderClient for reading leader info from meta-core
* - ServiceRegistration for service registration (discovery is centralized in meta-core) * Services announce and discover over UDP (beacon v2); there is no
* separate registration step, so no ServiceRegistration export.
* - KVManager for unified management of all components * - KVManager for unified management of all components
* - MetadataUtils for flattening/reconstructing nested metadata * - MetadataUtils for flattening/reconstructing nested metadata
*/ */
@@ -13,6 +14,5 @@
export * from './IKVClient.js'; export * from './IKVClient.js';
export * from './RedisClient.js'; export * from './RedisClient.js';
export * from './LeaderClient.js'; export * from './LeaderClient.js';
export * from './ServiceRegistration.js';
export * from './KVManager.js'; export * from './KVManager.js';
export * from './MetadataUtils.js'; export * from './MetadataUtils.js';
@@ -85,13 +85,10 @@ export class JellyfinFormatHandler implements IMetadataFormatHandler {
if (metadata.originalTitle) jellyfinFields.originalTitle = metadata.originalTitle; if (metadata.originalTitle) jellyfinFields.originalTitle = metadata.originalTitle;
if (metadata.movieYear) jellyfinFields.year = metadata.movieYear; if (metadata.movieYear) jellyfinFields.year = metadata.movieYear;
// Titles (first title or original title) // Title: the display name. `titles/*` is an unordered name set with no
if (metadata.titles) { // preferred member, so it is never the NFO title (METADATA_KEYS.md §3).
const titleEntries = Object.entries(metadata.titles); const title = metadata.title || metadata.originalTitle;
if (titleEntries.length > 0) { if (title) jellyfinFields.title = title;
jellyfinFields.title = titleEntries[0][1];
}
}
// IDs // IDs
if (metadata.imdbid) jellyfinFields.imdbid = metadata.imdbid; if (metadata.imdbid) jellyfinFields.imdbid = metadata.imdbid;
@@ -195,6 +195,16 @@ export class UnifiedProcessingStateManager {
performanceMetrics.recordHashProcessing(hashProcessingTime); performanceMetrics.recordHashProcessing(hashProcessingTime);
} }
// File-level metric, one per file that finished cleanly. Nothing recorded
// it before, so the monitor's "Processed" tile stayed at 0. Files announced
// over SSE are never marked discovered, so fall back to the light phase.
if (!error) {
const fileTime = totalProcessingTime
?? (hashState?.lightProcessingStartedAt !== undefined ? now - hashState.lightProcessingStartedAt : hashProcessingTime)
?? 0;
performanceMetrics.recordFileProcessing(filePath, fileTime);
}
const doneState: UnifiedFileState = { const doneState: UnifiedFileState = {
filePath, filePath,
state: 'done', state: 'done',
@@ -8,6 +8,7 @@
import type { HashMeta } from '@metazla/meta-interface'; import type { HashMeta } from '@metazla/meta-interface';
import type { IKVClient } from '../kv/IKVClient.js'; import type { IKVClient } from '../kv/IKVClient.js';
import type { FileTuplesResult } from '../kv/MetaCoreApiWriter.js';
import type { FileAnalyzerInterface } from '../logic/fileProcessor/FileAnalyzerInterface.js'; import type { FileAnalyzerInterface } from '../logic/fileProcessor/FileAnalyzerInterface.js';
import type { ContainerTask } from '../container-plugins/types.js'; import type { ContainerTask } from '../container-plugins/types.js';
import type { UnifiedProcessingSnapshot } from '../logic/UnifiedProcessingStateManager.js'; import type { UnifiedProcessingSnapshot } from '../logic/UnifiedProcessingStateManager.js';
@@ -41,6 +42,21 @@ export function hasPublish(client: IKVClient): client is IKVClientWithPubSub {
return typeof (client as IKVClientWithPubSub).publish === 'function'; return typeof (client as IKVClientWithPubSub).publish === 'function';
} }
/**
* IKVClient that can list meta-core's file-backed roots in one call
* (GET /api/files/tuples). Resolves null when meta-core predates the endpoint.
*/
export interface IKVClientWithFileTuples extends IKVClient {
getFileTuples(opts?: { summary?: boolean }): Promise<FileTuplesResult | null>;
}
/**
* Type guard to check if a KVClient can list file-backed roots
*/
export function hasFileTuples(client: IKVClient): client is IKVClientWithFileTuples {
return typeof (client as IKVClientWithFileTuples).getFileTuples === 'function';
}
/** /**
* IKVClient with Redis Streams capabilities * IKVClient with Redis Streams capabilities
* *
+1 -1
View File
@@ -9,7 +9,7 @@ export interface FileMetadata {
// Filename parsing // Filename parsing
title?: string; title?: string;
titles?: string[]; titles?: Record<string, Record<string, true | string>>; // key-set: lang3 → {name: "true"}
originaltitle?: string; originaltitle?: string;
showtitle?: string; showtitle?: string;
season?: number; season?: number;
@@ -1,164 +1,53 @@
import { useState, useEffect } from 'react'; import { useEffect } from 'react';
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
// @ts-ignore -- plain ES module, no .d.ts; imported for its side effect
// of defining the custom element.
import '../meta-service-menu.js';
interface ServiceInfo { /**
name: string; * ServiceNav — the neighbouring meta-* services, as a burger menu.
hostname: string; *
baseUrl: string; * Since meta-discovery v1 this is a thin wrapper around <meta-service-menu>,
status: string; * a framework-free custom element shared by every meta-* UI (see
lastHeartbeat: string; * scripts/check-mirrors.sh). The old implementation was ~165 lines of React
} * that existed in four near-identical copies and polled meta-core's
* /api/services; the element polls this service's own /api/neighbors instead,
* so the nav still renders when meta-core is down.
*
* The element lives in shadow DOM and cannot see this app's CSS variables, so
* the four --mm-nav-* tokens below map our palette onto it.
*/
interface ServicesResponse { declare global {
services: ServiceInfo[]; // eslint-disable-next-line @typescript-eslint/no-namespace
current: string; namespace JSX {
} interface IntrinsicElements {
'meta-service-menu': React.DetailedHTMLProps<
const serviceIcons: Record<string, string> = { React.HTMLAttributes<HTMLElement>,
'meta-sort': '📁', HTMLElement
'meta-fuse': '🗂️', > & { current?: string; endpoint?: string; label?: string };
'meta-stremio': '🎬', }
'meta-core': '⚙️', }
'meta-dup': '🔍',
'meta-orbit': '🌐',
'default': '📦'
};
function formatServiceName(name: string): string {
return name.split('-').map(word =>
word.charAt(0).toUpperCase() + word.slice(1)
).join(' ');
} }
function ServiceNav() { function ServiceNav() {
const [services, setServices] = useState<ServiceInfo[]>([]);
const [loading, setLoading] = useState(true);
const currentService = 'meta-sort';
useEffect(() => { useEffect(() => {
const fetchServices = async () => { // Import above is for the side effect of defining the element; this
try { // keeps bundlers from tree-shaking it away in production builds.
const response = await fetch('/api/services');
if (response.ok) {
const data: ServicesResponse = await response.json();
setServices(data.services || []);
}
} catch (error) {
console.error('Failed to fetch services:', error);
} finally {
setLoading(false);
}
};
fetchServices();
const interval = setInterval(fetchServices, 30000);
return () => clearInterval(interval);
}, []); }, []);
if (loading) return null;
const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name));
if (sortedServices.length === 0) return null;
return ( return (
<nav className="services-nav"> <>
<span className="services-nav-label">Services:</span> <meta-service-menu current="meta-sort" />
<div className="services-nav-items">
{sortedServices.map(service => {
const icon = serviceIcons[service.name] || serviceIcons.default;
const isActive = service.name === currentService;
return (
<a
key={service.name}
href={isActive ? '#' : service.baseUrl}
className={`service-link${isActive ? ' active' : ''}`}
onClick={(e) => {
e.preventDefault();
if (!isActive) {
window.location.href = service.baseUrl;
}
}}
>
<span className="service-icon">{icon}</span>
<span>{formatServiceName(service.name)}</span>
<span className="service-status"></span>
</a>
);
})}
</div>
<style>{` <style>{`
.services-nav { meta-service-menu {
background: var(--bg-secondary); --mm-nav-fg: var(--text-primary, #e0e0e0);
border-radius: 12px; --mm-nav-bg: var(--bg-tertiary, var(--bg-secondary, #1a1a2e));
padding: 0.5rem; --mm-nav-border: var(--border-color, rgba(255,255,255,0.14));
margin-bottom: 1.5rem; --mm-nav-accent: var(--accent-primary, #4ecdc4);
display: flex;
align-items: center;
gap: 0.5rem;
flex-wrap: wrap;
border: 1px solid var(--border-color);
}
.services-nav-label {
color: var(--text-secondary);
font-size: 0.85rem;
padding: 0.5rem 0.75rem;
white-space: nowrap;
}
.services-nav-items {
display: flex;
gap: 0.5rem;
flex-wrap: wrap;
}
.service-link {
display: inline-flex;
align-items: center;
gap: 0.4rem;
padding: 0.5rem 1rem;
border-radius: 8px;
text-decoration: none;
font-size: 0.9rem;
font-weight: 500;
transition: all 0.2s;
color: var(--text-primary);
background: var(--bg-tertiary);
border: 1px solid var(--border-color);
}
.service-link:hover {
background: var(--border-color);
border-color: rgba(78, 205, 196, 0.5);
text-decoration: none;
}
.service-link.active {
background: linear-gradient(135deg, rgba(78, 205, 196, 0.2), rgba(68, 160, 141, 0.2));
border-color: rgba(78, 205, 196, 0.5);
}
.service-link .service-icon {
font-size: 1.1rem;
}
.service-link .service-status {
width: 6px;
height: 6px;
border-radius: 50%;
background: var(--accent-primary);
margin-left: 0.25rem;
}
@media (max-width: 600px) {
.services-nav {
flex-direction: column;
align-items: flex-start;
}
} }
`}</style> `}</style>
</nav> </>
); );
} }
@@ -0,0 +1,308 @@
/**
* <meta-service-menu> — the neighbouring meta-* services, as a burger menu.
*
* Replaces the eight divergent "services bar" implementations (four copies of
* a React `ServiceNav.tsx`, two mirrored `services.js`, two bespoke inline
* ones) with a single framework-free custom element. It drops into the React
* dashboards as a plain tag — no React dependency — and into the vanilla UIs
* as an ES module.
*
* Data comes from the host service's own `GET /api/neighbors`, same-origin.
* Never cross-origin: the dev stack's self-signed certs and the OIDC redirects
* make a fetch to a neighbour a dead end. Links are plain navigations.
*
* ⚠ MIRRORED FILE. Byte-identical copies live in every meta-* UI tree; see
* `scripts/check-mirrors.sh`. Edit one → edit all.
*
* Shadow DOM is not optional here. The eight UIs use four incompatible token
* vocabularies, and two of them define *different colours under the same
* names* (`--bg-secondary` is #12121a in meta-sort, #16213e in meta-core,
* #161b22 in meta-dup). meta-search and meta-share are also the only
* light-mode-capable UIs, so a hardcoded dark palette would break them.
* The element therefore styles itself from four tokens with dark defaults:
*
* --mm-nav-fg, --mm-nav-bg, --mm-nav-border, --mm-nav-accent
*
* Each host maps its own variables onto those, e.g.
* meta-service-menu { --mm-nav-bg: var(--bg-secondary); }
*
* Attributes:
* current name of the host service, so it can be marked and not linked
* endpoint override the default "/api/neighbors"
* label button tooltip (default "Services")
*
* Property:
* rewriteUrl optional (name, url, neighbour) => url, for hosts that must
* translate internal hostnames into browser-reachable ones
* (meta-gateway does this for its dev/prod split).
*/
const ICONS = {
"meta-core": "⚙️",
"meta-sort": "📁",
"meta-fuse": "🗂️",
"meta-stremio": "🎬",
"meta-dup": "🔍",
"meta-search": "🔎",
"meta-share": "🔗",
"meta-gateway": "🌉",
"meta-watch": "📺",
"meta-listen": "🎧",
"meta-read": "📖",
};
const DEFAULT_ICON = "📦";
const POLL_MS = 30000;
function titleCase(name) {
return String(name || "")
.split("-")
.map((w) => w.charAt(0).toUpperCase() + w.slice(1))
.join(" ");
}
const TEMPLATE = `
<style>
:host {
--_fg: var(--mm-nav-fg, #e8eaed);
--_bg: var(--mm-nav-bg, #1a1a2e);
--_border: var(--mm-nav-border, rgba(255, 255, 255, 0.14));
--_accent: var(--mm-nav-accent, #4ecdc4);
position: relative;
display: inline-block;
font-family: inherit;
font-size: 14px;
line-height: 1.4;
}
button.trigger {
display: inline-flex;
align-items: center;
justify-content: center;
gap: 0.4rem;
width: 2.1rem;
height: 2.1rem;
padding: 0;
cursor: pointer;
color: var(--_fg);
background: transparent;
border: 1px solid var(--_border);
border-radius: 8px;
font: inherit;
transition: border-color 0.15s, background 0.15s;
}
button.trigger:hover { border-color: var(--_accent); }
button.trigger:focus-visible {
outline: 2px solid var(--_accent);
outline-offset: 2px;
}
button.trigger[aria-expanded="true"] { border-color: var(--_accent); }
.bars { display: block; width: 15px; }
.bars span {
display: block;
height: 2px;
margin: 3px 0;
border-radius: 2px;
background: currentColor;
}
.panel {
position: absolute;
top: calc(100% + 6px);
right: 0;
z-index: 1000;
min-width: 13rem;
max-height: 70vh;
overflow-y: auto;
padding: 0.3rem;
background: var(--_bg);
border: 1px solid var(--_border);
border-radius: 10px;
box-shadow: 0 10px 30px rgba(0, 0, 0, 0.35);
}
.panel[hidden] { display: none; }
a.item, .item {
display: flex;
align-items: center;
gap: 0.55rem;
padding: 0.45rem 0.6rem;
border-radius: 6px;
color: var(--_fg);
text-decoration: none;
white-space: nowrap;
}
a.item:hover { background: color-mix(in srgb, var(--_accent) 18%, transparent); }
.item.active { color: var(--_accent); font-weight: 600; cursor: default; }
.item.dead { opacity: 0.55; cursor: not-allowed; }
.icon { font-size: 1rem; line-height: 1; }
.dot {
width: 6px;
height: 6px;
margin-left: auto;
border-radius: 50%;
background: var(--_accent);
}
.empty {
padding: 0.6rem;
opacity: 0.7;
font-size: 0.85em;
white-space: nowrap;
}
/* Respect a host that opts out of motion. */
@media (prefers-reduced-motion: reduce) {
button.trigger { transition: none; }
}
</style>
<button class="trigger" type="button" aria-haspopup="menu" aria-expanded="false">
<span class="bars" aria-hidden="true"><span></span><span></span><span></span></span>
</button>
<div class="panel" role="menu" hidden></div>
`;
class MetaServiceMenu extends HTMLElement {
constructor() {
super();
this.attachShadow({ mode: "open" });
this.shadowRoot.innerHTML = TEMPLATE;
this._btn = this.shadowRoot.querySelector("button.trigger");
this._panel = this.shadowRoot.querySelector(".panel");
this._neighbors = [];
this._timer = null;
this._open = false;
this._onDocClick = (e) => {
if (!this._open) return;
if (e.composedPath().includes(this)) return;
this._setOpen(false);
};
this._onKey = (e) => {
if (e.key === "Escape" && this._open) {
this._setOpen(false);
this._btn.focus();
}
};
}
connectedCallback() {
this._btn.title = this.getAttribute("label") || "Services";
this._btn.setAttribute("aria-label", this._btn.title);
this._btn.addEventListener("click", () => this._setOpen(!this._open));
document.addEventListener("click", this._onDocClick);
document.addEventListener("keydown", this._onKey);
this.refresh();
this._timer = setInterval(() => this.refresh(), POLL_MS);
}
disconnectedCallback() {
if (this._timer) clearInterval(this._timer);
this._timer = null;
document.removeEventListener("click", this._onDocClick);
document.removeEventListener("keydown", this._onKey);
}
get endpoint() {
return this.getAttribute("endpoint") || "/api/neighbors";
}
get current() {
return this.getAttribute("current") || "";
}
async refresh() {
try {
const res = await fetch(this.endpoint, {
headers: { Accept: "application/json" },
});
if (!res.ok) throw new Error(String(res.status));
const data = await res.json();
// `services` is the legacy alias some backends still emit.
this._neighbors = data.neighbors || data.services || [];
} catch {
// Silent: the menu is navigation, not a health indicator. Keep whatever
// we last had rather than blanking the list on one failed poll.
}
this._render();
}
_setOpen(open) {
this._open = open;
this._panel.hidden = !open;
this._btn.setAttribute("aria-expanded", String(open));
if (open) this.refresh();
}
_render() {
const panel = this._panel;
panel.textContent = "";
const seen = new Set();
const rows = [];
for (const n of this._neighbors) {
const name = n.name || "";
if (!name || seen.has(name)) continue;
// Beacon v2 neighbours include plugins; the nav lists services only.
// (A row without `caps` comes from a pre-v2 backend — keep it.)
const isService = (c) => String(c).startsWith("metamesh.service/");
if (Array.isArray(n.caps) && !n.caps.some(isService)) continue;
seen.add(name);
rows.push(n);
}
// Always show the host itself, even before its own echo comes back.
if (this.current && !seen.has(this.current)) {
rows.push({ name: this.current, baseUrl: "" });
}
rows.sort((a, b) => String(a.name).localeCompare(String(b.name)));
if (rows.length === 0) {
const empty = document.createElement("div");
empty.className = "empty";
empty.textContent = "No services found";
panel.appendChild(empty);
return;
}
for (const n of rows) {
const isActive = n.name === this.current;
let url = n.baseUrl || "";
if (typeof this.rewriteUrl === "function") {
try {
url = this.rewriteUrl(n.name, url, n) || "";
} catch {
/* a bad hook must not blank the menu */
}
}
// A neighbour with no reachable URL is shown but not linked — better
// than a link that goes nowhere.
const el = document.createElement(isActive || !url ? "span" : "a");
el.className =
"item" + (isActive ? " active" : "") + (!isActive && !url ? " dead" : "");
if (el.tagName === "A") {
el.href = url;
el.setAttribute("role", "menuitem");
}
if (!isActive && !url) el.title = "No reachable URL advertised";
const icon = document.createElement("span");
icon.className = "icon";
icon.textContent = ICONS[n.name] || DEFAULT_ICON;
const label = document.createElement("span");
label.textContent = titleCase(n.name);
el.append(icon, label);
// Beacon v2 omits `status` when running.
if ((n.status || "running") === "running" || isActive) {
const dot = document.createElement("span");
dot.className = "dot";
dot.title = n.status || "running";
el.appendChild(dot);
}
panel.appendChild(el);
}
}
}
if (!customElements.get("meta-service-menu")) {
customElements.define("meta-service-menu", MetaServiceMenu);
}
export default MetaServiceMenu;
+29 -23
View File
@@ -1,4 +1,4 @@
import { useState, useEffect, useCallback } from 'react'; import { useState, useEffect, useCallback, useRef } from 'react';
import { import {
ProcessingStatus, ProcessingStatus,
Metrics, Metrics,
@@ -60,29 +60,35 @@ function Monitor() {
const [failedFiles, setFailedFiles] = useState<FailedFile[]>([]); const [failedFiles, setFailedFiles] = useState<FailedFile[]>([]);
const [showPipelineHelp, setShowPipelineHelp] = useState(false); const [showPipelineHelp, setShowPipelineHelp] = useState(false);
const fetchData = useCallback(async () => { // Each endpoint renders as soon as it answers, so one slow or failing call
try { // cannot blank the page. A tick is skipped while the previous one is still
const [statusRes, metricsRes, queueRes, statsRes, failedRes] = await Promise.all([ // out, so a slow backend never collects a pile of overlapping polls.
fetch('/api/processing/status'), const pollInFlight = useRef(false);
fetch('/api/metrics'),
fetch('/api/processing/queue'),
fetch('/api/stats'),
fetch('/api/processing/failed')
]);
if (statusRes.ok) setStatus(await statusRes.json()); const fetchData = useCallback(async () => {
if (metricsRes.ok) setMetrics(await metricsRes.json()); if (pollInFlight.current) return;
if (queueRes.ok) { pollInFlight.current = true;
const queueData = await queueRes.json();
setQueue(queueData.items || []); const load = async <T,>(url: string, apply: (data: T) => void) => {
const res = await fetch(url);
if (res.ok) apply(await res.json());
};
try {
const results = await Promise.allSettled([
load<ProcessingStatus>('/api/processing/status', setStatus),
load<Metrics>('/api/metrics', setMetrics),
load<{ items?: QueueItem[] }>('/api/processing/queue', (data) => setQueue(data.items || [])),
load<RedisStats>('/api/stats', setRedisStats),
load<{ failedFiles?: FailedFile[] }>('/api/processing/failed', (data) => setFailedFiles(data.failedFiles || [])),
]);
for (const result of results) {
if (result.status === 'rejected') {
console.error('Failed to fetch data:', result.reason);
} }
if (statsRes.ok) setRedisStats(await statsRes.json());
if (failedRes.ok) {
const failedData = await failedRes.json();
setFailedFiles(failedData.failedFiles || []);
} }
} catch (err) { } finally {
console.error('Failed to fetch data:', err); pollInFlight.current = false;
} }
}, []); }, []);
@@ -410,7 +416,7 @@ function Monitor() {
<h3 className="section-title">File Coverage</h3> <h3 className="section-title">File Coverage</h3>
<div className="stats-grid"> <div className="stats-grid">
<div className="stat-box"> <div className="stat-box">
<div className="stat-value">{status?.uniqueHashes ? formatNumber(status.uniqueHashes) : (redisStats ? formatNumber(redisStats.fileCount) : '-')}</div> <div className="stat-value">{status?.uniqueHashes ? formatNumber(status.uniqueHashes) : (redisStats?.fileCount != null ? formatNumber(redisStats.fileCount) : '-')}</div>
<div className="stat-label">Unique Files</div> <div className="stat-label">Unique Files</div>
</div> </div>
<div className="stat-box"> <div className="stat-box">
@@ -432,7 +438,7 @@ function Monitor() {
<div className="card total-size-card"> <div className="card total-size-card">
<h2>Total Size</h2> <h2>Total Size</h2>
<div className="total-size-value"> <div className="total-size-value">
{redisStats ? formatBytes(redisStats.totalSize) : '-'} {redisStats?.totalSize != null ? formatBytes(redisStats.totalSize) : '-'}
</div> </div>
<div className="total-size-label">Watch Folders</div> <div className="total-size-label">Watch Folders</div>
</div> </div>
+5 -5
View File
@@ -184,12 +184,12 @@ export interface Metrics {
// NOTE: DuplicateGroup and DuplicateData types have been moved to meta-dup service // NOTE: DuplicateGroup and DuplicateData types have been moved to meta-dup service
/** GET /api/stats: totals over meta-core's file-backed roots. */
export interface RedisStats { export interface RedisStats {
fileCount: number; /** null when meta-core cannot supply file totals (source 'unavailable') */
keyCount: number; fileCount: number | null;
totalSize: number; totalSize: number | null;
memoryUsage?: string; source?: 'meta-core' | 'unavailable';
memoryUsageBytes?: number;
} }
export interface QueueItem { export interface QueueItem {
@@ -1,10 +1,17 @@
/** /**
* Multicodec / multihash codes for every digest meta-hash can compute. * Multicodec / multihash codes for every digest meta-hash can compute.
* *
* Source of truth for these numbers, mirrored in: * Source of truth for these numbers. The same codes appear in the CID utility
* RANK, which is a separate thing this file does not implement:
* - `packages/meta-core/internal/cid/rank.go` * - `packages/meta-core/internal/cid/rank.go`
* - the `cid_rank.rs` in each of meta-search and meta-share * - the `cid_rank.rs` in each of meta-search and meta-share
* - `/cid-rank-vectors.json` (the golden fixture that pins them all together) * - `/cid-rank-vectors.json` (the golden fixture pinning those three together)
*
* ⚠ This package used to vendor that fixture at `src/tests/cid-rank-vectors.json`
* and no test ever loaded it, so every vector added cost a commit here for
* nothing. Removed 2026-09-09. If a TypeScript rank implementation is ever
* written, re-add the path to `scripts/check-cid-vectors.sh` *and* a test that
* asserts it — a vendored fixture nobody reads is not a guard.
* *
* ⚠ meta-hash encodes `codec == multihash code` (see `ComputeHash.cidFinalize`: * ⚠ meta-hash encodes `codec == multihash code` (see `ComputeHash.cidFinalize`:
* `CID.createV1(code, create(code, digest))`), whereas the fullhash plugin and * `CID.createV1(code, create(code, digest))`), whereas the fullhash plugin and
@@ -1,178 +0,0 @@
{
"$schema": "cid-rank-vectors/v1",
"$comment": [
"CANONICAL golden vectors for the CID utility rank. THE single source of truth.",
"The rank decides which of a record's several CIDs is elected canonical - the cid a",
"record is advertised and addressed by. It is reimplemented in Go, 2x Rust and 3x TS;",
"they cannot share code (independent submodules, no cargo path-dep). These vectors are",
"the only thing holding them in sync: every implementation's test suite loads this file.",
"",
"This file is VENDORED into each consuming submodule (submodule CI cannot read above its",
"own tree). scripts/check-cid-vectors.sh diffs every copy against this one.",
"",
"Ranking is by UTILITY for external retrieval, not cryptographic strength.",
"Ties break on the lexicographically smaller cid, so the election is deterministic",
"across peers and does not churn.",
"",
"See METADATA_KEYS.md section 2/14.13 and docs/cid-formats.md."
],
"ladder": {
"40": "dag-pb (IPFS block)",
"30": "sha2-256 / sha3-256 / sha3-384 - real content digests",
"25": "nzb-posting - content-derived (Message-ID set digest), but not itself fetchable",
"20": "BitTorrent infohashes - swarm interop",
"10": "midhash256 - the record address",
"5": "opaque locators (nzb-release, url, card, yt-video, ext-play) - never outrank a real digest",
"0": "weak/unknown digests, unparseable input"
},
"tieBreak": "lexicographically smaller cid wins",
"vectors": [
{
"name": "ipfs-dagpb",
"cid": "bafybeiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 40,
"note": "dag-pb codec 0x70 - chunked IPFS file, broadest network reach"
},
{
"name": "sha2-256-raw-codec",
"cid": "bafkreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 30,
"note": "fullhash/gateway convention: raw codec 0x55, mh names the algo"
},
{
"name": "sha2-256-codec-eq-mh",
"cid": "baejbeibku6ua2l6r5olnggx4pto2sii2s5jeeawgh4cbkhlbmp52gudbua",
"rank": 30,
"note": "REAL meta-hash fixture; meta-hash sets codec == mh"
},
{
"name": "sha3-256",
"cid": "baelbmidner7x3dd7t2kldh6wiq4w6lsye7qrwo6uoqot2t52zkhqk56fnq",
"rank": 30,
"note": "REAL meta-hash fixture"
},
{
"name": "sha3-384",
"cid": "baekrkmeglipkjf33exqexdvijl4twipruay4wsbvt44akce43lq5rgccxgr7v6dwksmj3sxedakevc53g6ga",
"rank": 30,
"note": "REAL meta-hash fixture"
},
{
"name": "nzb-posting",
"cid": "bagbsaeraaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 25,
"note": "codec 0x1003 + REAL sha2-256 multihash (0x12) over the normalised, sorted Message-ID set - NOT a locator, so it sits ABOVE tier 5, but it addresses no fetchable block (bitswap can never satisfy a want for it - the digest is over an id set, not over any block's bytes), so it stays BELOW a real content digest (30): once meta-share materialises the media and re-seeds it, that genuine sha2-256 cid must win the canonical election, not this one. Rank MUST match on codec BEFORE falling through to the mh switch, or this scores 30 (same trap as nzb-release/url, in reverse: those need a codec check because their mh is identity 0x00; this needs one because its mh is REAL sha2-256 and would otherwise tie with tier 30)."
},
{
"name": "btih-v2-codec-eq-mh",
"cid": "bag3sdnzbeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "meta-hash encoder convention"
},
{
"name": "btih-v2-raw-codec",
"cid": "bafk3oijaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "fullhash encoder convention - SAME tier, both conventions must match"
},
{
"name": "btih-v1-file",
"cid": "bagasbajacqaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "gateway per-file BitTorrent v1"
},
{
"name": "btih-v2-file",
"cid": "bagbcbaraeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 20,
"note": "gateway per-file BitTorrent v2"
},
{
"name": "midhash256-codec-eq-mh",
"cid": "bagacbabaeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 10,
"note": "the record address; internally meaningful, externally invisible"
},
{
"name": "midhash256-raw-codec",
"cid": "bafkyaibaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 10
},
{
"name": "nzb-release-locator",
"cid": "bagcsaaazcbqxa2jonz5gez3fmvvs42lomzx55ln657fp4aab",
"rank": 5,
"note": "codec 0x1005 + IDENTITY multihash 0x00. The locator code is in the CODEC slot, not the mh slot - rank MUST match on codec. Tier 5 = below midhash: a locator never outranks a real digest, it wins only as a record's sole cid."
},
{
"name": "url-locator",
"cid": "bagdcaaaznb2hi4dthixs6zlymfwxa3dffzxxezzpmexhazdg",
"rank": 5,
"note": "codec 0x1006 + IDENTITY multihash 0x00. Same tier/reasoning as nzb-release."
},
{
"name": "card-locator",
"cid": "bagdsaaanar2g2zdcor3duojvgq3ts",
"rank": 5,
"note": "codec 0x1007 + IDENTITY multihash 0x00, digest = varint(len(source)) || source || id — here (\"tmdb\",\"tv:95479\"). Same codec-slot rule as nzb-release/url: match on the CODEC, never the mh. The limit case of the family — the only CID in MetaMesh with NO bytes behind it anywhere, ever: it addresses a *work* (a series, a film), not a file, so the byte path 404s by design and bitswap never offers it (seeding is gated on sha2-256). Tier 5 is load-bearing for the planned cross-source merge: one card record may carry BOTH a TMDB and a MyAnimeList locator, and if a real card body is ever added (see card-tier-search.md §4.3) its sha2-256 must win the canonical election over both locators without any addressing change. Ranking it 0 instead would still lose to a digest, but would put the implementations back out of agreement — which is the exact failure this fixture exists to prevent."
},
{
"name": "yt-video-locator",
"cid": "bagecaaarav3gszdfn42ein3vgvfumn2tka4a",
"rank": 5,
"note": "codec 0x1008 + IDENTITY multihash 0x00, digest = varint(len(kind)) || kind || id — here (\"video\",\"4D7u5KF7SP8\"). DELEGATED PLAYBACK: the limit case beyond card. A card addresses a work with no bytes anywhere; this addresses a *rendition* whose bytes exist but are permanently someone else's — an external player renders it, nothing is transported, seeded or content-addressed. Framing is byte-identical to card (0x1007) on purpose, so each implementation extends the parser it already wrote. Same codec-slot rule: match on the CODEC, never the mh (identity 0x00 would fall through and score 0). Tier 5 is load-bearing for the on-ramp: a delegated reference and a real archived copy can sit on ONE record, and that sha2-256 must win the canonical election over this locator with no addressing change. NOTE the byte path must REFUSE this (400), not 404 — see docs/cid-formats.md §7.4."
},
{
"name": "ext-play-locator",
"cid": "bagesaabenb2hi4dthixs6zlymfwxa3dffzrgc3temnqw24bomnxw2l3bnrrhk3jppa",
"rank": 5,
"note": "codec 0x1009 + IDENTITY multihash 0x00, digest = the verbatim UTF-8 URL. WIRE-IDENTICAL to url (0x1006) apart from the codec, and that difference is the entire contract: 0x1006 means 'fetch these bytes once and seed them', 0x1009 means 'open this in a browser, there are no bytes for us'. A decoder that matched on SHAPE rather than on the codec slot would see an identity multihash wrapping an http(s) URL, take the 0x1006 path, fetch a watch/Bandcamp page, chunk it, seed the HTML and link that content CID onto the record AS THE FILE — the MSR1 poisoning shape, except permanent and replicated. Same tier/reasoning as the rest of the locator family."
},
{
"name": "sha1",
"cid": "baeircff3eqvxxaqqot2lt242stkdtk3gq4c2vra",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "md5",
"cid": "bahkqdvibcatpcfl7ine2n625eb5ml5mzkfaa",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "crc32",
"cid": "bagzafmqcathsbnrk",
"rank": 0,
"note": "REAL meta-hash fixture; weak digest"
},
{
"name": "bt-pieces-root",
"cid": "bafkyf3qceaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 0,
"note": "torrent-internal merkle root, NOT a file-identity cid"
},
{
"name": "cidv0-rejected",
"cid": "QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG",
"rank": 0,
"note": "TRAP: Go rejected this (rank 0) while Rust's MsCid::from_str ACCEPTED it and ranked it 40 (dag-pb). Strict parse = CIDv1 base32-lower only. This vector pins the fix."
},
{
"name": "token-form-rejected",
"cid": "midhash256:bagacbabaeaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"rank": 0,
"note": "TRAP: the deprecated <algo>:<cid> token form. Rust used to strip the prefix and rank it 10. It must now be rejected."
},
{
"name": "garbage",
"cid": "not-a-cid",
"rank": 0
},
{
"name": "empty",
"cid": "",
"rank": 0
}
]
}
@@ -22,13 +22,13 @@ describe('MetaData', () => {
}); });
it('should set and get nested metadata using nested keys', () => { it('should set and get nested metadata using nested keys', () => {
root.at('titles').at('eng').set("Star Wars"); root.at('titles').at('eng').at('Star Wars').set(true);
root.at('titles').at('jpn').set("Staru Warsu"); root.at('titles').at('jpn').at('Staru Warsu').set(true);
expect(root.at('titles').at('eng').get()).to.equal("Star Wars"); expect(root.at('titles').at('eng').at('Star Wars').get()).to.equal(true);
expect(root.at('titles').at('jpn').get()).to.equal("Staru Warsu"); expect(root.at('titles').at('jpn').at('Staru Warsu').get()).to.equal(true);
expect(root.at('titles').get()).to.deep.equal({eng: "Star Wars", jpn: "Staru Warsu"}); expect(root.at('titles').get()).to.deep.equal({eng: {"Star Wars": true}, jpn: {"Staru Warsu": true}});
expect(root.get()).to.deep.equal({ expect(root.get()).to.deep.equal({
titles: {eng: "Star Wars", jpn: "Staru Warsu"} titles: {eng: {"Star Wars": true}, jpn: {"Staru Warsu": true}}
}); });
}); });
@@ -1,10 +1,25 @@
import {TypedRecordSet} from "./basic/RecordSet.js"; import {TypedRecordSet} from "./basic/RecordSet.js";
import {ISO639_1_CODE} from "./basic/LanguageData.js"; import {LANG3_CODE} from "./basic/LanguageData.js";
/**
* The language triad (METADATA_KEYS.md §9).
*
* `languages` is the **union** — every language the record offers by any route
* — and the only one a query filters on. The two splits refine it and are
* optional. Rule #7: a writer that adds a member to a split MUST add the same
* member to the union in the same write; nothing reconciles them afterwards.
*/
export interface LanguagesMeta{ export interface LanguagesMeta{
languages?: TypedRecordSet<ISO639_1_CODE>; /** The union: audio ∪ subtitles ∪ bare language statements. Filterable. */
languages?: TypedRecordSet<LANG3_CODE>;
/** Spoken/audio-track languages only. Refinement, never filtered on. */
audioLanguages?: TypedRecordSet<LANG3_CODE>;
/** Subtitle languages only (§8). Refinement, never filtered on. */
subtitleLanguages?: TypedRecordSet<LANG3_CODE>;
} }
export const LanguagesMetaFields = [ export const LanguagesMetaFields = [
"languages", "languages",
"audioLanguages",
"subtitleLanguages",
]; ];
@@ -8,8 +8,9 @@ export interface VideoMeta {
increment?: string; increment?: string;
movieYear?: string; movieYear?: string;
videoType?: VideoType; videoType?: VideoType;
title?: string;//display title (localized); the one name to show
originalTitle?: string;//title in original language originalTitle?: string;//title in original language
titles?: LanguageData<string>;// <ISO 639-3 lang codes, title> titles?: LanguageData<Record<string, true>>;// key-set: <ISO 639-3 lang code, {name: true}> — every clean name incl. AKAs (METADATA_KEYS.md §3)
} }
export const VideoMetaFields = [ export const VideoMetaFields = [
@@ -1,9 +1,21 @@
/** /**
* key is language ISO 639-3 code * key is a `lang3` code: ISO 639-2/**B** — `eng`, `jpn`, `fre`/`ger`/`chi`,
* https://en.wikipedia.org/wiki/List_of_ISO_639-3_codes * NOT the 639-2/T (= 639-3) spellings `fra`/`deu`/`zho`.
*
* This is the vocabulary the stack compares against: `languages:` query filters
* match it literally, and `meta_feeder_sdk::lang::normalize_lang_code` folds
* incoming codes onto it. See METADATA_KEYS.md §"Value formats" and §14.3 —
* the card tier (`titles/*`, `posters/*`) is still on /T and has not converged.
* https://en.wikipedia.org/wiki/List_of_ISO_639-2_codes
*/ */
export type LanguageData<T> = Record<string, T>; export type LanguageData<T> = Record<string, T>;
export type ISO639_1_CODE = string; export type LANG3_CODE = string;
/**
* @deprecated Misnamed: no writer has ever emitted 2-letter ISO 639-1 codes
* here. Use {@link LANG3_CODE}. Kept as an alias so existing imports compile.
*/
export type ISO639_1_CODE = LANG3_CODE;
/** /**
* jpn, jpr and jpa already exist. Creating jpl is for japan-latin. Used for japanese romaji NON ISO 639-3 standard. * jpn, jpr and jpa already exist. Creating jpl is for japan-latin. Used for japanese romaji NON ISO 639-3 standard.