From c2420aa1faa5840aff875db9b4e608970a45c7fb Mon Sep 17 00:00:00 2001 From: Pierre Date: Sat, 21 Feb 2026 18:48:14 +0100 Subject: [PATCH] bug fix on file dispaly, document update --- README.md | 202 +++-- docs/api-reference.md | 671 ++++++++++++++++ docs/streaming-architecture.md | 455 +++++++++++ docs/vfs-rebuild-architecture.md | 736 ++++++++---------- packages/meta-fuse-core/src/index.ts | 18 +- packages/meta-fuse-core/src/kv/IKVClient.ts | 5 +- .../meta-fuse-core/src/kv/LeaderClient.ts | 13 +- packages/meta-fuse-core/src/kv/RedisClient.ts | 70 ++ .../src/vfs/StreamingStateBuilder.ts | 15 + .../src/vfs/VirtualFileSystem.ts | 156 +++- 10 files changed, 1818 insertions(+), 523 deletions(-) create mode 100644 docs/api-reference.md create mode 100644 docs/streaming-architecture.md diff --git a/README.md b/README.md index 794c16b..30d6363 100644 --- a/README.md +++ b/README.md @@ -92,30 +92,37 @@ Meta-Fuse is a read-only process in the MetaMesh ecosystem that: ### KV Leader Election -All services in MetaMesh use a shared KV wrapper with leader election via flock(2): +All services in MetaMesh use leader discovery via meta-core's lock file and HTTP API: ``` /meta-core/locks/ -└── kv-leader.lock # flock-based leader election + endpoint info +├── kv-leader.lock # flock-based leader election +└── kv-leader.info # Leader API URL (plain text) -Lock File Format (JSON): +Lock Info File Format (plain text): +http://meta-core:9000 + +/urls API Response (JSON): { - "host": "meta-sort-instance-1", - "api": "redis://10.0.1.50:6379", - "http": "http://10.0.1.50:3000", - "timestamp": 1703808000000, - "pid": 12345 + "hostname": "meta-core-dev", + "baseUrl": "http://localhost:8083", + "apiUrl": "http://meta-core:9000", + "redisUrl": "redis://meta-core:6379", + "webdavUrl": "http://localhost:8083/webdav", + "webdavUrlInternal": "http://meta-core:9000/webdav", + "isLeader": true } -Leader Election Flow: -1. meta-sort attempts exclusive flock on kv-leader.lock -2. Winner (leader) spawns Redis, writes endpoint to lock file -3. meta-fuse reads lock file to discover Redis endpoint -4. meta-fuse connects as read-only client -5. On leader failure, flock auto-releases, new leader elected +Leader Discovery Flow: +1. meta-core acquires flock on kv-leader.lock +2. Winner (leader) spawns Redis, writes API URL to kv-leader.info +3. meta-fuse reads kv-leader.info to find API URL +4. meta-fuse calls /urls API to get Redis URL and other endpoints +5. meta-fuse connects to Redis as read-only client +6. On leader failure, flock auto-releases, new leader elected ``` -**Note**: meta-fuse never becomes leader - it only reads the lock file to discover the active Redis endpoint managed by meta-sort. +**Note**: meta-fuse never becomes leader - it reads the lock file and calls the /urls API to discover the active Redis endpoint managed by meta-core. ## Core Features @@ -150,15 +157,26 @@ meta-fuse/ │ │ ├── src/ │ │ │ ├── api/ # Fastify REST API (APIServer) │ │ │ │ └── APIServer.ts +│ │ │ ├── config/ # Configuration management +│ │ │ │ └── ConfigStorage.ts │ │ │ ├── kv/ # KV client wrapper (FOLLOWER mode only) │ │ │ │ ├── IKVClient.ts # Read-only interface │ │ │ │ ├── KVManager.ts # Leader discovery, connection management -│ │ │ │ ├── LeaderDiscovery.ts # Lock file reading -│ │ │ │ └── RedisClient.ts # Redis connection with pub/sub +│ │ │ │ ├── LeaderClient.ts # Lock file reading + /urls API +│ │ │ │ ├── RedisClient.ts # Redis connection with Streams support +│ │ │ │ └── ServiceDiscovery.ts # Service discovery client │ │ │ ├── vfs/ # Virtual filesystem logic -│ │ │ │ ├── VirtualFileSystem.ts # In-memory VFS representation -│ │ │ │ ├── MetaDataToFolderStruct.ts # Folder organization -│ │ │ │ └── RenamingRule.ts # Virtual path rules +│ │ │ │ ├── VirtualFileSystem.ts # In-memory VFS representation +│ │ │ │ ├── StreamingStateBuilder.ts # Event-driven state management +│ │ │ │ ├── RulesPropertyExtractor.ts # Property filtering from rules +│ │ │ │ ├── MetaDataToFolderStruct.ts # Folder organization +│ │ │ │ ├── RenamingRule.ts # Virtual path rules +│ │ │ │ ├── defaults/ # Default renaming rules +│ │ │ │ ├── template/ # Template engine components +│ │ │ │ │ ├── TemplateEngine.ts # Variable interpolation +│ │ │ │ │ └── ConditionEvaluator.ts # Rule condition evaluation +│ │ │ │ └── types/ # TypeScript type definitions +│ │ │ │ └── RenamingRuleTypes.ts │ │ │ └── index.ts # Entry point │ │ ├── package.json │ │ └── tsconfig.json @@ -174,6 +192,11 @@ meta-fuse/ │ ├── src/ │ └── package.json │ +├── docs/ # Architecture documentation +│ ├── vfs-rebuild-architecture.md +│ ├── streaming-architecture.md +│ └── api-reference.md +│ ├── docker/ │ ├── nginx.conf # Reverse proxy config │ ├── wsgidav.yaml # WebDAV server config @@ -191,21 +214,27 @@ meta-fuse/ | Component | File | Purpose | |-----------|------|---------| -| Entry Point | `src/index.ts` | Initializes KV manager, VFS, API server, pub/sub listener | -| API Server | `src/api/APIServer.ts` | Fastify REST API for FUSE operations (readdir, getattr, read) | +| Entry Point | `src/index.ts` | Initializes KV manager, VFS, API server, stream consumer | +| API Server | `src/api/APIServer.ts` | Fastify REST API for FUSE operations and rules management | | KV Manager | `src/kv/KVManager.ts` | **FOLLOWER-only**: leader discovery, Redis connection, reconnection loop | -| Leader Discovery | `src/kv/LeaderDiscovery.ts` | Reads `/meta-core/locks/kv-leader.lock` to find active Redis | -| Redis Client | `src/kv/RedisClient.ts` | Wrapper around ioredis with pub/sub support | +| Leader Client | `src/kv/LeaderClient.ts` | Reads `/meta-core/locks/kv-leader.info`, calls `/urls` API | +| Redis Client | `src/kv/RedisClient.ts` | Wrapper around ioredis with Redis Streams support | +| Service Discovery | `src/kv/ServiceDiscovery.ts` | Discovers all MetaMesh services via service files | | KV Interface | `src/kv/IKVClient.ts` | Read-only interface (get, scan, subscribe) | | Virtual FS | `src/vfs/VirtualFileSystem.ts` | In-memory VFS representation with caching | +| Streaming State Builder | `src/vfs/StreamingStateBuilder.ts` | Processes `meta:events` stream, builds VFS state incrementally | +| Rules Property Extractor | `src/vfs/RulesPropertyExtractor.ts` | Extracts VFS-relevant properties from renaming rules | | Folder Organizer | `src/vfs/MetaDataToFolderStruct.ts` | Converts flat metadata to organized folder structure | | Renaming Rules | `src/vfs/RenamingRule.ts` | Rules for virtual path organization | +| Template Engine | `src/vfs/template/TemplateEngine.ts` | Variable interpolation for renaming templates | +| Condition Evaluator | `src/vfs/template/ConditionEvaluator.ts` | Evaluates rule conditions against file metadata | **Note**: Unlike meta-sort, meta-fuse's KVManager is simplified: - **Never spawns Redis** (always FOLLOWER) - **Only reads metadata** (read-only interface) -- **Discovers leader** via lock file at `/meta-core/locks/kv-leader.lock` -- **No service discovery registration** (minimal footprint) +- **Discovers leader** via lock file at `/meta-core/locks/kv-leader.info` +- **Calls `/urls` API** to get Redis URL, WebDAV URL, and other endpoints +- **Uses streaming mode** - processes `meta:events` stream for real-time updates ## Configuration @@ -286,12 +315,28 @@ services: | Method | Endpoint | Description | |--------|----------|-------------| -| GET | `/api/fuse/health` | Health check | +| **Health & Status** | +| GET | `/health` | Health check | +| GET | `/api/health` | Health check (alias) | +| GET | `/api/fuse/health` | Health check (alias) | | GET | `/api/fuse/stats` | Filesystem statistics | +| **FUSE Operations** | | POST | `/api/fuse/readdir` | List directory contents | | POST | `/api/fuse/getattr` | Get file/directory attributes | -| POST | `/api/fuse/read` | Read file content | | POST | `/api/fuse/exists` | Check path existence | +| POST | `/api/fuse/read` | Read file content (returns source path or WebDAV URL) | +| POST | `/api/fuse/metadata` | Get full metadata for a path | +| GET | `/api/fuse/files` | List all virtual files | +| GET | `/api/fuse/directories` | List all virtual directories | +| POST | `/api/fuse/refresh` | Trigger VFS refresh | +| **Renaming Rules** | +| GET | `/api/fuse/rules` | Get current renaming rules configuration | +| PUT | `/api/fuse/rules` | Update renaming rules configuration | +| POST | `/api/fuse/rules/preview` | Preview how files would be renamed | +| POST | `/api/fuse/rules/validate` | Validate a single rule | +| GET | `/api/fuse/rules/variables` | Get list of available template variables | +| **Service Discovery** | +| GET | `/api/services` | List all discovered MetaMesh services | ### Example Responses @@ -474,15 +519,19 @@ cargo build --release ### KV Data Structure -meta-fuse reads metadata stored by meta-sort using the `meta-sort:` prefix: +meta-fuse reads metadata stored by meta-sort using flat Redis keys: ``` -# File metadata stored by meta-sort (hash-based keys) -meta-sort:file:{midhash256}:title → "Inception" -meta-sort:file:{midhash256}:year → "2010" -meta-sort:file:{midhash256}:filePath → "media1/Movies/Inception (2010)/Inception.mkv" -meta-sort:file:{midhash256}:size → 4831838208 -meta-sort:file:{midhash256}:video/codec → "h265" +# File metadata stored by meta-sort (flat key format) +file:{hashId}/title → "Inception" +file:{hashId}/year → "2010" +file:{hashId}/filePath → "media1/Movies/Inception (2010)/Inception.mkv" +file:{hashId}/size → 4831838208 +file:{hashId}/video/codec → "h265" +file:{hashId}/titles/eng → "Inception" + +# File index for enumeration +file:__index__ → SET of all hashIds # VFS paths are computed dynamically from metadata # meta-fuse builds virtual paths like: @@ -490,28 +539,45 @@ meta-sort:file:{midhash256}:video/codec → "h265" # → resolves to sourcePath: /files/media1/Movies/Inception (2010)/Inception.mkv ``` +**Key Format**: `file:{hashId}/{property}` where: +- `hashId` is the midhash256 content identifier +- `property` can be flat (e.g., `title`) or nested (e.g., `titles/eng`) + **Note**: File paths in Redis are **relative to FILES_VOLUME** (`/files`). meta-fuse prepends the FILES_VOLUME path when resolving actual file locations. -### Real-Time Updates via Pub/Sub +### Real-Time Updates via Redis Streams -meta-fuse subscribes to Redis pub/sub channel `meta-sort:file:batch` for real-time metadata updates from meta-sort: +meta-fuse uses the `meta:events` Redis Stream for real-time metadata updates. The streaming architecture provides: +- **Reliable delivery** - Messages persist until consumed +- **Replay capability** - Can rebuild state from stream position 0 +- **Memory efficiency** - Only VFS-relevant properties are fetched ```typescript -interface BatchUpdateMessage { - timestamp: number; - changes: Array<{ - action: 'add' | 'update' | 'remove'; - hashId: string; - }>; +// Stream message format (meta:events) +interface StreamMessage { + id: string; // Stream entry ID (e.g., "1703808000000-0") + type: 'set' | 'del'; // Operation type + key: string; // Redis key (e.g., "file:abc123/title") + ts: string; // Timestamp } ``` -When meta-sort adds/updates/removes files, it publishes batch updates. meta-fuse processes these to: -1. Invalidate affected caches -2. Rebuild virtual paths for modified files -3. Update directory listings without full refresh +**Startup Sequence**: +1. **Streaming Bootstrap**: Replay `meta:events` stream from position 0 +2. **Build State**: Process each event, fetch only VFS-relevant properties +3. **Go Live**: Continue consuming new events from last processed position -This ensures the virtual filesystem stays synchronized with metadata changes in near real-time. +**Event Processing Pipeline**: +1. Parse key to extract `hashId` and `property` +2. Check if property is VFS-relevant (based on renaming rules) +3. If relevant, fetch property value from Redis +4. Update internal state and notify VFS +5. When file has `filePath`, it appears in VFS + +This streaming architecture enables: +- Sub-second startup (no HGETALL/SCAN) +- Memory-efficient state (~500 bytes/file) +- Real-time VFS updates as files are processed --- @@ -537,10 +603,13 @@ journalctl -u meta-fuse-driver -f ```bash # Check lock file exists -cat /data/apps/meta-core/db/redis.lock +cat /meta-core/locks/kv-leader.info + +# Check /urls API +curl http://localhost:8083/api/urls # Verify redis is running (on leader) -redis-cli ping +docker exec meta-core-dev redis-cli ping # Check meta-fuse logs docker logs meta-fuse | grep "KV" @@ -562,14 +631,20 @@ docker logs meta-fuse | grep "wsgidav" ### Files Not Appearing ```bash -# Check KV has data -redis-cli keys "meta-fuse:vfs:*" +# Check Redis has file data (flat key format) +docker exec meta-core-dev redis-cli keys "file:*" | head -20 + +# Check file index +docker exec meta-core-dev redis-cli scard "file:__index__" # Verify source files exist -ls -la /data/watch/ +ls -la /files/watch/ # Check API stats curl http://localhost:3000/api/fuse/stats + +# Check streaming state builder stats +docker logs meta-fuse-dev | grep "State builder" ``` ## Technology Stack @@ -597,14 +672,25 @@ Meta-Fuse is designed to work seamlessly with other MetaMesh services: Ensure meta-sort is running and has processed files: ```bash -# On meta-sort host -curl http://localhost:3000/api/processing/status +# Check meta-sort processing status +curl http://localhost:8180/api/processing/status -# Verify KV has metadata -redis-cli keys "meta-sort:file:*" +# Verify Redis has metadata (via meta-core) +docker exec meta-core-dev redis-cli scard "file:__index__" + +# Check meta:events stream has events +docker exec meta-core-dev redis-cli xlen "meta:events" ``` -Meta-fuse will automatically discover the KV database through the shared lock file. +Meta-fuse will automatically discover the KV database through the shared lock file at `/meta-core/locks/kv-leader.info`. + +## Documentation + +For detailed architecture documentation, see the `docs/` directory: + +- [VFS Rebuild Architecture](docs/vfs-rebuild-architecture.md) - How VFS state is built from Redis Streams +- [Streaming Architecture](docs/streaming-architecture.md) - Event processing pipeline details +- [API Reference](docs/api-reference.md) - Complete REST API documentation ## License diff --git a/docs/api-reference.md b/docs/api-reference.md new file mode 100644 index 0000000..6abd38c --- /dev/null +++ b/docs/api-reference.md @@ -0,0 +1,671 @@ +# API Reference + +## Overview + +meta-fuse exposes a REST API for FUSE operations, VFS management, and service discovery. The API server is built with Fastify and listens on port 3000 by default. + +**Base URL**: `http://localhost:3000` + +--- + +## Health & Status + +### GET /health + +Health check endpoint. + +**Response**: +```json +{ + "status": "ok", + "timestamp": "2024-01-15T10:30:00.000Z", + "service": "meta-fuse" +} +``` + +### GET /api/health + +Alias for `/health`. + +### GET /api/fuse/health + +Alias for `/health`. + +### GET /api/fuse/stats + +Get VFS statistics and state information. + +**Response**: +```json +{ + "fileCount": 5123, + "directoryCount": 847, + "totalSize": 15847392847362, + "lastRefresh": "2024-01-15T10:30:00.000Z", + "categories": { + "Movies": 1523, + "TV": 2847, + "Anime": 753 + }, + "kvConnection": "connected", + "fuseMount": "mounted" +} +``` + +--- + +## FUSE Operations + +### POST /api/fuse/readdir + +List contents of a virtual directory. + +**Request Body**: +```json +{ + "path": "/Movies/Action" +} +``` + +**Response** (200 OK): +```json +{ + "entries": [ + "Die Hard (1988)", + "Mad Max Fury Road (2015)", + "The Dark Knight (2008)" + ] +} +``` + +**Response** (404 Not Found): +```json +{ + "error": "Directory not found" +} +``` + +### POST /api/fuse/getattr + +Get file or directory attributes. + +**Request Body**: +```json +{ + "path": "/Movies/Action/Die Hard (1988)/Die Hard.mkv" +} +``` + +**Response** (200 OK) - File: +```json +{ + "size": 4831838208, + "mode": "file", + "permissions": 644, + "uid": 1000, + "gid": 1000, + "atime": "2024-01-15T10:30:00Z", + "mtime": "2024-01-15T10:30:00Z", + "ctime": "2024-01-15T10:30:00Z", + "sourcePath": "/files/watch/downloads/Die.Hard.1988.1080p.mkv" +} +``` + +**Response** (200 OK) - Directory: +```json +{ + "mode": "directory", + "permissions": 755, + "uid": 1000, + "gid": 1000 +} +``` + +**Response** (404 Not Found): +```json +{ + "error": "Path not found" +} +``` + +### POST /api/fuse/exists + +Check if a path exists in the VFS. + +**Request Body**: +```json +{ + "path": "/Movies/Action/Die Hard (1988)" +} +``` + +**Response**: +```json +{ + "exists": true +} +``` + +### POST /api/fuse/read + +Get information for reading a file. Returns source path and optionally WebDAV URL. + +**Request Body**: +```json +{ + "path": "/Movies/Action/Die Hard (1988)/Die Hard.mkv" +} +``` + +**Response** (200 OK) - Regular file: +```json +{ + "sourcePath": "/files/watch/downloads/Die.Hard.1988.1080p.mkv", + "size": 4831838208 +} +``` + +**Response** (200 OK) - File with WebDAV access: +```json +{ + "sourcePath": "/files/watch/downloads/Die.Hard.1988.1080p.mkv", + "size": 4831838208, + "webdavUrl": "http://meta-core/webdav/watch/downloads/Die.Hard.1988.1080p.mkv" +} +``` + +**Response** (200 OK) - Virtual file with content: +```json +{ + "sourcePath": null, + "size": 256, + "content": "VGhpcyBpcyBhIHZpcnR1YWwgZmlsZQ==", + "contentEncoding": "base64" +} +``` + +**Response** (404 Not Found): +```json +{ + "error": "File not found" +} +``` + +### POST /api/fuse/metadata + +Get full metadata for a virtual path. + +**Request Body**: +```json +{ + "path": "/Movies/Action/Die Hard (1988)/Die Hard.mkv" +} +``` + +**Response** (200 OK): +```json +{ + "hashId": "bafkreiabc123...", + "filePath": "watch/downloads/Die.Hard.1988.1080p.mkv", + "title": "Die Hard", + "year": "1988", + "fileType": "video", + "size": "4831838208", + "extension": "mkv", + "video/codec": "h265", + "video/resolution": "1920x1080", + "audio/codec": "aac", + "audio/channels": "6" +} +``` + +**Response** (404 Not Found): +```json +{ + "error": "Metadata not found" +} +``` + +### GET /api/fuse/files + +List all virtual files in the VFS. + +**Response**: +```json +{ + "files": [ + "/Movies/Action/Die Hard (1988)/Die Hard.mkv", + "/Movies/Action/Mad Max Fury Road (2015)/Mad Max Fury Road.mkv", + "/TV/Breaking Bad/Season 01/S01E01 - Pilot.mkv" + ] +} +``` + +### GET /api/fuse/directories + +List all virtual directories in the VFS. + +**Response**: +```json +{ + "directories": [ + "/Movies", + "/Movies/Action", + "/Movies/Action/Die Hard (1988)", + "/TV", + "/TV/Breaking Bad", + "/TV/Breaking Bad/Season 01" + ] +} +``` + +### POST /api/fuse/refresh + +Trigger a VFS refresh. Re-reads metadata and rebuilds virtual paths. + +**Response**: +```json +{ + "status": "ok" +} +``` + +--- + +## Renaming Rules + +### GET /api/fuse/rules + +Get the current renaming rules configuration. + +**Response**: +```json +{ + "config": { + "rules": [ + { + "name": "Movies", + "enabled": true, + "priority": 100, + "template": "/Movies/{genre}/{title} ({year})/{fileName}", + "conditions": { + "operator": "and", + "conditions": [ + { "field": "fileType", "operator": "equals", "value": "video" }, + { "field": "mediaType", "operator": "equals", "value": "movie" } + ] + } + }, + { + "name": "TV Shows", + "enabled": true, + "priority": 90, + "template": "/TV/{title}/Season {season:pad2}/{title} - S{season:pad2}E{episode:pad2}.{extension}", + "conditions": { + "operator": "and", + "conditions": [ + { "field": "fileType", "operator": "equals", "value": "video" }, + { "field": "mediaType", "operator": "equals", "value": "tv" } + ] + } + } + ], + "defaultRule": { + "name": "Default", + "template": "/Other/{fileName}", + "conditions": { "operator": "and", "conditions": [] } + }, + "lastModified": "2024-01-15T10:30:00Z" + }, + "lastModified": "2024-01-15T10:30:00Z" +} +``` + +### PUT /api/fuse/rules + +Update the renaming rules configuration. Automatically triggers VFS refresh. + +**Request Body**: +```json +{ + "config": { + "rules": [ + { + "name": "Movies", + "enabled": true, + "priority": 100, + "template": "/Movies/{title} ({year})/{fileName}", + "conditions": { + "operator": "and", + "conditions": [ + { "field": "fileType", "operator": "equals", "value": "video" } + ] + } + } + ] + } +} +``` + +**Response** (200 OK): +```json +{ + "success": true, + "refreshed": true +} +``` + +**Response** (400 Bad Request) - Validation errors: +```json +{ + "success": false, + "errors": [ + "Rule \"Movies\": Unknown variable 'unknownVar' in template", + "Rule \"TV Shows\": Invalid condition operator 'matches'" + ] +} +``` + +### POST /api/fuse/rules/preview + +Preview how files would be renamed with the given rules. + +**Request Body**: +```json +{ + "rules": [ + { + "name": "Movies", + "template": "/New Movies/{title} ({year})/{fileName}", + "conditions": { + "operator": "and", + "conditions": [ + { "field": "fileType", "operator": "equals", "value": "video" } + ] + } + } + ], + "limit": 10 +} +``` + +**Response**: +```json +{ + "previews": [ + { + "hashId": "bafkreiabc123...", + "currentPath": "/Movies/Action/Die Hard (1988)/Die Hard.mkv", + "newPath": "/New Movies/Die Hard (1988)/Die Hard.mkv", + "ruleName": "Movies" + }, + { + "hashId": "bafkreixyz789...", + "currentPath": "/TV/Breaking Bad/Season 01/S01E01.mkv", + "newPath": null, + "ruleName": null, + "reason": "No matching rule" + } + ], + "total": 5123, + "sampled": 10 +} +``` + +### POST /api/fuse/rules/validate + +Validate a single renaming rule. + +**Request Body**: +```json +{ + "rule": { + "name": "Test Rule", + "template": "/Test/{title|originalTitle} ({year})/{fileName}", + "conditions": { + "operator": "and", + "conditions": [ + { "field": "fileType", "operator": "equals", "value": "video" } + ] + } + }, + "sampleMetadata": { + "title": "Inception", + "originalTitle": "Inception", + "year": "2010", + "fileName": "Inception.mkv" + } +} +``` + +**Response** (valid rule): +```json +{ + "valid": true, + "errors": [], + "warnings": [], + "sampleOutput": "/Test/Inception (2010)/Inception.mkv" +} +``` + +**Response** (invalid rule): +```json +{ + "valid": false, + "errors": [ + "Unknown variable 'unknownField' in template" + ], + "warnings": [ + "Variable 'director' is rarely available in metadata" + ], + "sampleOutput": null +} +``` + +### GET /api/fuse/rules/variables + +Get list of available template variables. + +**Response**: +```json +{ + "variables": [ + { + "name": "title", + "description": "Media title", + "example": "Inception" + }, + { + "name": "originalTitle", + "description": "Original language title", + "example": "Inception" + }, + { + "name": "year", + "description": "Release year", + "example": "2010" + }, + { + "name": "season", + "description": "Season number (TV shows)", + "example": "1", + "modifiers": ["pad2", "pad3"] + }, + { + "name": "episode", + "description": "Episode number (TV shows)", + "example": "5", + "modifiers": ["pad2", "pad3"] + }, + { + "name": "fileName", + "description": "Original filename", + "example": "Movie.2024.1080p.mkv" + }, + { + "name": "extension", + "description": "File extension", + "example": "mkv" + }, + { + "name": "fileType", + "description": "Type of file", + "example": "video" + }, + { + "name": "genre", + "description": "Primary genre", + "example": "Action" + } + ] +} +``` + +--- + +## Service Discovery + +### GET /api/services + +List all discovered MetaMesh services. + +**Response**: +```json +{ + "services": [ + { + "name": "meta-sort", + "url": "http://meta-sort-dev", + "api": "http://meta-sort-dev/api", + "status": "healthy", + "role": "writer" + }, + { + "name": "meta-fuse", + "url": "http://meta-fuse-dev", + "api": "http://meta-fuse-dev/api", + "status": "healthy", + "role": "reader" + }, + { + "name": "meta-stremio", + "url": "http://meta-stremio-dev", + "api": "http://meta-stremio-dev/api", + "status": "healthy", + "role": "reader" + } + ], + "current": "meta-fuse", + "leader": { + "hostname": "meta-core-dev", + "baseUrl": "http://localhost:8083", + "apiUrl": "http://meta-core:9000", + "redisUrl": "redis://meta-core:6379", + "webdavUrl": "http://localhost:8083/webdav" + } +} +``` + +--- + +## Error Responses + +All endpoints may return these error responses: + +### 400 Bad Request + +Missing or invalid parameters. + +```json +{ + "error": "Missing or invalid \"path\" parameter" +} +``` + +### 404 Not Found + +Resource not found. + +```json +{ + "error": "File not found" +} +``` + +### 500 Internal Server Error + +Server-side error. + +```json +{ + "error": "Failed to process request" +} +``` + +--- + +## Template Syntax + +### Basic Variables + +``` +{variableName} → Replaced with metadata value +{title} → "Inception" +{year} → "2010" +``` + +### Variable with Fallback + +``` +{title|originalTitle} → Uses title, falls back to originalTitle +{genre|"Unknown"} → Uses genre, falls back to literal "Unknown" +``` + +### Variable Modifiers + +``` +{season:pad2} → "01" (zero-padded to 2 digits) +{episode:pad3} → "001" (zero-padded to 3 digits) +{title:lower} → "inception" (lowercase) +{title:upper} → "INCEPTION" (uppercase) +``` + +### Condition Operators + +| Operator | Description | Example | +|----------|-------------|---------| +| `equals` | Exact match | `{ "field": "fileType", "operator": "equals", "value": "video" }` | +| `notEquals` | Not equal | `{ "field": "fileType", "operator": "notEquals", "value": "subtitle" }` | +| `contains` | Contains substring | `{ "field": "title", "operator": "contains", "value": "Star" }` | +| `startsWith` | Starts with | `{ "field": "fileName", "operator": "startsWith", "value": "[" }` | +| `endsWith` | Ends with | `{ "field": "extension", "operator": "endsWith", "value": "mkv" }` | +| `exists` | Field exists | `{ "field": "season", "operator": "exists" }` | +| `notExists` | Field missing | `{ "field": "season", "operator": "notExists" }` | +| `greaterThan` | Numeric > | `{ "field": "year", "operator": "greaterThan", "value": "2000" }` | +| `lessThan` | Numeric < | `{ "field": "season", "operator": "lessThan", "value": "10" }` | + +### Condition Groups + +```json +{ + "operator": "and", + "conditions": [ + { "field": "fileType", "operator": "equals", "value": "video" }, + { + "operator": "or", + "conditions": [ + { "field": "mediaType", "operator": "equals", "value": "movie" }, + { "field": "mediaType", "operator": "equals", "value": "tv" } + ] + } + ] +} +``` + +--- + +## Related Documentation + +- [README](../README.md) - Service overview +- [VFS Rebuild Architecture](vfs-rebuild-architecture.md) - State management +- [Streaming Architecture](streaming-architecture.md) - Event processing diff --git a/docs/streaming-architecture.md b/docs/streaming-architecture.md new file mode 100644 index 0000000..ebcb7e7 --- /dev/null +++ b/docs/streaming-architecture.md @@ -0,0 +1,455 @@ +# Streaming Architecture + +## Overview + +This document describes the streaming event processing system used by meta-fuse to maintain VFS state. The architecture uses Redis Streams for reliable, ordered event delivery with replay capability. + +## Core Components + +### StreamingStateBuilder + +**File**: `src/vfs/StreamingStateBuilder.ts` + +The StreamingStateBuilder is the central component for processing stream events and maintaining file metadata state. + +```typescript +interface StreamingStateBuilderConfig { + /** Redis client for fetching property values */ + redisClient: RedisClient; + /** Renaming rules config for extracting VFS-relevant properties */ + rulesConfig: RenamingConfig; + /** VFS update callback */ + vfsCallback?: VFSUpdateCallback; + /** FILES_VOLUME path prefix */ + filesPath?: string; +} +``` + +**Responsibilities**: +- Process events from `meta:events` stream +- Filter events by VFS-relevant properties +- Fetch property values from Redis +- Maintain in-memory state of all files +- Notify VFS when files are complete or change + +### RulesPropertyExtractor + +**File**: `src/vfs/RulesPropertyExtractor.ts` + +Analyzes renaming rules configuration to determine which metadata properties are needed for VFS path computation. + +```typescript +class RulesPropertyExtractor { + // Extract all VFS-relevant properties from rules config + extractVfsProperties(config: RenamingConfig): Set; + + // Get core properties that are always required + getCoreProperties(): Set; +} +``` + +**Core Properties** (always required): +- `filePath` - Source file location +- `size`, `fileSize` - File size +- `mtime`, `ctime` - Timestamps +- `fileName`, `extension` - Filename components + +**Rule-Derived Properties** (extracted from templates and conditions): +- Template variables: `{title}`, `{season:pad2}`, `{episode:pad2}` +- Condition fields: `fileType`, `year`, `mediaType` + +### VFSUpdateCallback + +**File**: `src/vfs/StreamingStateBuilder.ts:32-37` + +Interface for VFS to receive updates from StreamingStateBuilder: + +```typescript +interface VFSUpdateCallback { + // Called when a property value changes + onPropertyChange(hashId: string, property: string, value: string): void; + + // Called when a property is deleted + onPropertyDelete(hashId: string, property: string): void; + + // Called when a file is completely removed (filePath deleted) + onFileDelete(hashId: string): void; + + // Called when a file becomes complete (has filePath) or updates + onFileComplete(hashId: string, metadata: Record): void; +} +``` + +--- + +## Event Processing Pipeline + +### Stream Message Format + +**File**: `src/kv/RedisClient.ts:30-43` + +```typescript +interface StreamMessage { + id: string; // Stream entry ID (e.g., "1703808000000-0") + type: 'set' | 'del'; // Operation type + key?: string; // Redis key (e.g., "file:abc123/title") + ts?: string; // Timestamp + timestamp: string; // Fallback timestamp +} +``` + +### 5-Step Event Processing + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ Step 1: Receive Stream Message │ +│ ────────────────────────────────────────────────────────────── │ +│ {id: "1703808000-0", type: "set", key: "file:abc123/title"} │ +└───────────────────────────────┬─────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ Step 2: Parse Key │ +│ ────────────────────────────────────────────────────────────── │ +│ parseMetaEventKey("file:abc123/title") │ +│ → { hashId: "abc123", property: "title" } │ +└───────────────────────────────┬─────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ Step 3: Check VFS Relevance │ +│ ────────────────────────────────────────────────────────────── │ +│ isVfsRelevantProperty("title", vfsRelevantProps) │ +│ → true (if "title" is in renaming rules) │ +│ → false → skip, increment propertiesSkipped │ +└───────────────────────────────┬─────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ Step 4: Fetch Property Value │ +│ ────────────────────────────────────────────────────────────── │ +│ await redisClient.getProperty("abc123", "title") │ +│ → "Inception" │ +└───────────────────────────────┬─────────────────────────────────┘ + │ + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ Step 5: Update State & Notify VFS │ +│ ────────────────────────────────────────────────────────────── │ +│ updateProperty("abc123", "title", "Inception") │ +│ vfsCallback.onPropertyChange("abc123", "title", "Inception") │ +│ │ +│ If file just became complete (now has filePath): │ +│ vfsCallback.onFileComplete("abc123", metadata) │ +└─────────────────────────────────────────────────────────────────┘ +``` + +### Code Implementation + +```typescript +// StreamingStateBuilder.ts:163-206 +private async handlePropertySet(hashId: string, property: string): Promise { + // Step 3: Check if this property is VFS-relevant + if (!isVfsRelevantProperty(property, this.vfsRelevantProps)) { + this.stats.propertiesSkipped++; + return; + } + + // Step 4: Fetch the property value from Redis + const value = await this.redisClient.getProperty(hashId, property); + if (value === null) { + return; // Property was set then deleted before we could fetch + } + + this.stats.propertiesFetched++; + + // Check if file was complete BEFORE updating state + const wasComplete = this.isFileComplete(hashId); + + // Step 5: Update internal state + this.updateProperty(hashId, property, value); + + // Notify VFS of changes + if (this.vfsCallback) { + this.vfsCallback.onPropertyChange(hashId, property, value); + + // If file just became complete (has filePath now), notify VFS + if (!wasComplete && this.isFileComplete(hashId)) { + this.stats.filesComplete++; + const metadata = this.getFileMetadata(hashId); + if (metadata) { + this.vfsCallback.onFileComplete(hashId, metadata); + } + } + } +} +``` + +--- + +## Bootstrap vs Live Phases + +### Bootstrap Phase + +**When**: Application startup + +**Goal**: Rebuild complete VFS state from event history + +```typescript +// index.ts:158-174 +// Bootstrap: Replay meta:events stream from position 0 +const lastId = await redisClient.replayStream( + EVENTS_STREAM, + '0', // Start from beginning + async (message: StreamMessage) => { + await stateBuilder.processEvent(message); + }, + 100 // Batch size +); +``` + +**Characteristics**: +- Reads from stream position `0` (all history) +- Non-blocking batch reads (no XREADGROUP) +- Processes all events sequentially +- Returns last processed event ID + +### Live Phase + +**When**: After bootstrap completes + +**Goal**: Process new events in real-time + +```typescript +// index.ts:177-187 +redisClient.startSimpleStreamConsumer( + EVENTS_STREAM, + lastId, // Resume from bootstrap position + async (message: StreamMessage) => { + await stateBuilder.processEvent(message); + }, + 5000 // 5 second block timeout +); +``` + +**Characteristics**: +- Resumes from last bootstrap position +- Uses blocking reads (XREAD BLOCK) +- Processes events as they arrive +- Runs until shutdown + +--- + +## Memory Efficiency + +### State Structure + +```typescript +type FileState = Map; // property → value +type FilesState = Map; // hashId → properties +``` + +### Memory Per File + +| Component | Size | +|-----------|------| +| hashId (64 char) | ~128 bytes | +| Map overhead | ~56 bytes | +| 5-10 properties (avg) | ~200-400 bytes | +| **Total per file** | **~400-600 bytes** | + +### Comparison with Full Load + +| Approach | Memory per File | 10K Files | +|----------|-----------------|-----------| +| Full HGETALL (all props) | ~2-5 KB | 20-50 MB | +| Streaming (VFS props only) | ~500 bytes | 5 MB | +| **Savings** | **75-90%** | **15-45 MB** | + +### Why It Works + +1. **Property Filtering**: Only fetch properties used in renaming rules +2. **Lazy Loading**: Properties fetched on-demand from events +3. **No Duplication**: Values stored once in state map +4. **Garbage Collection**: File state removed when file deleted + +--- + +## Property Filtering Details + +### Extracting VFS-Relevant Properties + +```typescript +// RulesPropertyExtractor.ts:132-155 +extractVfsProperties(config: RenamingConfig): Set { + const properties = new Set(CORE_PROPERTIES); + + // Extract from all enabled rules + for (const rule of config.rules) { + if (!rule.enabled) continue; + + // From template: {title|originalTitle}, {season:pad2} + const templateVars = templateEngine.extractVariables(rule.template); + properties.add(...templateVars); + + // From conditions: { field: 'fileType', value: 'video' } + const conditionFields = extractConditionFields(rule.conditions); + properties.add(...conditionFields); + } + + return properties; +} +``` + +### Property Path Matching + +Handles both exact and nested property matching: + +```typescript +// RulesPropertyExtractor.ts:85-111 +function isVfsRelevantProperty(property: string, vfsProperties: Set): boolean { + const normalized = normalizePropertyPath(property); + + // Direct match + if (vfsProperties.has(normalized)) { + return true; + } + + // Nested property match + // e.g., if we track 'titles', then 'titles/eng' matches + for (const vfsProp of vfsProperties) { + if (normalized.startsWith(vfsProp + '.')) { + return true; + } + if (vfsProp.startsWith(normalized + '.')) { + return true; + } + } + + return false; +} +``` + +--- + +## Statistics and Monitoring + +### StreamingStateBuilder Stats + +```typescript +// StreamingStateBuilder.ts:68-74 +private stats = { + eventsProcessed: 0, // Total events received + propertiesFetched: 0, // Properties fetched from Redis + propertiesSkipped: 0, // Non-VFS-relevant properties skipped + filesComplete: 0, // Files that became complete + lastEventId: '0', // Last processed event ID +}; +``` + +### Accessing Stats + +```typescript +const stats = stateBuilder.getStats(); +// Returns: +{ + eventsProcessed: 15234, + propertiesFetched: 8934, + propertiesSkipped: 6300, + filesComplete: 5123, + lastEventId: "1703808000000-0", + fileCount: 5123, // Complete files (have filePath) + stateSize: 5234 // Total files in state +} +``` + +### Interpreting Stats + +| Metric | Meaning | +|--------|---------| +| `eventsProcessed` | Total stream events consumed | +| `propertiesFetched` | VFS-relevant property GET calls | +| `propertiesSkipped` | Non-relevant events filtered | +| `filesComplete` | Files added to VFS | +| `fileCount` | Current files in VFS | +| Fetch ratio | `propertiesFetched / eventsProcessed` (lower is better) | + +--- + +## Error Handling + +### Event Processing Errors + +```typescript +// StreamingStateBuilder.ts:123-136 +async processEvent(message: StreamMessage): Promise { + this.stats.eventsProcessed++; + this.stats.lastEventId = message.id; + + if ((message.type === 'set' || message.type === 'del') && message.key) { + await this.processMetaEvent(message); + return; + } + + // Ignore legacy event types - not supported in flat key architecture + logger.debug(`Ignoring legacy event type: ${message.type}`); +} +``` + +### Property Fetch Failures + +```typescript +// StreamingStateBuilder.ts:171-177 +const value = await this.redisClient.getProperty(hashId, property); +if (value === null) { + // Property was set then deleted before we could fetch it + // Or it's a different data type - skip silently + return; +} +``` + +### Stream Consumer Recovery + +```typescript +// index.ts:185-187 +.catch(error => { + logger.error('Stream consumer error:', error); +}); +``` + +On error, the consumer loop continues retrying with brief pauses. + +--- + +## Configuration + +### StreamingStateBuilder Configuration + +```typescript +const stateBuilder = new StreamingStateBuilder({ + redisClient, + rulesConfig: vfs.getRulesConfig(), + vfsCallback: vfs, + filesPath: '/files', +}); +``` + +### Dynamic Rules Update + +When renaming rules change: + +```typescript +// StreamingStateBuilder.ts:99-102 +updateRulesConfig(rulesConfig: RenamingConfig): void { + this.vfsRelevantProps = this.propertyExtractor.extractVfsProperties(rulesConfig); + logger.info(`Updated VFS-relevant properties: ${this.vfsRelevantProps.size}`); +} +``` + +--- + +## Related Documentation + +- [VFS Rebuild Architecture](vfs-rebuild-architecture.md) - High-level architecture overview +- [API Reference](api-reference.md) - REST API documentation +- [Redis Key Format](../../../meta-core/docs/redis-format.md) - Flat key storage format diff --git a/docs/vfs-rebuild-architecture.md b/docs/vfs-rebuild-architecture.md index 54c3da6..0194f15 100644 --- a/docs/vfs-rebuild-architecture.md +++ b/docs/vfs-rebuild-architecture.md @@ -2,7 +2,7 @@ ## Overview -This document describes the architecture and design philosophy for the Virtual Filesystem (VFS) rebuild process in MetaMesh. The system uses a **two-phase strategy** to provide fast startup times while maintaining data accuracy through continuous validation. +This document describes the architecture and design philosophy for the Virtual Filesystem (VFS) rebuild process in MetaMesh. The system uses a **streaming-based strategy** with Redis Streams to provide fast startup times while maintaining data accuracy through event-driven updates. ## Design Philosophy @@ -10,276 +10,242 @@ The VFS rebuild architecture is built on three core principles: | Component | Role | Responsibility | |-----------|------|----------------| -| **etcd** | Distributed Metadata Archive | Permanent record of all files (even if deleted from disk) | -| **In-Memory Database** | Current VFS State | Only files present on disk RIGHT NOW | -| **Discovery** | Source of Truth | Validates file existence and cleans stale entries | +| **Redis** | Metadata Storage | Flat key storage of all file properties | +| **Redis Streams** | Event Log | Ordered record of all metadata changes (`meta:events`) | +| **StreamingStateBuilder** | State Manager | Processes events, maintains VFS-relevant state in memory | ### Key Insight -> **etcd stores the historical truth** (what files have existed) -> **In-memory DB stores the current truth** (what files exist NOW) -> **Discovery enforces reality** (makes memory match disk state) +> **Redis stores the current truth** (what file properties exist) +> **Redis Streams store the change log** (what changed and when) +> **StreamingStateBuilder filters and projects** (extracts only VFS-relevant properties) This separation enables: -- **Fast cold starts** (1-2 seconds for 10,000 files) -- **Distributed metadata sharing** via meta-orbit (etcd never deletes) -- **Accurate VFS state** (discovery continuously validates) -- **Graceful handling of offline storage** (network drives, removable media) +- **Fast cold starts** (1-2 seconds for 10,000 files via stream replay) +- **Memory efficiency** (~500 bytes/file, only VFS-relevant properties) +- **Real-time updates** (live stream consumption after bootstrap) +- **Graceful handling of offline storage** (state rebuilds on reconnection) --- -## Two-Phase Strategy +## Streaming Architecture -### Phase 1: Fast Bootstrap (Startup) +### Phase 1: Streaming Bootstrap (Startup) -**Goal**: Make VFS available as quickly as possible +**Goal**: Rebuild VFS state as quickly as possible from event stream -**Implementation**: `WatchedFileProcessor.rebuildVFSFromEtcd()` +**Implementation**: See `index.ts:158-174` ```typescript -async rebuildVFSFromEtcd(): Promise { - // 1. Fetch all hash IDs from etcd - const hashIds = await etcdClient.getAllHashIds(); +// Bootstrap: Replay meta:events stream from position 0 +const lastId = await redisClient.replayStream( + EVENTS_STREAM, // 'meta:events' + '0', // Start from beginning + async (message: StreamMessage) => { + await stateBuilder.processEvent(message); + }, + 100 // Batch size +); - // 2. Batch fetch ALL metadata in parallel (no disk I/O) - const metadataList = await Promise.all( - hashIds.map(hashId => etcdClient.getMetadataFlat(hashId)) - ); - - // 3. Load into in-memory DB without validation - for (const metadata of metadataList) { - metadata._lastVerified = 0; // Mark as unverified - this.fileProcessor.getDatabase().set(metadata.filePath, metadata); - } - - // 4. Build VFS immediately (1-2 seconds) - await this.finalize(); -} +const stats = vfs.getStats(); +logger.info(`Streaming bootstrap complete: ${stats.fileCount} files`); ``` +**How It Works**: +1. Read stream entries in batches using `XREAD` +2. For each `set` event, check if property is VFS-relevant +3. If relevant, fetch property value with `GET file:{hashId}/{property}` +4. Update in-memory state and notify VFS when file has `filePath` + **Performance**: -- 10,000 files: ~1-2 seconds (previously 60+ seconds) -- Zero disk I/O during startup -- VFS immediately usable (may contain stale entries) +- 10,000 files: ~1-2 seconds +- Zero SCAN/HGETALL operations +- Only fetches properties needed for VFS path computation **Characteristics**: -- ✅ Fast startup -- ✅ No blocking on disk validation -- ✅ All files appear immediately -- ⚠️ May include deleted/offline files temporarily +- Stream replay from position 0 +- Incremental state building +- VFS populated as files become "complete" (have `filePath`) --- -### Phase 2: Continuous Validation (Discovery) +### Phase 2: Live Stream Consumption -**Goal**: Make in-memory DB match current disk reality +**Goal**: Keep VFS synchronized with metadata changes in real-time -**Implementation**: Discovery scans folders progressively and validates file existence +**Implementation**: See `index.ts:177-187` ```typescript -// Progressive scan: Process folders one at a time -for (const folder of folderList) { - const discoveredFilesInFolder = new Set(); - - // Track discovered files for this folder - async function* trackDiscoveredFiles(stream: AsyncGenerator) { - for await (const filePath of stream) { - discoveredFilesInFolder.add(filePath); // Track what exists - yield filePath; - } - } - - const discoveryStream = folderWatcher.discoverFiles([folder]); - await pipeline.start(trackDiscoveredFiles(discoveryStream)); - - // Progressive cleanup: Remove stale entries after each folder - await fileProcessor.cleanupStaleEntries([folder], discoveredFilesInFolder); -} -``` - -**Cleanup Logic**: `WatchedFileProcessor.cleanupStaleEntries()` - -```typescript -async cleanupStaleEntries(scannedFolders: string[], discoveredFiles: Set) { - for (const folder of scannedFolders) { - // Find all in-memory files for this folder - const filesToCheck = db.getAllFilesStartingWith(folder); - - // Remove files not seen during scan - for (const filePath of filesToCheck) { - if (!discoveredFiles.has(filePath)) { - // File no longer exists → remove from memory - db.delete(filePath); - vfs.removeFileByRealPath(filePath); - stateManager.removeFile(filePath); - } - } - } - - // Rebuild VFS if anything changed - await this.finalize(); -} -``` - -**Validation Rules** (applied progressively per folder): - -| Scenario | Action | VFS Update | -|----------|--------|------------| -| File found during scan | Update `_lastVerified` timestamp, keep in memory | No change | -| File not found in scanned folder | Remove from in-memory DB, keep in etcd | VFS rebuilt incrementally | -| Empty folder scanned | Remove ALL child paths from memory immediately | VFS rebuilt (files disappear) | -| Folder missing/inaccessible | Remove ALL child paths from memory immediately | VFS rebuilt (files disappear) | - -**Key Behavior**: When a NAS/network drive is disconnected (folder becomes empty/inaccessible), all files from that folder are **progressively removed from the VFS as discovery proceeds** through each folder. - -**Characteristics**: -- ✅ Discovery is authoritative -- ✅ Progressive cleanup (files disappear as folders are scanned) -- ✅ Handles deleted files automatically -- ✅ Handles offline network drives (files removed immediately when folder is scanned) -- ✅ No manual TTL/expiry needed -- ✅ Real-time VFS updates during discovery - ---- - -## Architecture Comparison - -### Old Approach (Disk Validation at Startup) - -``` -┌─────────────────────────────────────────────┐ -│ Startup: rebuildVFSFromEtcd() │ -│ │ -│ For each file in etcd (sequential): │ -│ 1. Fetch metadata from etcd (1-5ms) │ -│ 2. Check if file exists (fs.access) │ -│ - Local disk: 1-10ms │ -│ - Network drive: 10-100ms │ -│ 3. Add to memory if exists │ -│ │ -│ Result: 60-120 seconds for 10k files │ -└─────────────────────────────────────────────┘ -``` - -**Problems**: -- ❌ Slow startup (blocks on disk I/O) -- ❌ Sequential processing (no parallelization) -- ❌ Network latency multiplies wait time -- ❌ VFS unavailable during validation - -### New Approach (Fast Load + Progressive Discovery) - -``` -┌─────────────────────────────────────────────┐ -│ Startup: rebuildVFSFromEtcd() │ -│ │ -│ 1. Batch fetch ALL metadata (parallel) │ -│ → 10k files in ~500ms │ -│ │ -│ 2. Load into memory (no disk checks) │ -│ → 10k files in ~200ms │ -│ │ -│ 3. Build VFS │ -│ → Ready in 1-2 seconds total │ -│ │ -│ Result: VFS immediately usable │ -└─────────────────────────────────────────────┘ - -┌─────────────────────────────────────────────┐ -│ Background: Progressive Discovery Scan │ -│ │ -│ For each folder in watch list: │ -│ 1. Walk folder and track discovered files│ -│ 2. Update _lastVerified for found files │ -│ 3. Remove stale entries for this folder │ -│ 4. Rebuild VFS (files disappear) │ -│ 5. Move to next folder │ -│ │ -│ Result: Files progressively disappear as │ -│ folders are scanned (real-time) │ -└─────────────────────────────────────────────┘ -``` - -**Benefits**: -- ✅ 10-50x faster startup -- ✅ Parallel etcd fetching -- ✅ No disk I/O blocking -- ✅ VFS immediately available -- ✅ Progressive cleanup (real-time feedback) -- ✅ Files disappear as folders are scanned - ---- - -## Implementation Details - -### File Metadata Structure - -```typescript -interface FileMetadata { - filePath: string; - cid_midhash256: string; // Permanent file ID - title: string; - // ... other metadata fields - - // Validation tracking (optional) - _lastVerified?: number; // Timestamp of last disk validation -} -``` - -The `_lastVerified` field: -- Set to `0` during fast load (unverified) -- Updated to `Date.now()` when discovery sees the file -- Used for debugging and monitoring (not enforced) - -### State Management - -**In-Memory Database** (`FileProcessor.database`): -- `Map` -- Cleared on restart -- Reflects only currently accessible files - -**etcd Storage**: -- `/file/midhash256:{hash}/property` - Permanent metadata -- Never deleted (even if file removed from disk) -- Shared across meta-orbit network - -**Virtual Filesystem** (`VirtualFileSystem`): -- Generated from in-memory database -- Rebuilt after cleanup operations -- Accessible via FUSE/WebDAV - -### Discovery Integration - -Discovery runs in two modes: - -**1. Manual Scan** (startup or triggered via API) - Progressive: -```typescript -// Process folders one at a time -for (const folder of folderList) { - const discoveredFiles = new Set(); - const discoveryStream = folderWatcher.discoverFiles([folder]); - - // Track and process files for this folder - await pipeline.start(trackFiles(discoveryStream, discoveredFiles)); - - // Progressive cleanup: Remove stale entries immediately - await fileProcessor.cleanupStaleEntries([folder], discoveredFiles); - // → VFS rebuilt, files from disconnected NAS disappear -} -``` - -**2. Watch Mode** (continuous monitoring): -```typescript -folderWatcher.watch(folderList, { - onAdd: (filePath) => { - // File added → process and add to memory - pipeline.processFile(filePath); +// Start live stream consumer from where replay left off +redisClient.startSimpleStreamConsumer( + EVENTS_STREAM, + lastId, // Resume from bootstrap position + async (message: StreamMessage) => { + await stateBuilder.processEvent(message); }, - onUnlink: (filePath) => { - // File removed → remove from memory immediately - fileProcessor.deleteFile(filePath); + 5000 // 5 second block timeout +); +``` + +**Event Processing** (see `StreamingStateBuilder.ts:123-136`): + +```typescript +async processEvent(message: StreamMessage): Promise { + // Handle new meta:events format (set/del with key field) + if ((message.type === 'set' || message.type === 'del') && message.key) { + await this.processMetaEvent(message); + return; } -}); + // Legacy events ignored +} +``` + +**Property Filtering** (see `RulesPropertyExtractor.ts`): + +```typescript +// Only fetch VFS-relevant properties +if (!isVfsRelevantProperty(property, this.vfsRelevantProps)) { + this.stats.propertiesSkipped++; + return; +} + +// Fetch and update state +const value = await this.redisClient.getProperty(hashId, property); +this.updateProperty(hashId, property, value); +``` + +--- + +## Architecture Diagrams + +### Startup Sequence + +``` +┌─────────────────────────────────────────────────────────────────────────┐ +│ Streaming Bootstrap │ +├─────────────────────────────────────────────────────────────────────────┤ +│ │ +│ 1. Initialize KVManager (leader discovery) │ +│ ├── Read /meta-core/locks/kv-leader.info │ +│ ├── Call /urls API for Redis URL │ +│ └── Connect to Redis │ +│ │ +│ 2. Create StreamingStateBuilder │ +│ ├── Extract VFS-relevant properties from rules │ +│ └── Wire up VFS callback │ +│ │ +│ 3. Replay meta:events stream from position 0 │ +│ ┌────────────────────────────────────────────────────────────┐ │ +│ │ For each event in stream: │ │ +│ │ ├── Parse key: file:{hashId}/{property} │ │ +│ │ ├── Check if property is VFS-relevant │ │ +│ │ ├── If yes: GET property value │ │ +│ │ ├── Update in-memory state │ │ +│ │ └── If file has filePath: notify VFS (file appears) │ │ +│ └────────────────────────────────────────────────────────────┘ │ +│ │ +│ 4. Start live consumer from last processed ID │ +│ └── Continue processing new events in real-time │ +│ │ +│ 5. Start API server │ +│ └── VFS ready for FUSE/WebDAV operations │ +│ │ +└─────────────────────────────────────────────────────────────────────────┘ +``` + +### Event Processing Pipeline + +``` +┌─────────────────────────────────────────────────────────────────────────┐ +│ Event Processing Pipeline │ +├─────────────────────────────────────────────────────────────────────────┤ +│ │ +│ Redis Stream (meta:events) │ +│ ┌──────────────────────────────────────────────────────────────┐ │ +│ │ {id: "1234-0", type: "set", key: "file:abc123/title"} │ │ +│ │ {id: "1234-1", type: "set", key: "file:abc123/filePath"} │ │ +│ │ {id: "1234-2", type: "del", key: "file:xyz789/filePath"} │ │ +│ └───────────────────────────────┬──────────────────────────────┘ │ +│ │ │ +│ ▼ │ +│ ┌─────────────────────────────────────────────────────────────┐ │ +│ │ StreamingStateBuilder.processEvent() │ │ +│ │ 1. Parse key → {hashId: "abc123", property: "title"} │ │ +│ │ 2. Check VFS relevance (from RulesPropertyExtractor) │ │ +│ │ 3. If relevant: GET file:abc123/title → "Inception" │ │ +│ │ 4. Update state: state.get("abc123").set("title", ...) │ │ +│ └─────────────────────────────────────────────────────────────┘ │ +│ │ │ +│ ▼ │ +│ ┌─────────────────────────────────────────────────────────────┐ │ +│ │ VFSUpdateCallback │ │ +│ │ - onPropertyChange(hashId, property, value) │ │ +│ │ - onFileComplete(hashId, metadata) → file appears in VFS │ │ +│ │ - onFileDelete(hashId) → file removed from VFS │ │ +│ └─────────────────────────────────────────────────────────────┘ │ +│ │ +└─────────────────────────────────────────────────────────────────────────┘ +``` + +--- + +## State Management + +### StreamingStateBuilder State + +```typescript +// FilesState: Map of hashId to property map +type FileState = Map; // property → value +type FilesState = Map; // hashId → properties + +// Example state +{ + "abc123": { + "filePath": "watch/Movies/Inception.mkv", + "title": "Inception", + "year": "2010", + "size": "4831838208" + }, + "xyz789": { + "filePath": "watch/TV/Breaking Bad/S01E01.mkv", + "title": "Breaking Bad", + "season": "1", + "episode": "1" + } +} +``` + +### VFS-Relevant Properties + +Properties are extracted from renaming rules (see `RulesPropertyExtractor.ts`): + +**Core Properties** (always required): +- `filePath` - Source file location (required for file to appear) +- `size`, `fileSize` - File size +- `mtime`, `ctime` - Timestamps +- `fileName`, `extension` - Filename components + +**Rule-Derived Properties**: +- From templates: `{title}`, `{season:pad2}`, `{episode:pad2}` +- From conditions: `{ field: 'fileType', value: 'video' }` + +### Redis Data Structure + +``` +# Flat key format +file:{hashId}/{property} → value + +# Examples +file:abc123/filePath → "watch/Movies/Inception.mkv" +file:abc123/title → "Inception" +file:abc123/titles/eng → "Inception" (nested property) +file:abc123/year → "2010" + +# Index +file:__index__ → SET["abc123", "xyz789", ...] + +# Event stream +meta:events → Stream of {type, key, ts} events ``` --- @@ -288,74 +254,92 @@ folderWatcher.watch(folderList, { ### Startup Performance -| Dataset Size | Old Approach | New Approach | Speedup | -|--------------|-------------|--------------|---------| -| 1,000 files | 5-10s | 0.5-1s | 10x | -| 10,000 files | 60-120s | 1-2s | 60x | -| 100,000 files | 10-20m | 10-15s | 60-80x | - -### Storage Type Impact - -| Storage Type | Old Approach | New Approach | Notes | -|--------------|-------------|--------------|-------| -| Local SSD | 5-10s | 1-2s | Validation was fast, but sequential | -| Local HDD | 20-40s | 1-2s | Seek time penalty eliminated | -| Network (NFS/SMB) | 60-300s | 1-2s | Latency eliminated from startup | -| Offline/Slow Network | Timeout/Fail | 1-2s | Gracefully handles offline storage | +| Dataset Size | Streaming Bootstrap | Notes | +|--------------|---------------------|-------| +| 1,000 files | 0.5-1s | Stream replay + selective GET | +| 10,000 files | 1-2s | Batch processing (100 events) | +| 100,000 files | 10-15s | Memory-efficient incremental build | ### Memory Usage -- **In-Memory DB**: ~500 bytes/file (metadata only) +- **Per-file overhead**: ~500 bytes (only VFS-relevant properties) - **10,000 files**: ~5 MB - **100,000 files**: ~50 MB -- **Negligible overhead** compared to old approach +- **Comparison**: Previous HGETALL approach loaded all properties (~2-5KB/file) + +### Why Streaming is Fast + +| Old Approach | New Approach | +|--------------|--------------| +| `SCAN file:*` to find all keys | `XREAD meta:events` from position 0 | +| `HGETALL file:{id}` for each file | `GET file:{id}/{prop}` only for relevant props | +| All properties loaded | Only VFS-relevant properties | +| O(n × m) where m = all properties | O(n × k) where k = relevant properties | --- ## Handling Edge Cases +### New Files + +1. meta-sort processes file, writes properties to Redis +2. meta-sort publishes `set` events to `meta:events` stream +3. meta-fuse consumes events, fetches relevant properties +4. When `filePath` is set, file appears in VFS + ### Deleted Files -**Scenario**: User deletes a file after startup but before discovery scans +1. meta-sort publishes `del file:{hashId}/filePath` event +2. meta-fuse processes event, removes file from state +3. VFS updated, file disappears -**Old Behavior**: File appears in VFS until next full rebuild +### Redis Reconnection -**New Behavior**: -1. Fast load: File appears in VFS (loaded from etcd) -2. Discovery scan: File not found → removed from memory -3. VFS rebuilt without the file -4. etcd still has metadata (for meta-orbit sharing) +1. Connection lost, stream consumer stops +2. KVManager detects disconnect, attempts reconnect +3. On reconnect, can replay stream from last position +4. State incrementally updated with missed events -### Offline Network Drives +### VFS Refresh -**Scenario**: NFS mount becomes unavailable +Triggered via `POST /api/fuse/refresh`: -**Old Behavior**: Startup hangs or fails with timeout +```typescript +// Fallback for when stream is unavailable +async refresh(): Promise { + // Use flat key scanning fallback + const files = await redisClient.scanFlatKeysForFiles( + stateBuilder.getVfsRelevantProperties() + ); + // Rebuild VFS from scanned data +} +``` -**New Behavior**: -1. Fast load: All files appear in VFS (1-2s) -2. Discovery scan: Folder inaccessible → removes all child paths -3. When drive comes back online: Next discovery re-adds files +--- -### Moved Files +## VFSUpdateCallback Interface -**Scenario**: User moves file outside watch folders +The VFS implements this callback interface to receive updates from StreamingStateBuilder: -**Old Behavior**: File appears in VFS until manual cleanup +```typescript +interface VFSUpdateCallback { + // Property changed for a file + onPropertyChange(hashId: string, property: string, value: string): void; -**New Behavior**: -1. Discovery scans old location → file not found → removed from memory -2. If moved within watch folders: Discovered as new file -3. etcd has both old and new paths (deduplication via midhash256) + // Property deleted from a file + onPropertyDelete(hashId: string, property: string): void; -### Partial Scans + // File completely removed (filePath deleted) + onFileDelete(hashId: string): void; -**Scenario**: Discovery interrupted mid-scan (crash, kill signal) + // File has all required properties, ready to appear in VFS + onFileComplete(hashId: string, metadata: Record): void; +} +``` -**Safety**: Cleanup only runs after scan completes successfully -- Incomplete scan → no cleanup -- Memory state unchanged -- Next scan will complete and clean up +The `onFileComplete` callback is called when: +1. A file first receives a `filePath` property +2. Subsequent property changes after file is already complete --- @@ -364,164 +348,64 @@ folderWatcher.watch(folderList, { ### Startup Logs ``` -[VFS Rebuild] Starting fast-load from etcd... -[VFS Rebuild] Found 10000 files in etcd -[VFS Rebuild] Fetching metadata in parallel... -[VFS Rebuild] Loaded 9856/10000 files into memory (unverified) -[VFS Rebuild] VFS rebuilt in 1523ms -[VFS Rebuild] Discovery will validate file existence in background +[meta-fuse] Starting meta-fuse... +[KVManager] Waiting for leader... +[LeaderClient] Leader found: meta-core-dev at redis://meta-core:6379 +[meta-fuse] KV Manager ready, initializing VFS with streaming bootstrap... +[StreamingStateBuilder] Initialized with 12 VFS-relevant properties +[meta-fuse] Replaying meta:events stream from position 0... +[RedisClient] Stream replay complete: 15234 entries in 1823ms +[meta-fuse] Streaming bootstrap complete: 5123 files in 1893ms +[meta-fuse] Starting live stream consumer from 1703808000000-0... +[meta-fuse] meta-fuse is ready! +[meta-fuse] VFS: 5123 files, 847 directories +[meta-fuse] State builder: 15234 events, 8934 properties fetched, 6300 skipped ``` -### Discovery Logs (Progressive Mode) - -``` -[Discovery] Progressive scan mode: 3 folders -[Discovery] Scanning folder: /data/watch/local -[Discovery] Folder scan complete: /data/watch/local. Found 128 files. Running cleanup... -[Cleanup] Starting cleanup for 1 scanned folders... -[Cleanup] Removed 0 stale entries in 2ms -[Cleanup] Cleanup complete for folder: /data/watch/local - -[Discovery] Scanning folder: /data/watch/smb-nas -[Discovery] Folder scan complete: /data/watch/smb-nas. Found 0 files. Running cleanup... -[Cleanup] Starting cleanup for 1 scanned folders... -[Cleanup] Removed stale entry: /data/watch/smb-nas/Movies/deleted_movie.mkv -[Cleanup] Removed stale entry: /data/watch/smb-nas/Movies/another_movie.mkv -[Cleanup] Removed 13621 stale entries in 87ms -[Cleanup] Rebuilding VFS after cleanup... -Virtual filesystem update start -Duplicate find took 1ms - found 0 hash groups, 21 title groups -Generating virtual structure for 128 files -Virtual filesystem update took 167ms -[Cleanup] Cleanup complete for folder: /data/watch/smb-nas - -[Discovery] Scanning folder: /data/watch/downloads -[Discovery] Folder scan complete: /data/watch/downloads. Found 45 files. Running cleanup... -[Cleanup] Starting cleanup for 1 scanned folders... -[Cleanup] Removed 0 stale entries in 1ms -[Cleanup] Cleanup complete for folder: /data/watch/downloads - -[Discovery] All folders scanned and cleaned up -``` - -**Progressive Behavior**: Notice how the NAS folder (`/data/watch/smb-nas`) with 0 discovered files immediately triggers removal of 13,621 stale entries and VFS rebuild. Users see files disappear in real-time as each folder completes scanning. - -### API Endpoints - -Monitor VFS state via Unified API: +### Stats API ```bash -# VFS statistics curl http://localhost:3000/api/fuse/stats - -# Processing status -curl http://localhost:3000/api/processing/status - -# Trigger manual scan + cleanup -curl -X POST http://localhost:3000/api/scan/trigger ``` ---- - -## Best Practices - -### For Operators - -1. **Initial Startup**: First boot may show all historical files (even deleted ones) - - Wait for first discovery scan to clean up - - Or run manual scan: `POST /api/scan/trigger` - -2. **Network Drives**: If drives are slow or offline at startup - - VFS still loads in 1-2 seconds - - Files appear/disappear as drives come online/offline - - No manual intervention needed - -3. **Monitoring**: Watch for cleanup logs to understand disk changes - ``` - [Cleanup] Removed 622 stale entries - ``` - Large numbers indicate significant file deletions - -### For Developers - -1. **Adding New Metadata Sources**: Always update both: - - etcd (permanent record) - - In-memory DB (current state) - -2. **Testing Cleanup**: Simulate file deletions - ```bash - # Delete files from watch folder - rm -rf /data/watch/Movies/test_movie.mkv - - # Trigger scan - curl -X POST http://localhost:3000/api/scan/trigger - - # Verify cleanup in logs - docker logs meta-mesh-dev-container | grep Cleanup - ``` - -3. **Performance Testing**: Benchmark with large datasets - ```bash - # Time startup - docker restart meta-mesh-dev-container - docker logs -f meta-mesh-dev-container | grep "VFS rebuilt" - - # Time discovery + cleanup - time curl -X POST http://localhost:3000/api/scan/trigger - ``` - ---- - -## Future Optimizations - -### Potential Improvements - -1. **Incremental Cleanup**: Instead of scanning all folders, track modified folders - - Use filesystem events (inotify) to detect folder changes - - Only clean up modified folders - -2. **Background Validation Queue**: Validate files lazily over time - - Prioritize recently accessed files - - Validate cold files during idle periods - -3. **Smart Caching**: Cache validation results - - LRU cache with TTL (5-10 minutes) - - Reduces redundant disk checks for frequently accessed files - -4. **Partial VFS Rebuild**: Only rebuild changed directories - - Track which folders had files removed - - Rebuild only affected VFS subtrees - -### Not Recommended - -❌ **TTL-based expiry**: Requires guessing timeout values, fails with slow scans - -❌ **Storing validation state in etcd**: Wrong tool for transient state, creates write amplification - -❌ **Immediate removal on unlink events**: Race conditions with discovery, inconsistent state +```json +{ + "fileCount": 5123, + "directoryCount": 847, + "totalSize": 15847392847362, + "lastRefresh": "2024-01-15T10:30:00Z", + "stateBuilder": { + "eventsProcessed": 15234, + "propertiesFetched": 8934, + "propertiesSkipped": 6300, + "filesComplete": 5123, + "lastEventId": "1703808000000-0" + } +} +``` --- ## Related Documentation -- [Streaming Pipeline Architecture](streaming-pipeline-architecture.md) - File processing flow -- [etcd Architecture](../../packages/meta-mesh/doc/architecture/etcd-architecture.md) - Storage design -- [FUSE API](../../packages/meta-mesh/doc/architecture/fuse-api.md) - Virtual filesystem API +- [Streaming Architecture](streaming-architecture.md) - Detailed streaming event processing +- [API Reference](api-reference.md) - Complete API documentation +- [meta-core Architecture](../../../meta-core/docs/architecture.md) - Leader election and Redis management --- ## Summary -The VFS rebuild architecture achieves **10-50x faster startup** by separating concerns: +The VFS rebuild architecture achieves **fast startup and real-time updates** through: -- **etcd** = Permanent distributed metadata archive -- **In-memory DB** = Current filesystem state -- **Discovery** = Authoritative source of truth +- **Redis Streams** = Ordered event log with replay capability +- **StreamingStateBuilder** = Memory-efficient state projection +- **Property Filtering** = Only VFS-relevant properties fetched This design enables: -- ⚡ Sub-2-second cold starts (10k files) -- 🔄 Continuous accuracy through discovery validation -- 🌐 Distributed metadata sharing via meta-orbit -- 📦 Graceful handling of offline/slow storage +- Sub-2-second cold starts (10k files) +- Real-time VFS updates via stream consumption +- Memory-efficient state (~500 bytes/file) +- Graceful reconnection and recovery -The key insight: **Fast bootstrap + continuous validation > Blocking validation at startup** +The key insight: **Stream replay + selective property fetching > Full scan + full data loading** diff --git a/packages/meta-fuse-core/src/index.ts b/packages/meta-fuse-core/src/index.ts index 1e364f1..9e2eabf 100644 --- a/packages/meta-fuse-core/src/index.ts +++ b/packages/meta-fuse-core/src/index.ts @@ -151,6 +151,10 @@ async function main(): Promise { filesPath: config.filesPath, }); + // Wire up StateBuilder to VFS for refresh operations + // This enables vfs.refresh() to replay the event stream + vfs.setStreamingStateBuilder(stateBuilder); + // Bootstrap: Replay meta:events stream from position 0 // This rebuilds state from all historical events logger.info(`Replaying ${EVENTS_STREAM} stream from position 0 (streaming bootstrap)...`); @@ -166,21 +170,9 @@ async function main(): Promise { ); const bootstrapTime = Date.now() - bootstrapStart; - let stats = vfs.getStats(); + const stats = vfs.getStats(); logger.info(`Streaming bootstrap complete: ${stats.fileCount} files in ${bootstrapTime}ms`); - // Fallback: If stream was empty but Redis has files, bootstrap from Redis keys - // This handles the case where meta-core restarted and cleared the stream - if (stats.fileCount === 0) { - const indexCount = await redisClient.getIndexCount(); - if (indexCount > 0) { - logger.info(`Stream was empty but Redis has ${indexCount} files - bootstrapping from Redis keys...`); - await vfs.refresh(); - stats = vfs.getStats(); - logger.info(`Redis key bootstrap complete: ${stats.fileCount} files`); - } - } - // Start live stream consumer from where replay left off logger.info(`Starting live stream consumer from ${lastId}...`); redisClient.startSimpleStreamConsumer( diff --git a/packages/meta-fuse-core/src/kv/IKVClient.ts b/packages/meta-fuse-core/src/kv/IKVClient.ts index 9a7944b..afaa516 100644 --- a/packages/meta-fuse-core/src/kv/IKVClient.ts +++ b/packages/meta-fuse-core/src/kv/IKVClient.ts @@ -74,9 +74,12 @@ export interface LeaderLockInfo { /** Redis connection URL (e.g., redis://10.0.1.50:6379) */ redisUrl: string; - /** WebDAV URL for file access */ + /** WebDAV URL for file access (external, via nginx/HTTPS) */ webdavUrl: string; + /** WebDAV URL for internal container-to-container access (direct to port 9000) */ + webdavUrlInternal: string; + /** Timestamp when leadership was acquired */ timestamp: number; diff --git a/packages/meta-fuse-core/src/kv/LeaderClient.ts b/packages/meta-fuse-core/src/kv/LeaderClient.ts index 50c8b1b..a01877e 100644 --- a/packages/meta-fuse-core/src/kv/LeaderClient.ts +++ b/packages/meta-fuse-core/src/kv/LeaderClient.ts @@ -32,6 +32,7 @@ export interface URLsResponse { apiUrl: string; redisUrl: string; webdavUrl: string; + webdavUrlInternal: string; isLeader: boolean; } @@ -120,6 +121,7 @@ export class LeaderClient { apiUrl: urls.apiUrl, redisUrl: urls.redisUrl, webdavUrl: urls.webdavUrl, + webdavUrlInternal: urls.webdavUrlInternal, timestamp: Date.now(), pid: 0 // Unknown for remote leader }; @@ -139,13 +141,22 @@ export class LeaderClient { } /** - * Get WebDAV URL from leader info + * Get WebDAV URL from leader info (external, via nginx/HTTPS) */ async getWebdavUrl(): Promise { const info = await this.getLeaderInfo(); return info?.webdavUrl ?? null; } + /** + * Get internal WebDAV URL from leader info (direct to port 9000) + * Use this for container-to-container communication + */ + async getWebdavUrlInternal(): Promise { + const info = await this.getLeaderInfo(); + return info?.webdavUrlInternal ?? null; + } + /** * Get meta-core API URL from leader info */ diff --git a/packages/meta-fuse-core/src/kv/RedisClient.ts b/packages/meta-fuse-core/src/kv/RedisClient.ts index ec473a7..94c7082 100644 --- a/packages/meta-fuse-core/src/kv/RedisClient.ts +++ b/packages/meta-fuse-core/src/kv/RedisClient.ts @@ -463,6 +463,76 @@ export class RedisClient implements Partial { } } + /** + * Scan flat keys to build file metadata for all indexed files + * This is used as a fallback when the meta:events stream is empty or unavailable. + * + * Algorithm: + * 1. Get all hash IDs from the index + * 2. For each hash ID, fetch VFS-relevant properties using MGET + * 3. Return a Map of hashId to flat properties + * + * This replaces the broken getAllFiles() which used HGETALL on flat keys. + * + * @param vfsRelevantProperties - Set of property names that are VFS-relevant + * @returns Map of hashId to properties (only files with filePath) + */ + async scanFlatKeysForFiles(vfsRelevantProperties: Set): Promise>> { + const result = new Map>(); + + if (!this.client || !this.isConnected) { + logger.warn('Redis not connected, returning empty result'); + return result; + } + + try { + // Get all hash IDs from the index + const hashIds = await this.getAllHashIds(); + logger.info(`Scanning flat keys for ${hashIds.length} files...`); + + if (hashIds.length === 0) { + return result; + } + + // Batch process hashIds to avoid overwhelming Redis + const batchSize = 100; + const propertiesArray = Array.from(vfsRelevantProperties); + + for (let i = 0; i < hashIds.length; i += batchSize) { + const batch = hashIds.slice(i, i + batchSize); + + // Process each hashId in the batch + const batchPromises = batch.map(async (hashId) => { + const props = await this.getProperties(hashId, propertiesArray); + + // Only include files that have a filePath + if (props.has('filePath')) { + const propsObj: Record = {}; + for (const [key, value] of props) { + propsObj[key] = value; + } + return { hashId, props: propsObj }; + } + return null; + }); + + const batchResults = await Promise.all(batchPromises); + + for (const item of batchResults) { + if (item) { + result.set(item.hashId, item.props); + } + } + } + + logger.info(`Flat key scan complete: ${result.size} files with filePath`); + return result; + } catch (error: any) { + logger.error('Failed to scan flat keys:', error.message); + return result; + } + } + /** * Scan Redis keys matching pattern */ diff --git a/packages/meta-fuse-core/src/vfs/StreamingStateBuilder.ts b/packages/meta-fuse-core/src/vfs/StreamingStateBuilder.ts index 2b45b40..0d103d1 100644 --- a/packages/meta-fuse-core/src/vfs/StreamingStateBuilder.ts +++ b/packages/meta-fuse-core/src/vfs/StreamingStateBuilder.ts @@ -101,6 +101,21 @@ export class StreamingStateBuilder { logger.info(`Updated VFS-relevant properties: ${this.vfsRelevantProps.size}`); } + /** + * Get the set of VFS-relevant properties + * Used by VFS refresh for flat key scanning fallback + */ + getVfsRelevantProperties(): Set { + return this.vfsRelevantProps; + } + + /** + * Get the current VFS callback + */ + getVFSCallback(): VFSUpdateCallback | undefined { + return this.vfsCallback; + } + /** * Process a stream message event * This is the main entry point for event processing diff --git a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts index 8826d3c..2d47d05 100644 --- a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts +++ b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts @@ -15,7 +15,7 @@ import { TemplateEngine } from './template/TemplateEngine.js'; import { ConditionEvaluator } from './template/ConditionEvaluator.js'; import { ConfigStorage } from '../config/ConfigStorage.js'; import type { RenamingConfig, RenamingRule, PreviewItem, PreviewResponse } from './types/RenamingRuleTypes.js'; -import type { VFSUpdateCallback } from './StreamingStateBuilder.js'; +import type { VFSUpdateCallback, StreamingStateBuilder } from './StreamingStateBuilder.js'; const logger = new Logger({ name: 'VirtualFileSystem' }); @@ -72,6 +72,7 @@ export class VirtualFileSystem implements VFSUpdateCallback { private config: Required; private lastRefresh: Date | null = null; private useStreamingMode: boolean = false; // Flag to indicate streaming mode + private stateBuilder: StreamingStateBuilder | null = null; // For stream replay during refresh private cachedStats = { fileCount: 0, @@ -128,6 +129,15 @@ export class VirtualFileSystem implements VFSUpdateCallback { logger.info('VFS streaming mode enabled'); } + /** + * Set the streaming state builder for refresh operations + * This enables refresh() to replay the event stream instead of using broken HGETALL + */ + setStreamingStateBuilder(builder: StreamingStateBuilder): void { + this.stateBuilder = builder; + logger.debug('StreamingStateBuilder set for VFS refresh'); + } + /** * Convert flat properties map to FileMetadata * Used when processing streaming events @@ -281,46 +291,144 @@ export class VirtualFileSystem implements VFSUpdateCallback { } /** - * Refresh the VFS from Redis - * Called once on startup for initial bootstrap. - * Subsequent updates come via Redis Streams (onFileUpdate). - * Uses template-based rules if configured, otherwise falls back to MetaDataToFolderStruct + * Refresh the VFS from Redis by replaying the event stream + * + * This method replays the meta:events stream from position 0, applying updates + * incrementally without wiping the existing VFS state. Files not seen during + * the replay are considered deleted and removed. + * + * Design principles: + * - Do NOT wipe the internal model (no initRoot() call) + * - Track which files are seen during replay + * - Remove only files that are no longer in the stream (deleted) + * - Single code path - reuses StreamingStateBuilder event processing */ async refresh(): Promise { - logger.debug('Loading VFS from Redis (bootstrap)'); + logger.info('Refreshing VFS from event stream...'); + const startTime = Date.now(); + + // Load rules configuration + this.loadRulesConfig(); + + // Check if we have a streaming state builder configured + if (!this.stateBuilder) { + logger.warn('No StreamingStateBuilder set - cannot refresh from stream'); + logger.warn('VFS refresh skipped. Set streaming state builder via setStreamingStateBuilder()'); + return; + } try { - // Load rules configuration - this.loadRulesConfig(); + // Track which hashIds we see during this refresh + const seenHashIds = new Set(); - // Get all files from Redis (keyed by sourcePath) - const files = await this.redisClient.getAllFiles(); + // Create a tracking callback that records seen hashIds + const originalCallback = this.stateBuilder.getVFSCallback(); + const trackingVFS: VFSUpdateCallback = { + onPropertyChange: (hashId, property, value) => { + this.onPropertyChange(hashId, property, value); + }, + onPropertyDelete: (hashId, property) => { + this.onPropertyDelete(hashId, property); + }, + onFileDelete: (hashId) => { + this.onFileDelete(hashId); + }, + onFileComplete: (hashId, metadata) => { + seenHashIds.add(hashId); + this.onFileComplete(hashId, metadata); + }, + }; - if (files.size === 0) { - logger.info('No files found in Redis'); - this.initRoot(); - return; - } + // Temporarily set tracking callback + this.stateBuilder.setVFSCallback(trackingVFS); - // Rebuild VFS with computed paths - this.initRoot(); + // Clear state builder's internal state (but NOT VFS nodes) + this.stateBuilder.clear(); - for (const [sourcePath, metadata] of files) { - // Compute virtual path using rules or fallback - const virtualPath = this.computeVirtualPath(metadata, sourcePath); - if (virtualPath) { - await this.addFile(virtualPath, metadata); - this.sourcePathToVirtualPath.set(sourcePath, virtualPath); + // Replay meta:events stream from position 0 + const lastId = await this.redisClient.replayStream( + 'meta:events', + '0', + async (message) => { + await this.stateBuilder!.processEvent(message); + }, + 100 // Batch size + ); + + // Fallback: If no files found from stream replay, scan flat keys directly + // This handles the case where meta:events stream isn't being populated + // but data exists in flat key format + if (seenHashIds.size === 0) { + const indexCount = await this.redisClient.getIndexCount(); + if (indexCount > 0) { + logger.info(`Stream replay found no files but index has ${indexCount} - scanning flat keys...`); + + // Get VFS-relevant properties from state builder + const vfsRelevantProps = this.stateBuilder.getVfsRelevantProperties(); + + // Scan flat keys + const flatKeyFiles = await this.redisClient.scanFlatKeysForFiles(vfsRelevantProps); + + // Process each file through onFileComplete + for (const [hashId, props] of flatKeyFiles) { + seenHashIds.add(hashId); + this.onFileComplete(hashId, props); + } + + logger.info(`Flat key scan populated ${seenHashIds.size} files`); } } + // Restore original callback + this.stateBuilder.setVFSCallback(originalCallback || this); + + // Remove files not seen during replay (these were deleted) + const toRemove: string[] = []; + for (const [hashId, virtualPath] of this.hashIdToVirtualPath) { + if (!seenHashIds.has(hashId)) { + logger.debug(`Removing deleted file: ${hashId} -> ${virtualPath}`); + toRemove.push(hashId); + } + } + + for (const hashId of toRemove) { + const virtualPath = this.hashIdToVirtualPath.get(hashId); + if (virtualPath) { + this.removeFile(virtualPath); + } + this.hashIdToVirtualPath.delete(hashId); + } + this.lastRefresh = new Date(); - logger.info(`VFS refreshed: ${this.cachedStats.fileCount} files, ${this.cachedStats.directoryCount} directories`); + this.updateStats(); + const elapsed = Date.now() - startTime; + logger.info(`VFS refresh complete: ${this.cachedStats.fileCount} files, ${this.cachedStats.directoryCount} directories in ${elapsed}ms`); + logger.info(`Replay ended at stream position: ${lastId}, removed ${toRemove.length} deleted files`); } catch (error) { logger.error('Failed to refresh VFS:', error); } } + /** + * Update cached stats from current nodes + */ + private updateStats(): void { + let fileCount = 0; + let directoryCount = 0; + let totalSize = 0; + + for (const node of this.nodes.values()) { + if (node.type === 'file') { + fileCount++; + totalSize += node.size ?? 0; + } else { + directoryCount++; + } + } + + this.cachedStats = { fileCount, directoryCount, totalSize }; + } + /** * Load rules configuration from file * ConfigStorage creates the default config file if it doesn't exist