fix queue and stats
This commit is contained in:
Pierre
@@ -822,7 +822,6 @@ gdrive: /files/gdrive rclone vfs-cache-mode=full 0 0
|
||||
"totalLightProcessing": 16,
|
||||
"totalHashProcessing": 8,
|
||||
"totalDone": 45000,
|
||||
"totalFilesInVFS": 45150,
|
||||
"throughput": {
|
||||
"filesPerSecond": 34,
|
||||
"avgLightProcessingMs": 470,
|
||||
|
||||
@@ -1,173 +0,0 @@
|
||||
/**
|
||||
* FUSE API for meta-mesh
|
||||
*
|
||||
* This API provides a virtual filesystem interface that can be consumed by a FUSE driver.
|
||||
* It exposes the computed folder structure without creating actual hardlinks.
|
||||
*/
|
||||
|
||||
import { HashMeta } from '@metazla/meta-interface';
|
||||
import { Stats } from 'fs';
|
||||
|
||||
/**
|
||||
* Represents a file or directory in the virtual filesystem
|
||||
*/
|
||||
export interface VirtualNode {
|
||||
/** Virtual path in the output filesystem */
|
||||
path: string;
|
||||
/** Node type */
|
||||
type: 'file' | 'directory';
|
||||
/** File name */
|
||||
name: string;
|
||||
/** Parent directory path */
|
||||
parent: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Represents a regular file in the virtual filesystem
|
||||
*/
|
||||
export interface VirtualFile extends VirtualNode {
|
||||
type: 'file';
|
||||
/** Source file path on the real filesystem (null for virtual files like .meta) */
|
||||
sourcePath: string | null;
|
||||
/** File size in bytes */
|
||||
size: number;
|
||||
/** Last modified time */
|
||||
mtime: Date;
|
||||
/** Creation time */
|
||||
ctime: Date;
|
||||
/** File permissions (octal) */
|
||||
mode: number;
|
||||
/** Associated metadata */
|
||||
metadata?: HashMeta;
|
||||
}
|
||||
|
||||
/**
|
||||
* Represents a directory in the virtual filesystem
|
||||
*/
|
||||
export interface VirtualDirectory extends VirtualNode {
|
||||
type: 'directory';
|
||||
/** Child nodes (file/directory names) */
|
||||
children: string[];
|
||||
/** Directory permissions (octal) */
|
||||
mode: number;
|
||||
/** Last modified time */
|
||||
mtime: Date;
|
||||
}
|
||||
|
||||
/**
|
||||
* File attributes returned by getattr() - compatible with FUSE stat structure
|
||||
*/
|
||||
export interface FileAttributes {
|
||||
/** File size in bytes */
|
||||
size: number;
|
||||
/** File mode (permissions + type) */
|
||||
mode: number;
|
||||
/** Last modified time (Unix timestamp in seconds) */
|
||||
mtime: number;
|
||||
/** Last access time (Unix timestamp in seconds) */
|
||||
atime: number;
|
||||
/** Creation/change time (Unix timestamp in seconds) */
|
||||
ctime: number;
|
||||
/** Number of hard links */
|
||||
nlink: number;
|
||||
/** User ID */
|
||||
uid: number;
|
||||
/** Group ID */
|
||||
gid: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read result for file content
|
||||
*/
|
||||
export interface ReadResult {
|
||||
/** For regular files: source path to read from. For virtual files: null */
|
||||
sourcePath: string | null;
|
||||
/** For virtual files: generated content. For regular files: null */
|
||||
content: Buffer | null;
|
||||
/** Total file size */
|
||||
size: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* FUSE API operations
|
||||
*/
|
||||
export interface IFuseAPI {
|
||||
/**
|
||||
* List directory contents
|
||||
* @param path Virtual directory path (e.g., "/Anime/Naruto/Season 1")
|
||||
* @returns Array of file/directory names in the directory, or null if not found
|
||||
*/
|
||||
readdir(path: string): string[] | null;
|
||||
|
||||
/**
|
||||
* Get file or directory attributes
|
||||
* @param path Virtual file or directory path
|
||||
* @returns File attributes, or null if not found
|
||||
*/
|
||||
getattr(path: string): FileAttributes | null;
|
||||
|
||||
/**
|
||||
* Check if a path exists
|
||||
* @param path Virtual path
|
||||
* @returns true if path exists, false otherwise
|
||||
*/
|
||||
exists(path: string): boolean;
|
||||
|
||||
/**
|
||||
* Get read information for a file
|
||||
* @param path Virtual file path
|
||||
* @returns Read result with source path or content, or null if not found
|
||||
*/
|
||||
read(path: string): ReadResult | null;
|
||||
|
||||
/**
|
||||
* Get metadata for a file
|
||||
* @param path Virtual file path
|
||||
* @returns HashMeta if available, null otherwise
|
||||
*/
|
||||
getMetadata(path: string): HashMeta | null;
|
||||
|
||||
/**
|
||||
* Get the complete virtual filesystem tree
|
||||
* @returns Root directory node
|
||||
*/
|
||||
getTree(): VirtualDirectory;
|
||||
|
||||
/**
|
||||
* Refresh the virtual filesystem from current state
|
||||
* Called after processing completes to update the VFS
|
||||
*/
|
||||
refresh(): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Events emitted by the FUSE API
|
||||
*/
|
||||
export interface FuseAPIEvents {
|
||||
/** Emitted when the virtual filesystem is updated */
|
||||
'vfs-updated': () => void;
|
||||
/** Emitted when a file is added */
|
||||
'file-added': (path: string) => void;
|
||||
/** Emitted when a file is updated (e.g., metadata changed) */
|
||||
'file-updated': (path: string) => void;
|
||||
/** Emitted when a file is removed */
|
||||
'file-removed': (path: string) => void;
|
||||
/** Emitted when a directory is added */
|
||||
'directory-added': (path: string) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Configuration for FUSE API
|
||||
*/
|
||||
export interface FuseAPIConfig {
|
||||
/** Base path for the virtual filesystem (default: "/") */
|
||||
basePath?: string;
|
||||
/** Default file permissions (default: 0o644) */
|
||||
fileMode?: number;
|
||||
/** Default directory permissions (default: 0o755) */
|
||||
directoryMode?: number;
|
||||
/** User ID for files (default: process.getuid()) */
|
||||
uid?: number;
|
||||
/** Group ID for files (default: process.getgid()) */
|
||||
gid?: number;
|
||||
}
|
||||
@@ -1,345 +0,0 @@
|
||||
/**
|
||||
* HTTP REST API Server for FUSE operations
|
||||
*
|
||||
* Exposes the VirtualFileSystem over HTTP for consumption by a FUSE driver.
|
||||
*/
|
||||
|
||||
import * as http from 'http';
|
||||
import { VirtualFileSystem } from './VirtualFileSystem.js';
|
||||
import { FileAttributes, ReadResult } from './FuseAPI.js';
|
||||
|
||||
export interface FuseAPIServerConfig {
|
||||
/** Port to listen on (default: 3000) */
|
||||
port?: number;
|
||||
/** Host to bind to (default: 'localhost') */
|
||||
host?: string;
|
||||
/** Enable CORS (default: false) */
|
||||
enableCors?: boolean;
|
||||
}
|
||||
|
||||
export class FuseAPIServer {
|
||||
private server: http.Server | null = null;
|
||||
private vfs: VirtualFileSystem;
|
||||
private config: Required<FuseAPIServerConfig>;
|
||||
|
||||
constructor(vfs: VirtualFileSystem, config: FuseAPIServerConfig = {}) {
|
||||
this.vfs = vfs;
|
||||
this.config = {
|
||||
port: config.port ?? 3000,
|
||||
host: config.host ?? 'localhost',
|
||||
enableCors: config.enableCors ?? false
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the HTTP server
|
||||
*/
|
||||
public async start(): Promise<void> {
|
||||
return new Promise((resolve, reject) => {
|
||||
this.server = http.createServer(this.handleRequest.bind(this));
|
||||
|
||||
this.server.on('error', reject);
|
||||
|
||||
this.server.listen(this.config.port, this.config.host, () => {
|
||||
console.log(`FUSE API Server listening on http://${this.config.host}:${this.config.port}`);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the HTTP server
|
||||
*/
|
||||
public async stop(): Promise<void> {
|
||||
if (!this.server) {
|
||||
return;
|
||||
}
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
this.server!.close((err) => {
|
||||
if (err) {
|
||||
reject(err);
|
||||
} else {
|
||||
console.log('FUSE API Server stopped');
|
||||
this.server = null;
|
||||
resolve();
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle incoming HTTP requests
|
||||
*/
|
||||
private async handleRequest(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> {
|
||||
// CORS headers
|
||||
if (this.config.enableCors) {
|
||||
res.setHeader('Access-Control-Allow-Origin', '*');
|
||||
res.setHeader('Access-Control-Allow-Methods', 'GET, POST, OPTIONS');
|
||||
res.setHeader('Access-Control-Allow-Headers', 'Content-Type');
|
||||
|
||||
if (req.method === 'OPTIONS') {
|
||||
res.writeHead(204);
|
||||
res.end();
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const url = new URL(req.url!, `http://${req.headers.host}`);
|
||||
const pathname = url.pathname;
|
||||
const method = req.method!;
|
||||
|
||||
// Route the request
|
||||
if (method === 'GET' && pathname === '/api/fuse/health') {
|
||||
this.handleHealth(res);
|
||||
} else if (method === 'GET' && pathname === '/api/fuse/stats') {
|
||||
this.handleStats(res);
|
||||
} else if (method === 'POST' && pathname === '/api/fuse/readdir') {
|
||||
await this.handleReaddir(req, res);
|
||||
} else if (method === 'POST' && pathname === '/api/fuse/getattr') {
|
||||
await this.handleGetattr(req, res);
|
||||
} else if (method === 'POST' && pathname === '/api/fuse/exists') {
|
||||
await this.handleExists(req, res);
|
||||
} else if (method === 'POST' && pathname === '/api/fuse/read') {
|
||||
await this.handleRead(req, res);
|
||||
} else if (method === 'POST' && pathname === '/api/fuse/metadata') {
|
||||
await this.handleMetadata(req, res);
|
||||
} else if (method === 'GET' && pathname === '/api/fuse/tree') {
|
||||
this.handleTree(res);
|
||||
} else if (method === 'GET' && pathname === '/api/fuse/files') {
|
||||
this.handleFiles(res);
|
||||
} else if (method === 'GET' && pathname === '/api/fuse/directories') {
|
||||
this.handleDirectories(res);
|
||||
} else if (method === 'POST' && pathname === '/api/fuse/refresh') {
|
||||
await this.handleRefresh(req, res);
|
||||
} else {
|
||||
this.sendError(res, 404, 'Not Found');
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error handling request:', error);
|
||||
this.sendError(res, 500, 'Internal Server Error', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse JSON body from request
|
||||
*/
|
||||
private async parseBody(req: http.IncomingMessage): Promise<any> {
|
||||
return new Promise((resolve, reject) => {
|
||||
let body = '';
|
||||
req.on('data', (chunk) => {
|
||||
body += chunk.toString();
|
||||
});
|
||||
req.on('end', () => {
|
||||
try {
|
||||
resolve(JSON.parse(body));
|
||||
} catch (error) {
|
||||
reject(new Error('Invalid JSON'));
|
||||
}
|
||||
});
|
||||
req.on('error', reject);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Send JSON response
|
||||
*/
|
||||
private sendJson(res: http.ServerResponse, data: any, statusCode: number = 200): void {
|
||||
res.writeHead(statusCode, { 'Content-Type': 'application/json' });
|
||||
res.end(JSON.stringify(data));
|
||||
}
|
||||
|
||||
/**
|
||||
* Send error response
|
||||
*/
|
||||
private sendError(res: http.ServerResponse, statusCode: number, message: string, error?: any): void {
|
||||
this.sendJson(res, {
|
||||
error: message,
|
||||
details: error?.message ?? error
|
||||
}, statusCode);
|
||||
}
|
||||
|
||||
// API Handlers
|
||||
|
||||
/**
|
||||
* GET /api/health
|
||||
* Health check endpoint
|
||||
*/
|
||||
private handleHealth(res: http.ServerResponse): void {
|
||||
this.sendJson(res, {
|
||||
status: 'ok',
|
||||
timestamp: new Date().toISOString()
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* GET /api/stats
|
||||
* Get VFS statistics
|
||||
*/
|
||||
private handleStats(res: http.ServerResponse): void {
|
||||
const stats = this.vfs.getStats();
|
||||
this.sendJson(res, stats);
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/readdir
|
||||
* List directory contents
|
||||
* Body: { path: string }
|
||||
* Response: { entries: string[] } | { error: string }
|
||||
*/
|
||||
private async handleReaddir(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> {
|
||||
const { path } = await this.parseBody(req);
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
this.sendError(res, 400, 'Missing or invalid "path" parameter');
|
||||
return;
|
||||
}
|
||||
|
||||
const entries = this.vfs.readdir(path);
|
||||
|
||||
if (entries === null) {
|
||||
this.sendError(res, 404, 'Directory not found');
|
||||
return;
|
||||
}
|
||||
|
||||
this.sendJson(res, { entries });
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/getattr
|
||||
* Get file attributes
|
||||
* Body: { path: string }
|
||||
* Response: FileAttributes | { error: string }
|
||||
*/
|
||||
private async handleGetattr(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> {
|
||||
const { path } = await this.parseBody(req);
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
this.sendError(res, 400, 'Missing or invalid "path" parameter');
|
||||
return;
|
||||
}
|
||||
|
||||
const attrs = this.vfs.getattr(path);
|
||||
|
||||
if (attrs === null) {
|
||||
this.sendError(res, 404, 'Path not found');
|
||||
return;
|
||||
}
|
||||
|
||||
this.sendJson(res, attrs);
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/exists
|
||||
* Check if path exists
|
||||
* Body: { path: string }
|
||||
* Response: { exists: boolean }
|
||||
*/
|
||||
private async handleExists(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> {
|
||||
const { path } = await this.parseBody(req);
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
this.sendError(res, 400, 'Missing or invalid "path" parameter');
|
||||
return;
|
||||
}
|
||||
|
||||
const exists = this.vfs.exists(path);
|
||||
this.sendJson(res, { exists });
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/read
|
||||
* Read file
|
||||
* Body: { path: string }
|
||||
* Response: ReadResult | { error: string }
|
||||
*/
|
||||
private async handleRead(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> {
|
||||
const { path } = await this.parseBody(req);
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
this.sendError(res, 400, 'Missing or invalid "path" parameter');
|
||||
return;
|
||||
}
|
||||
|
||||
const result = this.vfs.read(path);
|
||||
|
||||
if (result === null) {
|
||||
this.sendError(res, 404, 'File not found');
|
||||
return;
|
||||
}
|
||||
|
||||
// Convert Buffer to base64 for JSON serialization
|
||||
const response: any = {
|
||||
sourcePath: result.sourcePath,
|
||||
size: result.size
|
||||
};
|
||||
|
||||
if (result.content !== null) {
|
||||
response.content = result.content.toString('base64');
|
||||
response.contentEncoding = 'base64';
|
||||
}
|
||||
|
||||
this.sendJson(res, response);
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/metadata
|
||||
* Get file metadata
|
||||
* Body: { path: string }
|
||||
* Response: HashMeta | { error: string }
|
||||
*/
|
||||
private async handleMetadata(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> {
|
||||
const { path } = await this.parseBody(req);
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
this.sendError(res, 400, 'Missing or invalid "path" parameter');
|
||||
return;
|
||||
}
|
||||
|
||||
const metadata = this.vfs.getMetadata(path);
|
||||
|
||||
if (metadata === null) {
|
||||
this.sendError(res, 404, 'Metadata not found');
|
||||
return;
|
||||
}
|
||||
|
||||
this.sendJson(res, metadata);
|
||||
}
|
||||
|
||||
/**
|
||||
* GET /api/tree
|
||||
* Get complete VFS tree
|
||||
*/
|
||||
private handleTree(res: http.ServerResponse): void {
|
||||
const tree = this.vfs.getTree();
|
||||
this.sendJson(res, tree);
|
||||
}
|
||||
|
||||
/**
|
||||
* GET /api/files
|
||||
* Get all file paths
|
||||
*/
|
||||
private handleFiles(res: http.ServerResponse): void {
|
||||
const files = this.vfs.getAllFiles();
|
||||
this.sendJson(res, { files });
|
||||
}
|
||||
|
||||
/**
|
||||
* GET /api/directories
|
||||
* Get all directory paths
|
||||
*/
|
||||
private handleDirectories(res: http.ServerResponse): void {
|
||||
const directories = this.vfs.getAllDirectories();
|
||||
this.sendJson(res, { directories });
|
||||
}
|
||||
|
||||
/**
|
||||
* POST /api/refresh
|
||||
* Refresh the VFS (no-op, VFS is updated by processing pipeline)
|
||||
*/
|
||||
private async handleRefresh(req: http.IncomingMessage, res: http.ServerResponse): Promise<void> {
|
||||
await this.vfs.refresh();
|
||||
this.sendJson(res, { status: 'ok' });
|
||||
}
|
||||
}
|
||||
@@ -1,299 +0,0 @@
|
||||
# FUSE API for Meta-Mesh
|
||||
|
||||
This module provides a FUSE-compatible HTTP API that runs alongside meta-mesh's hardlink system. The API exposes a virtual filesystem that can be consumed by a future FUSE driver.
|
||||
|
||||
## Overview
|
||||
|
||||
The FUSE API runs **always-on alongside hardlinks**, providing:
|
||||
|
||||
- **Dual output**: Both hardlinks (for local access) and virtual filesystem (for FUSE driver)
|
||||
- **Dynamic updates**: VirtualFileSystem rebuilds on every processing cycle
|
||||
- **Virtual .meta files**: Metadata files are generated on-the-fly without disk I/O
|
||||
- **HTTP REST API**: Can be consumed by local or remote FUSE drivers
|
||||
- **Zero configuration**: Always enabled, no opt-in required
|
||||
|
||||
## Architecture
|
||||
|
||||
### Components
|
||||
|
||||
1. **VirtualFileSystem** - In-memory representation of the output directory structure
|
||||
2. **FuseAPIServer** - HTTP REST API server exposing FUSE operations
|
||||
3. **FuseAPI** - TypeScript interfaces defining the API contract
|
||||
|
||||
### Data Flow
|
||||
|
||||
```
|
||||
Input Files → File Processing → Metadata Extraction → Duplicate Removal
|
||||
↓
|
||||
Virtual Folder Structure Computation (diff generation)
|
||||
↓
|
||||
├─→ VirtualFileSystem Build (in-memory) → FUSE API Server → Future FUSE Driver
|
||||
│
|
||||
└─→ Hardlink Creation (on disk) → Direct File Access
|
||||
```
|
||||
|
||||
Both outputs are generated in parallel from the same computed folder structure.
|
||||
|
||||
## Configuration
|
||||
|
||||
The FUSE API is **always enabled** and runs alongside hardlinks. You can optionally configure the server port and host:
|
||||
|
||||
```bash
|
||||
# FUSE API Server configuration (optional)
|
||||
FUSE_API_PORT=3000 # Default: 3000
|
||||
FUSE_API_HOST=localhost # Default: localhost
|
||||
```
|
||||
|
||||
The API server starts automatically when meta-mesh starts.
|
||||
|
||||
## API Endpoints
|
||||
|
||||
The FUSE API Server exposes the following HTTP REST endpoints:
|
||||
|
||||
### Health & Info
|
||||
|
||||
- **GET /api/health** - Health check
|
||||
```json
|
||||
{ "status": "ok", "timestamp": "2025-01-15T10:30:00.000Z" }
|
||||
```
|
||||
|
||||
- **GET /api/stats** - VFS statistics
|
||||
```json
|
||||
{
|
||||
"fileCount": 1250,
|
||||
"directoryCount": 45,
|
||||
"totalSize": 52428800,
|
||||
"metaFileCount": 1250
|
||||
}
|
||||
```
|
||||
|
||||
### Directory Operations
|
||||
|
||||
- **POST /api/readdir** - List directory contents
|
||||
```json
|
||||
// Request
|
||||
{ "path": "/Anime/Naruto/Season 1" }
|
||||
|
||||
// Response
|
||||
{ "entries": ["Naruto S01E01.mkv", "Naruto S01E01.mkv.meta", ...] }
|
||||
```
|
||||
|
||||
### File Operations
|
||||
|
||||
- **POST /api/getattr** - Get file attributes
|
||||
```json
|
||||
// Request
|
||||
{ "path": "/Anime/Naruto/Season 1/Naruto S01E01.mkv" }
|
||||
|
||||
// Response
|
||||
{
|
||||
"size": 524288000,
|
||||
"mode": 33188, // 0o100644 (regular file, rw-r--r--)
|
||||
"mtime": 1705315200,
|
||||
"atime": 1705315200,
|
||||
"ctime": 1705315200,
|
||||
"nlink": 1,
|
||||
"uid": 1000,
|
||||
"gid": 1000
|
||||
}
|
||||
```
|
||||
|
||||
- **POST /api/exists** - Check if path exists
|
||||
```json
|
||||
// Request
|
||||
{ "path": "/Anime/Naruto" }
|
||||
|
||||
// Response
|
||||
{ "exists": true }
|
||||
```
|
||||
|
||||
- **POST /api/read** - Read file
|
||||
```json
|
||||
// Request
|
||||
{ "path": "/Anime/Naruto/Season 1/Naruto S01E01.mkv" }
|
||||
|
||||
// Response (regular file)
|
||||
{
|
||||
"sourcePath": "/data/input/naruto_01.mkv",
|
||||
"size": 524288000
|
||||
}
|
||||
|
||||
// Response (virtual .meta file)
|
||||
{
|
||||
"sourcePath": null,
|
||||
"content": "..base64-encoded content..",
|
||||
"contentEncoding": "base64",
|
||||
"size": 1024
|
||||
}
|
||||
```
|
||||
|
||||
- **POST /api/metadata** - Get file metadata
|
||||
```json
|
||||
// Request
|
||||
{ "path": "/Anime/Naruto/Season 1/Naruto S01E01.mkv" }
|
||||
|
||||
// Response
|
||||
{
|
||||
"title": "Naruto",
|
||||
"season": "01",
|
||||
"episode": "01",
|
||||
"anime": true,
|
||||
"sha256": "abc123...",
|
||||
// ... full HashMeta object
|
||||
}
|
||||
```
|
||||
|
||||
### Tree Operations
|
||||
|
||||
- **GET /api/tree** - Get complete VFS tree structure
|
||||
- **GET /api/files** - Get all file paths
|
||||
- **GET /api/directories** - Get all directory paths
|
||||
- **POST /api/refresh** - Refresh VFS (no-op, VFS updates automatically)
|
||||
|
||||
## Usage Example
|
||||
|
||||
### Starting meta-mesh with FUSE API
|
||||
|
||||
The FUSE API starts automatically with meta-mesh:
|
||||
|
||||
```bash
|
||||
# Run meta-mesh (FUSE API starts automatically on port 3000)
|
||||
pnpm run start:mesh
|
||||
|
||||
# Or customize the port
|
||||
export FUSE_API_PORT=8080
|
||||
pnpm run start:mesh
|
||||
```
|
||||
|
||||
You'll see output like:
|
||||
```
|
||||
FUSE API Server started on http://localhost:3000
|
||||
VirtualFileSystem build took 15ms
|
||||
VirtualFileSystem: 1250 files (1250 .meta), 45 directories, 50.00 MB
|
||||
```
|
||||
|
||||
### Consuming the API
|
||||
|
||||
```typescript
|
||||
import { VirtualFileSystem, FuseAPIServer } from '@metazla/meta-mesh/api';
|
||||
|
||||
// Create VFS instance
|
||||
const vfs = new VirtualFileSystem({
|
||||
fileMode: 0o644,
|
||||
directoryMode: 0o755,
|
||||
uid: process.getuid(),
|
||||
gid: process.getgid()
|
||||
});
|
||||
|
||||
// Build from computed structure
|
||||
vfs.buildFromComputed(sourceToDestMap, metadataMap);
|
||||
|
||||
// Start API server
|
||||
const server = new FuseAPIServer(vfs, {
|
||||
port: 3000,
|
||||
host: 'localhost'
|
||||
});
|
||||
|
||||
await server.start();
|
||||
```
|
||||
|
||||
### FUSE Driver Integration
|
||||
|
||||
A FUSE driver would make HTTP requests to these endpoints:
|
||||
|
||||
```typescript
|
||||
// Example pseudo-code for FUSE driver
|
||||
async function readdir(path: string): Promise<string[]> {
|
||||
const response = await fetch('http://localhost:3000/api/readdir', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ path })
|
||||
});
|
||||
|
||||
const { entries } = await response.json();
|
||||
return entries;
|
||||
}
|
||||
|
||||
async function read(path: string): Promise<Buffer> {
|
||||
const response = await fetch('http://localhost:3000/api/read', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ path })
|
||||
});
|
||||
|
||||
const result = await response.json();
|
||||
|
||||
if (result.sourcePath) {
|
||||
// Regular file - read from source
|
||||
return fs.readFileSync(result.sourcePath);
|
||||
} else {
|
||||
// Virtual file - use provided content
|
||||
return Buffer.from(result.content, 'base64');
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Virtual .meta Files
|
||||
|
||||
`.meta` files are entirely virtual and generated on-the-fly:
|
||||
|
||||
- **No disk I/O**: Content is serialized from in-memory metadata
|
||||
- **Always up-to-date**: Reflects current metadata state
|
||||
- **YAML format**: Uses `MetaMeshFormat.serialize()` for consistency
|
||||
- **No source path**: `sourcePath` is `null`, `content` contains the YAML
|
||||
|
||||
## Events
|
||||
|
||||
The VirtualFileSystem emits events for monitoring:
|
||||
|
||||
```typescript
|
||||
vfs.on('vfs-updated', () => {
|
||||
console.log('Virtual filesystem updated');
|
||||
});
|
||||
|
||||
vfs.on('file-added', (path: string) => {
|
||||
console.log(`File added: ${path}`);
|
||||
});
|
||||
|
||||
vfs.on('file-removed', (path: string) => {
|
||||
console.log(`File removed: ${path}`);
|
||||
});
|
||||
|
||||
vfs.on('directory-added', (path: string) => {
|
||||
console.log(`Directory added: ${path}`);
|
||||
});
|
||||
```
|
||||
|
||||
## Performance
|
||||
|
||||
The VirtualFileSystem is highly performant:
|
||||
|
||||
- **In-memory**: All operations are O(1) or O(n) in-memory
|
||||
- **No disk I/O**: Except for reading source files when FUSE driver requests content
|
||||
- **Lazy .meta generation**: Content is only serialized when requested
|
||||
- **Fast updates**: Rebuilding the VFS takes milliseconds
|
||||
|
||||
## Dual Output: Hardlinks + FUSE API
|
||||
|
||||
Meta-mesh now provides **both** output methods simultaneously:
|
||||
|
||||
| Feature | Hardlinks | FUSE API | Best For |
|
||||
|---------|-----------|----------|----------|
|
||||
| Disk usage | 1 inode per file | 0 (virtual) | Hardlinks: local use |
|
||||
| Update speed | Slow (file I/O) | Fast (in-memory) | FUSE: frequent updates |
|
||||
| .meta files | Written to disk | Virtual (on-demand) | FUSE: dynamic metadata |
|
||||
| Remote access | No | Yes (HTTP) | FUSE: network drives |
|
||||
| Compatibility | Any filesystem | Requires FUSE driver | Hardlinks: broad compat |
|
||||
| Access method | Direct file access | HTTP + FUSE driver | Hardlinks: native apps |
|
||||
|
||||
**Use hardlinks** for local applications like Jellyfin, Plex, Kodi that read files directly.
|
||||
|
||||
**Use FUSE API** for future FUSE drivers that want dynamic, virtual filesystem access.
|
||||
|
||||
## Future Enhancements
|
||||
|
||||
- **WebSocket support**: Real-time updates for FUSE drivers
|
||||
- **Authentication**: API key or JWT-based auth
|
||||
- **Compression**: Gzip/Brotli for API responses
|
||||
- **Caching**: ETag/Last-Modified headers
|
||||
- **Batch operations**: Multi-file operations in single request
|
||||
@@ -7,7 +7,6 @@
|
||||
|
||||
import Fastify, { FastifyInstance } from 'fastify';
|
||||
import fastifyCors from '@fastify/cors';
|
||||
import { VirtualFileSystem } from './VirtualFileSystem.js';
|
||||
import { performanceMetrics } from '../metrics/PerformanceMetrics.js';
|
||||
import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js';
|
||||
import type { IKVClient } from '../kv/IKVClient.js';
|
||||
@@ -32,7 +31,6 @@ export interface UnifiedAPIServerConfig {
|
||||
|
||||
export class UnifiedAPIServer {
|
||||
private app: FastifyInstance;
|
||||
private vfs: VirtualFileSystem;
|
||||
private unifiedStateManager: UnifiedProcessingStateManager | null = null;
|
||||
private kvClient: IKVClient | null = null;
|
||||
private kvManager: KVManager | null = null;
|
||||
@@ -51,7 +49,6 @@ export class UnifiedAPIServer {
|
||||
private static TOTAL_SIZE_CACHE_TTL_MS = 30000; // 30 seconds
|
||||
|
||||
constructor(
|
||||
vfs: VirtualFileSystem,
|
||||
config: UnifiedAPIServerConfig = {},
|
||||
unifiedStateManager?: UnifiedProcessingStateManager,
|
||||
kvClient?: IKVClient,
|
||||
@@ -62,7 +59,6 @@ export class UnifiedAPIServer {
|
||||
getPluginManager?: () => PluginManager | null,
|
||||
getTaskScheduler?: () => TaskScheduler | null
|
||||
) {
|
||||
this.vfs = vfs;
|
||||
this.unifiedStateManager = unifiedStateManager || null;
|
||||
this.kvClient = kvClient || null;
|
||||
this.kvManager = kvManager || null;
|
||||
@@ -155,7 +151,6 @@ export class UnifiedAPIServer {
|
||||
status: 'ok',
|
||||
timestamp: new Date().toISOString(),
|
||||
services: {
|
||||
fuse: 'ok',
|
||||
metrics: 'ok',
|
||||
unifiedProcessing: this.unifiedStateManager ? 'ok' : 'unavailable'
|
||||
}
|
||||
@@ -176,9 +171,6 @@ export class UnifiedAPIServer {
|
||||
};
|
||||
});
|
||||
|
||||
// FUSE API routes (under /api/fuse)
|
||||
this.setupFuseRoutes();
|
||||
|
||||
// Metrics API routes (under /api/metrics)
|
||||
this.setupMetricsRoutes();
|
||||
|
||||
@@ -288,191 +280,6 @@ export class UnifiedAPIServer {
|
||||
// - POST /api/mounts/:id/unmount - Request unmount
|
||||
// - GET /api/mounts/rclone/remotes - List rclone remotes
|
||||
|
||||
/**
|
||||
* Setup FUSE API routes
|
||||
*/
|
||||
private setupFuseRoutes(): void {
|
||||
// Health check
|
||||
this.app.get('/api/fuse/health', async (request, reply) => {
|
||||
return {
|
||||
status: 'ok',
|
||||
timestamp: new Date().toISOString()
|
||||
};
|
||||
});
|
||||
|
||||
// VFS statistics
|
||||
this.app.get('/api/fuse/stats', async (request, reply) => {
|
||||
return this.vfs.getStats();
|
||||
});
|
||||
|
||||
// Database statistics (KV)
|
||||
this.app.get('/api/fuse/db-stats', async (request, reply) => {
|
||||
if (!this.kvClient) {
|
||||
return reply.status(503).send({
|
||||
error: 'KV client not available'
|
||||
});
|
||||
}
|
||||
|
||||
try {
|
||||
// Count files in database with /file/ prefix
|
||||
const fileCount = await this.kvClient.countKeysWithPrefix('/file/');
|
||||
return {
|
||||
fileCount,
|
||||
source: 'KV'
|
||||
};
|
||||
} catch (error: any) {
|
||||
return reply.status(500).send({
|
||||
error: 'Failed to retrieve database stats',
|
||||
details: error.message
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
// List directory contents
|
||||
this.app.post<{ Body: { path: string } }>('/api/fuse/readdir', async (request, reply) => {
|
||||
const { path } = request.body;
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
return reply.status(400).send({
|
||||
error: 'Missing or invalid "path" parameter'
|
||||
});
|
||||
}
|
||||
|
||||
const entries = this.vfs.readdir(path);
|
||||
|
||||
if (entries === null) {
|
||||
return reply.status(404).send({ error: 'Directory not found' });
|
||||
}
|
||||
|
||||
return { entries };
|
||||
});
|
||||
|
||||
// Get file attributes
|
||||
this.app.post<{ Body: { path: string } }>('/api/fuse/getattr', async (request, reply) => {
|
||||
const { path } = request.body;
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
return reply.status(400).send({
|
||||
error: 'Missing or invalid "path" parameter'
|
||||
});
|
||||
}
|
||||
|
||||
const attrs = this.vfs.getattr(path);
|
||||
|
||||
if (attrs === null) {
|
||||
return reply.status(404).send({ error: 'Path not found' });
|
||||
}
|
||||
|
||||
return attrs;
|
||||
});
|
||||
|
||||
// Check if path exists
|
||||
this.app.post<{ Body: { path: string } }>('/api/fuse/exists', async (request, reply) => {
|
||||
const { path } = request.body;
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
return reply.status(400).send({
|
||||
error: 'Missing or invalid "path" parameter'
|
||||
});
|
||||
}
|
||||
|
||||
const exists = this.vfs.exists(path);
|
||||
return { exists };
|
||||
});
|
||||
|
||||
// Read file
|
||||
this.app.post<{ Body: { path: string } }>('/api/fuse/read', async (request, reply) => {
|
||||
const { path } = request.body;
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
return reply.status(400).send({
|
||||
error: 'Missing or invalid "path" parameter'
|
||||
});
|
||||
}
|
||||
|
||||
const result = this.vfs.read(path);
|
||||
|
||||
if (result === null) {
|
||||
return reply.status(404).send({ error: 'File not found' });
|
||||
}
|
||||
|
||||
// Convert Buffer to base64 for JSON serialization
|
||||
const response: any = {
|
||||
sourcePath: result.sourcePath,
|
||||
size: result.size
|
||||
};
|
||||
|
||||
if (result.content !== null) {
|
||||
response.content = result.content.toString('base64');
|
||||
response.contentEncoding = 'base64';
|
||||
}
|
||||
|
||||
return response;
|
||||
});
|
||||
|
||||
// Get file metadata
|
||||
this.app.post<{ Body: { path: string } }>('/api/fuse/metadata', async (request, reply) => {
|
||||
const { path } = request.body;
|
||||
|
||||
if (typeof path !== 'string') {
|
||||
return reply.status(400).send({
|
||||
error: 'Missing or invalid "path" parameter'
|
||||
});
|
||||
}
|
||||
|
||||
const metadata = this.vfs.getMetadata(path);
|
||||
|
||||
if (metadata === null) {
|
||||
return reply.status(404).send({ error: 'Metadata not found' });
|
||||
}
|
||||
|
||||
return metadata;
|
||||
});
|
||||
|
||||
// Get complete VFS tree
|
||||
this.app.get('/api/fuse/tree', async (request, reply) => {
|
||||
return this.vfs.getTree();
|
||||
});
|
||||
|
||||
// Get all file paths
|
||||
this.app.get('/api/fuse/files', async (request, reply) => {
|
||||
const files = this.vfs.getAllFiles();
|
||||
return { files };
|
||||
});
|
||||
|
||||
// Get all source file paths (for Stremio addon validation)
|
||||
this.app.get('/api/fuse/source-paths', async (request, reply) => {
|
||||
const files = this.vfs.getAllFiles();
|
||||
const sourcePaths: string[] = [];
|
||||
|
||||
for (const file of files) {
|
||||
// Skip virtual metadata files (.meta, .nfo, .xml)
|
||||
if (file.endsWith('.meta') || file.endsWith('.nfo') || file.endsWith('.xml')) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const result = this.vfs.read(file);
|
||||
if (result && result.sourcePath) {
|
||||
sourcePaths.push(result.sourcePath);
|
||||
}
|
||||
}
|
||||
|
||||
return { sourcePaths, count: sourcePaths.length };
|
||||
});
|
||||
|
||||
// Get all directory paths
|
||||
this.app.get('/api/fuse/directories', async (request, reply) => {
|
||||
const directories = this.vfs.getAllDirectories();
|
||||
return { directories };
|
||||
});
|
||||
|
||||
// Refresh VFS
|
||||
this.app.post('/api/fuse/refresh', async (request, reply) => {
|
||||
await this.vfs.refresh();
|
||||
return { status: 'ok' };
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Setup Metrics API routes
|
||||
*/
|
||||
@@ -573,7 +380,7 @@ export class UnifiedAPIServer {
|
||||
// Get snapshot from state manager
|
||||
// Note: VFS-related features have been moved to meta-fuse
|
||||
// totalDone now uses state manager's count of fully processed files
|
||||
const snapshot = this.unifiedStateManager!.getSnapshot(0, this.fastQueueConcurrency, this.backgroundQueueConcurrency) as ExtendedProcessingSnapshot;
|
||||
const snapshot = this.unifiedStateManager!.getSnapshot(this.fastQueueConcurrency, this.backgroundQueueConcurrency) as ExtendedProcessingSnapshot;
|
||||
|
||||
// Add queue status if available
|
||||
if (this.getQueueStatus) {
|
||||
@@ -681,59 +488,6 @@ export class UnifiedAPIServer {
|
||||
return { status: 'ok', message: `${queued} files queued for retry` };
|
||||
});
|
||||
|
||||
// POST /api/processing/stop - Stop all processing
|
||||
// Pauses pipeline queues and closes container plugin gate
|
||||
// Running tasks will complete, but no new tasks will start
|
||||
this.app.post('/api/processing/stop', async (request, reply) => {
|
||||
// Pause the streaming pipeline (stops new files from being processed)
|
||||
if (this.streamingPipeline) {
|
||||
this.streamingPipeline.pause();
|
||||
}
|
||||
|
||||
// Close the container plugin gate (stops new plugin tasks)
|
||||
if (this.containerPluginScheduler) {
|
||||
this.containerPluginScheduler.setGate(false);
|
||||
}
|
||||
|
||||
const gateStatus = this.containerPluginScheduler?.getGateStatus() ?? null;
|
||||
const pipelinePaused = this.streamingPipeline?.isPaused() ?? false;
|
||||
|
||||
console.log('[API] Processing stopped - pipeline paused, gate closed');
|
||||
|
||||
return {
|
||||
status: 'ok',
|
||||
message: 'Processing stopped - pipeline paused, gate closed',
|
||||
gateStatus,
|
||||
pipelinePaused
|
||||
};
|
||||
});
|
||||
|
||||
// POST /api/processing/start - Resume all processing
|
||||
// Resumes pipeline queues and opens container plugin gate
|
||||
this.app.post('/api/processing/start', async (request, reply) => {
|
||||
// Resume the streaming pipeline
|
||||
if (this.streamingPipeline) {
|
||||
this.streamingPipeline.resume();
|
||||
}
|
||||
|
||||
// Open the container plugin gate
|
||||
if (this.containerPluginScheduler) {
|
||||
this.containerPluginScheduler.setGate(true);
|
||||
}
|
||||
|
||||
const gateStatus = this.containerPluginScheduler?.getGateStatus() ?? null;
|
||||
const pipelinePaused = this.streamingPipeline?.isPaused() ?? false;
|
||||
|
||||
console.log('[API] Processing started - pipeline resumed, gate opened');
|
||||
|
||||
return {
|
||||
status: 'ok',
|
||||
message: 'Processing started - pipeline resumed, gate opened',
|
||||
gateStatus,
|
||||
pipelinePaused
|
||||
};
|
||||
});
|
||||
|
||||
// POST /api/processing/wait-empty - Wait for queues to drain
|
||||
// Optional query param: timeout (ms, default 60000)
|
||||
this.app.post<{ Querystring: { timeout?: string } }>('/api/processing/wait-empty', async (request, reply) => {
|
||||
@@ -809,56 +563,12 @@ export class UnifiedAPIServer {
|
||||
}
|
||||
|
||||
/**
|
||||
* Setup Metadata Clear API routes
|
||||
* NOTE: /api/scan/trigger has been removed - use meta-core's endpoint instead
|
||||
* Setup Scan-related API routes
|
||||
* NOTE: /api/scan/trigger and /api/metadata/clear have been removed
|
||||
* Use meta-core's endpoints instead
|
||||
*/
|
||||
private setupScanRoutes(): void {
|
||||
// Clear all metadata
|
||||
// NOTE: To trigger a rescan after clearing, call meta-core's /api/scan/trigger endpoint
|
||||
this.app.post('/api/metadata/clear', async (request, reply) => {
|
||||
if (!this.kvClient) {
|
||||
return reply.status(503).send({
|
||||
status: 'error',
|
||||
message: 'KV client not available'
|
||||
});
|
||||
}
|
||||
|
||||
try {
|
||||
console.log('Clearing all metadata via API');
|
||||
|
||||
// Get all file keys and delete them using flat key deletion
|
||||
const hashIds = await this.kvClient.getAllHashIds();
|
||||
let deletedCount = 0;
|
||||
|
||||
for (const hashId of hashIds) {
|
||||
try {
|
||||
// Use deleteMetadataFlat to delete all flat keys for this file
|
||||
await this.kvClient.deleteMetadataFlat(hashId);
|
||||
deletedCount++;
|
||||
} catch (err) {
|
||||
console.warn(`Failed to delete metadata for ${hashId}:`, err);
|
||||
}
|
||||
}
|
||||
|
||||
// Note: deleteMetadataFlat already removes from file:__index__
|
||||
// No need to delete the index separately
|
||||
|
||||
console.log(`Cleared ${deletedCount} file metadata entries`);
|
||||
|
||||
return {
|
||||
status: 'ok',
|
||||
message: `Cleared ${deletedCount} files. To rescan, call meta-core's /api/scan/trigger endpoint.`,
|
||||
deletedCount
|
||||
};
|
||||
} catch (error: any) {
|
||||
console.error('Error clearing metadata:', error);
|
||||
return reply.status(500).send({
|
||||
status: 'error',
|
||||
message: 'Failed to clear metadata',
|
||||
details: error.message
|
||||
});
|
||||
}
|
||||
});
|
||||
// No routes - all scan-related functionality moved to meta-core
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -1785,7 +1495,6 @@ export class UnifiedAPIServer {
|
||||
host: this.config.host
|
||||
});
|
||||
console.log(`Unified API Server listening on http://${this.config.host}:${this.config.port}`);
|
||||
console.log(` - FUSE API: http://${this.config.host}:${this.config.port}/api/fuse/*`);
|
||||
console.log(` - Metrics API: http://${this.config.host}:${this.config.port}/api/metrics`);
|
||||
if (this.kvClient) {
|
||||
console.log(` - Metadata API: http://${this.config.host}:${this.config.port}/api/metadata/* (GET/PUT/DELETE/POST)`);
|
||||
|
||||
@@ -1,630 +0,0 @@
|
||||
/**
|
||||
* VirtualFileSystem - In-memory representation of the virtual output filesystem
|
||||
*
|
||||
* This class maintains the virtual directory structure and provides FUSE-compatible operations.
|
||||
*/
|
||||
|
||||
import { EventEmitter } from 'events';
|
||||
import * as path from 'path';
|
||||
import * as fs from 'fs';
|
||||
import { HashMeta } from '@metazla/meta-interface';
|
||||
import type { ExtendedHashMeta } from '../types/ExtendedInterfaces.js';
|
||||
import {
|
||||
IFuseAPI,
|
||||
VirtualDirectory,
|
||||
VirtualFile,
|
||||
FileAttributes,
|
||||
ReadResult,
|
||||
FuseAPIConfig,
|
||||
FuseAPIEvents
|
||||
} from './FuseAPI.js';
|
||||
import { MetaMeshFormat } from '../logic/MetaMeshFormat.js';
|
||||
import { MetadataFormatRegistry, IMetadataFormatHandler } from '../logic/MetadataFormatRegistry.js';
|
||||
import { config } from '../config/EnvConfig.js';
|
||||
|
||||
interface VFSNode {
|
||||
type: 'file' | 'directory';
|
||||
name: string;
|
||||
parent: string | null;
|
||||
// For files
|
||||
sourcePath?: string | null;
|
||||
size?: number;
|
||||
mtime?: Date;
|
||||
ctime?: Date;
|
||||
metadata?: HashMeta;
|
||||
// For virtual metadata files
|
||||
formatHandler?: IMetadataFormatHandler;
|
||||
// For directories
|
||||
children?: Set<string>;
|
||||
}
|
||||
|
||||
export class VirtualFileSystem extends EventEmitter implements IFuseAPI {
|
||||
private nodes: Map<string, VFSNode> = new Map();
|
||||
private config: Required<FuseAPIConfig>;
|
||||
private metaFormat: MetaMeshFormat = new MetaMeshFormat();
|
||||
private formatRegistry: MetadataFormatRegistry = new MetadataFormatRegistry();
|
||||
private activeFormatHandlers: IMetadataFormatHandler[] = [];
|
||||
|
||||
// Cached stats - updated incrementally on each modification
|
||||
private cachedStats = {
|
||||
fileCount: 0,
|
||||
directoryCount: 0,
|
||||
totalSize: 0,
|
||||
metaFileCount: 0,
|
||||
metadataFileCount: 0
|
||||
};
|
||||
|
||||
constructor(fuseConfig: FuseAPIConfig = {}) {
|
||||
super();
|
||||
this.config = {
|
||||
basePath: fuseConfig.basePath ?? '/',
|
||||
fileMode: fuseConfig.fileMode ?? 0o644,
|
||||
directoryMode: fuseConfig.directoryMode ?? 0o755,
|
||||
uid: fuseConfig.uid ?? process.getuid(),
|
||||
gid: fuseConfig.gid ?? process.getgid()
|
||||
};
|
||||
|
||||
// Initialize active format handlers from configuration
|
||||
this.activeFormatHandlers = this.formatRegistry.getHandlersForFormats(config.METADATA_FORMATS);
|
||||
|
||||
if (this.activeFormatHandlers.length > 0) {
|
||||
console.log(`VirtualFileSystem: Active metadata formats: ${this.activeFormatHandlers.map(h => h.formatType).join(', ')}`);
|
||||
} else {
|
||||
console.log('VirtualFileSystem: No metadata formats configured');
|
||||
}
|
||||
|
||||
// Initialize root directory
|
||||
this.nodes.set('/', {
|
||||
type: 'directory',
|
||||
name: '',
|
||||
parent: null,
|
||||
children: new Set()
|
||||
});
|
||||
|
||||
// Initialize cached stats with root directory
|
||||
this.cachedStats.directoryCount = 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the virtual filesystem from a computed folder structure
|
||||
* @param computed Map of source paths to destination paths
|
||||
* @param metadata Map of source paths to HashMeta
|
||||
* @param basePath Base path to strip from destination paths (e.g., OUTPUT_FOLDER)
|
||||
*/
|
||||
public buildFromComputed(
|
||||
computed: Map<string, string>,
|
||||
metadata: Map<string, HashMeta>,
|
||||
basePath?: string
|
||||
): void {
|
||||
// Clear existing structure (except root)
|
||||
this.nodes.clear();
|
||||
this.nodes.set('/', {
|
||||
type: 'directory',
|
||||
name: '',
|
||||
parent: null,
|
||||
children: new Set()
|
||||
});
|
||||
|
||||
// Reset cached stats (root directory will be counted)
|
||||
this.cachedStats = {
|
||||
fileCount: 0,
|
||||
directoryCount: 1, // Root directory
|
||||
totalSize: 0,
|
||||
metaFileCount: 0,
|
||||
metadataFileCount: 0
|
||||
};
|
||||
|
||||
// Normalize basePath if provided
|
||||
const normalizedBasePath = basePath ? path.normalize(basePath) : null;
|
||||
|
||||
// Add all files and their parent directories
|
||||
for (const [sourcePath, destPath] of computed.entries()) {
|
||||
// Strip basePath from destPath to create relative virtual path
|
||||
let virtualPath = destPath;
|
||||
if (normalizedBasePath && destPath.startsWith(normalizedBasePath)) {
|
||||
virtualPath = destPath.slice(normalizedBasePath.length);
|
||||
// Ensure path starts with /
|
||||
if (!virtualPath.startsWith('/')) {
|
||||
virtualPath = '/' + virtualPath;
|
||||
}
|
||||
}
|
||||
this.addFile(virtualPath, sourcePath, metadata.get(sourcePath));
|
||||
}
|
||||
|
||||
this.emit('vfs-updated');
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a file to the virtual filesystem, creating parent directories as needed
|
||||
*/
|
||||
public addFile(virtualPath: string, sourcePath: string, metadata?: HashMeta): void {
|
||||
// Normalize path
|
||||
virtualPath = this.normalizePath(virtualPath);
|
||||
|
||||
// Create parent directories
|
||||
const dirname = path.dirname(virtualPath);
|
||||
this.ensureDirectory(dirname);
|
||||
|
||||
// Try to use metadata for file stats (avoid disk I/O during fast load)
|
||||
let size: number;
|
||||
let mtime: Date;
|
||||
let ctime: Date;
|
||||
|
||||
const extMeta = metadata as ExtendedHashMeta | undefined;
|
||||
if (extMeta && extMeta.sizeByte) {
|
||||
// Use cached metadata (fast path - no disk I/O)
|
||||
size = parseInt(String(extMeta.sizeByte), 10);
|
||||
// Use current time as placeholder for mtime/ctime (will be corrected on next scan)
|
||||
mtime = new Date();
|
||||
ctime = new Date();
|
||||
} else {
|
||||
// Fallback: stat the file (slow path - disk I/O)
|
||||
try {
|
||||
const stats = fs.statSync(sourcePath);
|
||||
size = stats.size;
|
||||
mtime = stats.mtime;
|
||||
ctime = stats.ctime;
|
||||
} catch (error) {
|
||||
console.warn(`Cannot stat source file: ${sourcePath}`, error);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Add the file node
|
||||
const filename = path.basename(virtualPath);
|
||||
this.nodes.set(virtualPath, {
|
||||
type: 'file',
|
||||
name: filename,
|
||||
parent: dirname,
|
||||
sourcePath,
|
||||
size,
|
||||
mtime,
|
||||
ctime,
|
||||
metadata
|
||||
});
|
||||
|
||||
// Add to parent's children
|
||||
const parentNode = this.nodes.get(dirname);
|
||||
if (parentNode && parentNode.type === 'directory') {
|
||||
parentNode.children!.add(filename);
|
||||
}
|
||||
|
||||
// Update cached stats
|
||||
this.cachedStats.fileCount++;
|
||||
this.cachedStats.totalSize += size;
|
||||
|
||||
// Also add corresponding metadata files based on configured formats
|
||||
this.addMetadataFiles(virtualPath, metadata, mtime);
|
||||
|
||||
this.emit('file-added', virtualPath);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add virtual metadata files for all configured formats
|
||||
*/
|
||||
private addMetadataFiles(mediaPath: string, metadata: HashMeta | undefined, mtime: Date): void {
|
||||
if (!metadata) return;
|
||||
|
||||
// Create a metadata file for each configured format
|
||||
for (const handler of this.activeFormatHandlers) {
|
||||
const metaPath = `${mediaPath}${handler.extension}`;
|
||||
const dirname = path.dirname(metaPath);
|
||||
const filename = path.basename(metaPath);
|
||||
|
||||
// Pre-generate metadata content to calculate size
|
||||
const metaContent = handler.serialize(metadata);
|
||||
const metaBuffer = Buffer.from(metaContent, 'utf-8');
|
||||
|
||||
// Add the virtual metadata file node
|
||||
this.nodes.set(metaPath, {
|
||||
type: 'file',
|
||||
name: filename,
|
||||
parent: dirname,
|
||||
sourcePath: null, // Virtual file, no source
|
||||
size: metaBuffer.length,
|
||||
mtime: mtime,
|
||||
ctime: mtime,
|
||||
metadata,
|
||||
formatHandler: handler
|
||||
});
|
||||
|
||||
// Add to parent's children
|
||||
const parentNode = this.nodes.get(dirname);
|
||||
if (parentNode && parentNode.type === 'directory') {
|
||||
parentNode.children!.add(filename);
|
||||
}
|
||||
|
||||
// Update cached stats
|
||||
this.cachedStats.fileCount++;
|
||||
this.cachedStats.totalSize += metaBuffer.length;
|
||||
this.cachedStats.metadataFileCount++;
|
||||
// Legacy: Count .meta files separately for backward compatibility
|
||||
if (handler.extension === '.meta') {
|
||||
this.cachedStats.metaFileCount++;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure a directory exists in the VFS, creating parent directories as needed
|
||||
*/
|
||||
private ensureDirectory(dirPath: string): void {
|
||||
dirPath = this.normalizePath(dirPath);
|
||||
|
||||
if (dirPath === '/' || this.nodes.has(dirPath)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Recursively create parent
|
||||
const parentPath = path.dirname(dirPath);
|
||||
this.ensureDirectory(parentPath);
|
||||
|
||||
// Create this directory
|
||||
const dirName = path.basename(dirPath);
|
||||
this.nodes.set(dirPath, {
|
||||
type: 'directory',
|
||||
name: dirName,
|
||||
parent: parentPath,
|
||||
children: new Set()
|
||||
});
|
||||
|
||||
// Add to parent's children
|
||||
const parentNode = this.nodes.get(parentPath);
|
||||
if (parentNode && parentNode.type === 'directory') {
|
||||
parentNode.children!.add(dirName);
|
||||
}
|
||||
|
||||
// Update cached stats
|
||||
this.cachedStats.directoryCount++;
|
||||
|
||||
this.emit('directory-added', dirPath);
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalize path to use forward slashes and ensure leading slash
|
||||
*/
|
||||
private normalizePath(p: string): string {
|
||||
// Convert to forward slashes
|
||||
p = p.split(path.sep).join('/');
|
||||
|
||||
// Ensure leading slash
|
||||
if (!p.startsWith('/')) {
|
||||
p = '/' + p;
|
||||
}
|
||||
|
||||
// Remove trailing slash (except for root)
|
||||
if (p !== '/' && p.endsWith('/')) {
|
||||
p = p.slice(0, -1);
|
||||
}
|
||||
|
||||
return p;
|
||||
}
|
||||
|
||||
// IFuseAPI implementation
|
||||
|
||||
public readdir(dirPath: string): string[] | null {
|
||||
dirPath = this.normalizePath(dirPath);
|
||||
const node = this.nodes.get(dirPath);
|
||||
|
||||
if (!node || node.type !== 'directory') {
|
||||
return null;
|
||||
}
|
||||
|
||||
return Array.from(node.children!);
|
||||
}
|
||||
|
||||
public getattr(filepath: string): FileAttributes | null {
|
||||
filepath = this.normalizePath(filepath);
|
||||
const node = this.nodes.get(filepath);
|
||||
|
||||
if (!node) {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (node.type === 'directory') {
|
||||
return {
|
||||
size: 4096, // Standard directory size
|
||||
mode: 0o040000 | this.config.directoryMode, // S_IFDIR | permissions
|
||||
mtime: Date.now() / 1000,
|
||||
atime: Date.now() / 1000,
|
||||
ctime: Date.now() / 1000,
|
||||
nlink: 2, // . and ..
|
||||
uid: this.config.uid,
|
||||
gid: this.config.gid
|
||||
};
|
||||
} else {
|
||||
return {
|
||||
size: node.size!,
|
||||
mode: 0o100000 | this.config.fileMode, // S_IFREG | permissions
|
||||
mtime: node.mtime!.getTime() / 1000,
|
||||
atime: node.mtime!.getTime() / 1000,
|
||||
ctime: node.ctime!.getTime() / 1000,
|
||||
nlink: 1,
|
||||
uid: this.config.uid,
|
||||
gid: this.config.gid
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
public exists(filepath: string): boolean {
|
||||
filepath = this.normalizePath(filepath);
|
||||
return this.nodes.has(filepath);
|
||||
}
|
||||
|
||||
public read(filepath: string): ReadResult | null {
|
||||
filepath = this.normalizePath(filepath);
|
||||
const node = this.nodes.get(filepath);
|
||||
|
||||
if (!node || node.type !== 'file') {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Virtual metadata file (any format)
|
||||
if (node.sourcePath === null && node.metadata && node.formatHandler) {
|
||||
const metaContent = node.formatHandler.serialize(node.metadata);
|
||||
const metaBuffer = Buffer.from(metaContent, 'utf-8');
|
||||
|
||||
return {
|
||||
sourcePath: null,
|
||||
content: metaBuffer,
|
||||
size: metaBuffer.length
|
||||
};
|
||||
}
|
||||
|
||||
// Regular file - return source path for reading
|
||||
return {
|
||||
sourcePath: node.sourcePath!,
|
||||
content: null,
|
||||
size: node.size!
|
||||
};
|
||||
}
|
||||
|
||||
public getMetadata(filepath: string): HashMeta | null {
|
||||
filepath = this.normalizePath(filepath);
|
||||
const node = this.nodes.get(filepath);
|
||||
|
||||
if (!node || node.type !== 'file') {
|
||||
return null;
|
||||
}
|
||||
|
||||
return node.metadata ?? null;
|
||||
}
|
||||
|
||||
public getTree(): VirtualDirectory {
|
||||
const rootNode = this.nodes.get('/')!;
|
||||
|
||||
return {
|
||||
path: '/',
|
||||
type: 'directory',
|
||||
name: '',
|
||||
parent: '',
|
||||
children: Array.from(rootNode.children!),
|
||||
mode: this.config.directoryMode,
|
||||
mtime: new Date()
|
||||
};
|
||||
}
|
||||
|
||||
public async refresh(): Promise<void> {
|
||||
// Refresh is triggered by external calls after processing
|
||||
// The buildFromComputed method handles the actual refresh
|
||||
this.emit('vfs-updated');
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all file paths in the VFS
|
||||
*/
|
||||
public getAllFiles(): string[] {
|
||||
const files: string[] = [];
|
||||
for (const [filepath, node] of this.nodes.entries()) {
|
||||
if (node.type === 'file') {
|
||||
files.push(filepath);
|
||||
}
|
||||
}
|
||||
return files;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all directory paths in the VFS
|
||||
*/
|
||||
public getAllDirectories(): string[] {
|
||||
const dirs: string[] = [];
|
||||
for (const [dirpath, node] of this.nodes.entries()) {
|
||||
if (node.type === 'directory') {
|
||||
dirs.push(dirpath);
|
||||
}
|
||||
}
|
||||
return dirs;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get statistics about the VFS (returns cached values, updated incrementally)
|
||||
*/
|
||||
public getStats(): {
|
||||
fileCount: number;
|
||||
directoryCount: number;
|
||||
totalSize: number;
|
||||
metaFileCount: number;
|
||||
metadataFileCount: number;
|
||||
} {
|
||||
// Return cached stats - no iteration needed!
|
||||
return { ...this.cachedStats };
|
||||
}
|
||||
|
||||
/**
|
||||
* Count files in VFS that have completed hash computation
|
||||
* A file is considered "done" if hashStatus === 'complete' (all hashes computed without errors)
|
||||
*/
|
||||
public countFilesWithHash(): number {
|
||||
let count = 0;
|
||||
for (const [filepath, node] of this.nodes.entries()) {
|
||||
// Only count actual media files (not directories, not metadata files)
|
||||
if (node.type === 'file' && node.sourcePath && node.metadata) {
|
||||
// Check if file has completed processing (both light and hash phases)
|
||||
const extMeta = node.metadata as ExtendedHashMeta;
|
||||
if (extMeta.processingStatus === 'complete') {
|
||||
count++;
|
||||
}
|
||||
}
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
/**
|
||||
* Update metadata for an existing file in the VFS (e.g., when hash computation completes)
|
||||
* This also regenerates the metadata files with updated content
|
||||
*/
|
||||
public updateFileMetadata(virtualPath: string, metadata: HashMeta): void {
|
||||
virtualPath = this.normalizePath(virtualPath);
|
||||
const node = this.nodes.get(virtualPath);
|
||||
|
||||
if (!node || node.type !== 'directory') {
|
||||
// Update file metadata
|
||||
if (node) {
|
||||
node.metadata = metadata;
|
||||
|
||||
// Regenerate metadata files with updated content
|
||||
if (node.mtime) {
|
||||
// Get old metadata file sizes before regenerating
|
||||
const oldSizes = new Map<string, number>();
|
||||
for (const handler of this.activeFormatHandlers) {
|
||||
const metaPath = `${virtualPath}${handler.extension}`;
|
||||
const oldMetaNode = this.nodes.get(metaPath);
|
||||
if (oldMetaNode) {
|
||||
oldSizes.set(metaPath, oldMetaNode.size ?? 0);
|
||||
}
|
||||
}
|
||||
|
||||
// Regenerate metadata files (this will overwrite the nodes)
|
||||
this.addMetadataFilesUpdate(virtualPath, metadata, node.mtime, oldSizes);
|
||||
}
|
||||
|
||||
this.emit('file-updated', virtualPath);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Add/update virtual metadata files for all configured formats (used during updates)
|
||||
* This version adjusts cached stats based on size differences
|
||||
*/
|
||||
private addMetadataFilesUpdate(
|
||||
mediaPath: string,
|
||||
metadata: HashMeta,
|
||||
mtime: Date,
|
||||
oldSizes: Map<string, number>
|
||||
): void {
|
||||
// Create a metadata file for each configured format
|
||||
for (const handler of this.activeFormatHandlers) {
|
||||
const metaPath = `${mediaPath}${handler.extension}`;
|
||||
const dirname = path.dirname(metaPath);
|
||||
const filename = path.basename(metaPath);
|
||||
|
||||
// Pre-generate metadata content to calculate size
|
||||
const metaContent = handler.serialize(metadata);
|
||||
const metaBuffer = Buffer.from(metaContent, 'utf-8');
|
||||
const newSize = metaBuffer.length;
|
||||
|
||||
const oldSize = oldSizes.get(metaPath);
|
||||
const isNew = oldSize === undefined;
|
||||
|
||||
// Update the virtual metadata file node
|
||||
this.nodes.set(metaPath, {
|
||||
type: 'file',
|
||||
name: filename,
|
||||
parent: dirname,
|
||||
sourcePath: null, // Virtual file, no source
|
||||
size: newSize,
|
||||
mtime: mtime,
|
||||
ctime: mtime,
|
||||
metadata,
|
||||
formatHandler: handler
|
||||
});
|
||||
|
||||
// Add to parent's children (no-op if already exists)
|
||||
const parentNode = this.nodes.get(dirname);
|
||||
if (parentNode && parentNode.type === 'directory') {
|
||||
parentNode.children!.add(filename);
|
||||
}
|
||||
|
||||
// Update cached stats
|
||||
if (isNew) {
|
||||
// New metadata file
|
||||
this.cachedStats.fileCount++;
|
||||
this.cachedStats.totalSize += newSize;
|
||||
this.cachedStats.metadataFileCount++;
|
||||
if (handler.extension === '.meta') {
|
||||
this.cachedStats.metaFileCount++;
|
||||
}
|
||||
} else {
|
||||
// Update existing - adjust size difference
|
||||
this.cachedStats.totalSize += (newSize - oldSize);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a file and its associated metadata files from the VFS
|
||||
*/
|
||||
public removeFile(virtualPath: string): void {
|
||||
virtualPath = this.normalizePath(virtualPath);
|
||||
const node = this.nodes.get(virtualPath);
|
||||
|
||||
if (!node || node.type !== 'file') {
|
||||
return;
|
||||
}
|
||||
|
||||
// Update cached stats for main file
|
||||
this.cachedStats.fileCount--;
|
||||
this.cachedStats.totalSize -= node.size ?? 0;
|
||||
|
||||
// Remove the main file
|
||||
this.nodes.delete(virtualPath);
|
||||
|
||||
// Remove from parent's children
|
||||
const parentNode = this.nodes.get(node.parent!);
|
||||
if (parentNode && parentNode.type === 'directory') {
|
||||
parentNode.children!.delete(node.name);
|
||||
}
|
||||
|
||||
// Remove associated metadata files (.nfo, .meta, etc.)
|
||||
for (const handler of this.activeFormatHandlers) {
|
||||
const metaPath = `${virtualPath}${handler.extension}`;
|
||||
const metaNode = this.nodes.get(metaPath);
|
||||
|
||||
if (metaNode) {
|
||||
// Update cached stats for metadata file
|
||||
this.cachedStats.fileCount--;
|
||||
this.cachedStats.totalSize -= metaNode.size ?? 0;
|
||||
this.cachedStats.metadataFileCount--;
|
||||
if (handler.extension === '.meta') {
|
||||
this.cachedStats.metaFileCount--;
|
||||
}
|
||||
|
||||
this.nodes.delete(metaPath);
|
||||
|
||||
// Remove from parent's children
|
||||
const metaParent = this.nodes.get(metaNode.parent!);
|
||||
if (metaParent && metaParent.type === 'directory') {
|
||||
metaParent.children!.delete(metaNode.name);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
this.emit('file-removed', virtualPath);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a file exists in the VFS and return its metadata
|
||||
*/
|
||||
public getFileMetadata(virtualPath: string): HashMeta | undefined {
|
||||
virtualPath = this.normalizePath(virtualPath);
|
||||
const node = this.nodes.get(virtualPath);
|
||||
return node?.metadata;
|
||||
}
|
||||
}
|
||||
|
||||
// Type-safe event emitter
|
||||
export interface VirtualFileSystem {
|
||||
on<K extends keyof FuseAPIEvents>(event: K, listener: FuseAPIEvents[K]): this;
|
||||
emit<K extends keyof FuseAPIEvents>(event: K, ...args: Parameters<FuseAPIEvents[K]>): boolean;
|
||||
}
|
||||
@@ -7,7 +7,6 @@
|
||||
|
||||
import { expect } from 'chai';
|
||||
import { UnifiedAPIServer } from './UnifiedAPIServer.js';
|
||||
import { VirtualFileSystem } from './VirtualFileSystem.js';
|
||||
import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js';
|
||||
import type { IKVClient } from '../kv/IKVClient.js';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
@@ -240,12 +239,8 @@ describe('UnifiedAPIServer', function() {
|
||||
mockStateManager.addDone('/files/test/Test Movie.mp4');
|
||||
mockStateManager.addDiscovered('/files/test/pending.mp4');
|
||||
|
||||
// Create VFS (no params needed - uses defaults)
|
||||
const vfs = new VirtualFileSystem();
|
||||
|
||||
// Create server
|
||||
server = new UnifiedAPIServer(
|
||||
vfs,
|
||||
{ port: 3000, host: 'localhost', enableCors: true },
|
||||
mockStateManager as any,
|
||||
mockKV as any,
|
||||
@@ -276,7 +271,6 @@ describe('UnifiedAPIServer', function() {
|
||||
expect(response.statusCode).to.equal(200);
|
||||
const body = JSON.parse(response.payload);
|
||||
expect(body.status).to.equal('ok');
|
||||
expect(body.services).to.have.property('fuse');
|
||||
expect(body.services).to.have.property('metrics');
|
||||
});
|
||||
|
||||
@@ -416,97 +410,7 @@ describe('UnifiedAPIServer', function() {
|
||||
// NOTE: /api/scan/trigger has been removed from meta-sort (Architecture V3)
|
||||
// Scan is now handled by meta-core. Tests for scan trigger are in core.bats.
|
||||
|
||||
// =========================================================================
|
||||
// FUSE API (Virtual Filesystem)
|
||||
// =========================================================================
|
||||
|
||||
describe('FUSE API (/api/fuse/*)', function() {
|
||||
it('GET /api/fuse/health returns ok', async function() {
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/fuse/health'
|
||||
});
|
||||
|
||||
expect(response.statusCode).to.equal(200);
|
||||
const body = JSON.parse(response.payload);
|
||||
expect(body.status).to.equal('ok');
|
||||
});
|
||||
|
||||
it('GET /api/fuse/stats returns VFS statistics', async function() {
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/fuse/stats'
|
||||
});
|
||||
|
||||
expect(response.statusCode).to.equal(200);
|
||||
const body = JSON.parse(response.payload);
|
||||
expect(body).to.have.property('fileCount');
|
||||
expect(body).to.have.property('directoryCount');
|
||||
});
|
||||
|
||||
it('POST /api/fuse/readdir returns directory contents', async function() {
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/fuse/readdir',
|
||||
payload: { path: '/' }
|
||||
});
|
||||
|
||||
expect(response.statusCode).to.equal(200);
|
||||
const body = JSON.parse(response.payload);
|
||||
expect(body).to.have.property('entries');
|
||||
expect(Array.isArray(body.entries)).to.be.true;
|
||||
});
|
||||
|
||||
it('POST /api/fuse/exists checks path existence', async function() {
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/fuse/exists',
|
||||
payload: { path: '/' }
|
||||
});
|
||||
|
||||
expect(response.statusCode).to.equal(200);
|
||||
const body = JSON.parse(response.payload);
|
||||
expect(body).to.have.property('exists');
|
||||
expect(typeof body.exists).to.equal('boolean');
|
||||
});
|
||||
|
||||
it('GET /api/fuse/tree returns full VFS tree', async function() {
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/fuse/tree'
|
||||
});
|
||||
|
||||
expect(response.statusCode).to.equal(200);
|
||||
const body = JSON.parse(response.payload);
|
||||
// API returns tree structure directly (path, type, children)
|
||||
expect(body).to.have.property('path');
|
||||
expect(body).to.have.property('type');
|
||||
});
|
||||
|
||||
it('GET /api/fuse/files returns all file paths', async function() {
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/fuse/files'
|
||||
});
|
||||
|
||||
expect(response.statusCode).to.equal(200);
|
||||
const body = JSON.parse(response.payload);
|
||||
expect(body).to.have.property('files');
|
||||
expect(Array.isArray(body.files)).to.be.true;
|
||||
});
|
||||
|
||||
it('GET /api/fuse/directories returns all directory paths', async function() {
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/fuse/directories'
|
||||
});
|
||||
|
||||
expect(response.statusCode).to.equal(200);
|
||||
const body = JSON.parse(response.payload);
|
||||
expect(body).to.have.property('directories');
|
||||
expect(Array.isArray(body.directories)).to.be.true;
|
||||
});
|
||||
});
|
||||
// NOTE: FUSE API (/api/fuse/*) has been moved to meta-fuse service.
|
||||
|
||||
// =========================================================================
|
||||
// Plugins API (Plugins Page)
|
||||
|
||||
@@ -1,11 +1,8 @@
|
||||
/**
|
||||
* API exports
|
||||
*
|
||||
* This module exports all API components including FUSE API, Metrics API, and Unified API Server.
|
||||
* This module exports all API components including Metrics API and Unified API Server.
|
||||
*/
|
||||
|
||||
export { IFuseAPI, VirtualNode, VirtualFile, VirtualDirectory, FileAttributes, ReadResult, FuseAPIConfig, FuseAPIEvents } from './FuseAPI.js';
|
||||
export { VirtualFileSystem } from './VirtualFileSystem.js';
|
||||
export { FuseAPIServer, FuseAPIServerConfig } from './FuseAPIServer.js';
|
||||
export { MetricsAPI, MetricsAPIConfig } from './MetricsAPI.js';
|
||||
export { UnifiedAPIServer, UnifiedAPIServerConfig } from './UnifiedAPIServer.js';
|
||||
|
||||
@@ -489,6 +489,9 @@ export class ContainerPluginScheduler extends EventEmitter {
|
||||
|
||||
/**
|
||||
* Check if all tasks for a file are complete
|
||||
*
|
||||
* Note: Multiple files may share the same fileHash (e.g., duplicate files).
|
||||
* When this happens, we need to emit file:complete for EACH unique filePath.
|
||||
*/
|
||||
private checkFileCompletion(fileHash: string): void {
|
||||
const taskIds = this.fileTasks.get(fileHash);
|
||||
@@ -497,7 +500,7 @@ export class ContainerPluginScheduler extends EventEmitter {
|
||||
}
|
||||
|
||||
let allComplete = true;
|
||||
let filePath: string | undefined;
|
||||
const uniqueFilePaths = new Set<string>();
|
||||
|
||||
for (const taskId of taskIds) {
|
||||
const task = this.pendingTasks.get(taskId);
|
||||
@@ -505,7 +508,8 @@ export class ContainerPluginScheduler extends EventEmitter {
|
||||
continue;
|
||||
}
|
||||
|
||||
filePath = task.filePath;
|
||||
// Collect all unique filePaths that share this hash
|
||||
uniqueFilePaths.add(task.filePath);
|
||||
|
||||
if (
|
||||
task.status !== 'completed' &&
|
||||
@@ -517,7 +521,7 @@ export class ContainerPluginScheduler extends EventEmitter {
|
||||
}
|
||||
}
|
||||
|
||||
if (allComplete && filePath) {
|
||||
if (allComplete && uniqueFilePaths.size > 0) {
|
||||
// Clean up task tracking
|
||||
for (const taskId of taskIds) {
|
||||
this.pendingTasks.delete(taskId);
|
||||
@@ -533,9 +537,12 @@ export class ContainerPluginScheduler extends EventEmitter {
|
||||
}
|
||||
}
|
||||
|
||||
// Emit file:complete for EACH unique filePath (handles duplicate files with same hash)
|
||||
for (const filePath of uniqueFilePaths) {
|
||||
this.emit('file:complete', { fileHash, filePath });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch existing metadata for a file
|
||||
|
||||
@@ -65,15 +65,13 @@ const pipeline = new StreamingPipeline({
|
||||
fileProcessor: fileProcessor.fileProcessor,
|
||||
stateManager: fileProcessor.unifiedStateManager,
|
||||
kvClient: fileProcessor.getKVClient(),
|
||||
virtualFileSystem: fileProcessor.getVirtualFileSystem(),
|
||||
metaDataToFolderStruct: fileProcessor.metaDataToFolderStruct
|
||||
});
|
||||
|
||||
// Connect pipeline to fileProcessor for queue status monitoring
|
||||
fileProcessor.setPipeline(pipeline);
|
||||
|
||||
// VFS and state manager are available immediately
|
||||
const vfs = fileProcessor.getVirtualFileSystem();
|
||||
// State manager is available immediately
|
||||
const unifiedStateManager = fileProcessor.getUnifiedStateManager();
|
||||
|
||||
// API Server will be initialized after KV is ready (deferred to async init)
|
||||
@@ -160,7 +158,7 @@ process.on('SIGINT', async () => {
|
||||
|
||||
// Start Unified API Server AFTER KV is ready (so kvClient is available)
|
||||
const kvClient = fileProcessor.getKVClient();
|
||||
apiServer = new UnifiedAPIServer(vfs, {
|
||||
apiServer = new UnifiedAPIServer({
|
||||
port: config.FUSE_API_PORT,
|
||||
host: config.FUSE_API_HOST,
|
||||
enableCors: true,
|
||||
@@ -248,11 +246,6 @@ process.on('SIGINT', async () => {
|
||||
apiServer.setStreamingPipeline(pipeline);
|
||||
console.log('[Startup] StreamingPipeline connected to API server');
|
||||
|
||||
// First, rebuild VFS from KV to restore state from previous runs
|
||||
console.log('[Startup] Rebuilding VirtualFileSystem from KV...');
|
||||
await fileProcessor.rebuildVFSFromKV();
|
||||
console.log('[Startup] VFS rebuild completed');
|
||||
|
||||
// Log cache performance summary
|
||||
const metrics = performanceMetrics.getMetrics();
|
||||
if (metrics.cacheHits.midhash256 || metrics.cacheMisses.midhash256) {
|
||||
|
||||
@@ -51,7 +51,6 @@ export interface UnifiedProcessingSnapshot {
|
||||
totalLightProcessing: number;
|
||||
totalHashProcessing: number;
|
||||
totalDone: number;
|
||||
totalFilesInVFS: number; // Total number of files accessible in the Virtual File System
|
||||
|
||||
/** Files that completed validation but are waiting for fast queue */
|
||||
awaitingFastQueue: number;
|
||||
@@ -217,11 +216,10 @@ export class UnifiedProcessingStateManager {
|
||||
|
||||
/**
|
||||
* Get a snapshot of the current processing state
|
||||
* @param vfsFileCount - Optional total number of files in the Virtual File System
|
||||
* @param fastQueueConcurrency - Number of fast queue workers
|
||||
* @param backgroundQueueConcurrency - Number of background queue workers
|
||||
*/
|
||||
getSnapshot(vfsFileCount: number = 0, fastQueueConcurrency?: number, backgroundQueueConcurrency?: number): UnifiedProcessingSnapshot {
|
||||
getSnapshot(fastQueueConcurrency?: number, backgroundQueueConcurrency?: number): UnifiedProcessingSnapshot {
|
||||
// Limit returned arrays to prevent slow API responses with large queues
|
||||
const MAX_ITEMS_PER_STATE = 100;
|
||||
|
||||
@@ -269,7 +267,6 @@ export class UnifiedProcessingStateManager {
|
||||
totalLightProcessing: this.lightProcessing.size,
|
||||
totalHashProcessing: this.hashProcessing.size,
|
||||
totalDone: this.totalProcessedCount, // Total count of all processed files (not limited to history)
|
||||
totalFilesInVFS: vfsFileCount, // Total number of files accessible in the Virtual File System
|
||||
|
||||
// Files that completed validation but are waiting for fast queue (set by API layer from queue status)
|
||||
awaitingFastQueue: 0,
|
||||
|
||||
@@ -8,11 +8,10 @@ import {FileProcessorPiscina} from "./fileProcessor/FileProcessorPiscina.js";
|
||||
import {renamingRule} from "../config/RenamingRule.js";
|
||||
import {MetaMeshFormat} from "./MetaMeshFormat.js";
|
||||
import {JellyfinAPI} from "../jellyfin/JellyfinAPI.js";
|
||||
import {VirtualFileSystem} from "../api/VirtualFileSystem.js";
|
||||
import {performanceMetrics} from "../metrics/PerformanceMetrics.js";
|
||||
import {UnifiedProcessingStateManager} from "./UnifiedProcessingStateManager.js";
|
||||
import type {IKVClient} from "../kv/IKVClient.js";
|
||||
import { type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js";
|
||||
import { type ExtendedFileAnalyzer } from "../types/ExtendedInterfaces.js";
|
||||
import * as webdav from '../webdav/WebdavClient.js';
|
||||
import PQueue from "p-queue";
|
||||
import os from 'os';
|
||||
@@ -25,7 +24,6 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
jellyfinAPI = new JellyfinAPI();
|
||||
metaMeshFormat = new MetaMeshFormat();
|
||||
metaDataToFolderStruct = new MetaDataToFolderStruct(renamingRule);
|
||||
virtualFileSystem: VirtualFileSystem;
|
||||
unifiedStateManager = new UnifiedProcessingStateManager();
|
||||
fileProcessor:FileAnalyzerInterface = new FileProcessorPiscina();
|
||||
fileType = new FileType();
|
||||
@@ -41,12 +39,6 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
private pipeline: StreamingPipeline | null = null;
|
||||
|
||||
constructor() {
|
||||
// Always initialize VirtualFileSystem for FUSE API
|
||||
this.virtualFileSystem = new VirtualFileSystem({
|
||||
fileMode: 0o644,
|
||||
directoryMode: config.CHMOD_FOR_NEW_FOLDER || 0o755
|
||||
});
|
||||
|
||||
// Pass the unified state manager to the fileProcessor
|
||||
if (this.fileProcessor instanceof FileProcessorPiscina) {
|
||||
this.fileProcessor.setUnifiedStateManager(this.unifiedStateManager);
|
||||
@@ -62,10 +54,6 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
console.log(` - TaskScheduler handles plugin execution (fast/background queues)`);
|
||||
}
|
||||
|
||||
getVirtualFileSystem(): VirtualFileSystem {
|
||||
return this.virtualFileSystem;
|
||||
}
|
||||
|
||||
getUnifiedStateManager(): UnifiedProcessingStateManager {
|
||||
return this.unifiedStateManager;
|
||||
}
|
||||
@@ -193,7 +181,7 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
fastQueueConcurrency: config.FAST_QUEUE_CONCURRENCY,
|
||||
});
|
||||
|
||||
// Set up event listeners for VFS updates
|
||||
// Set up event listeners for state manager updates
|
||||
this.setupTaskSchedulerEvents();
|
||||
|
||||
console.log(`[WatchedFileProcessor] TaskScheduler initialized with:`);
|
||||
@@ -201,30 +189,17 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
}
|
||||
|
||||
/**
|
||||
* Set up TaskScheduler event listeners for VFS integration
|
||||
* Set up TaskScheduler event listeners
|
||||
* Note: Event publishing to meta-fuse is handled by meta-core (file:events stream)
|
||||
*/
|
||||
private setupTaskSchedulerEvents(): void {
|
||||
if (!this.taskScheduler) return;
|
||||
|
||||
// When a task completes, check if we should update VFS
|
||||
this.taskScheduler.on('task:complete', (event: any) => {
|
||||
const { task } = event;
|
||||
|
||||
// Update VFS when filename-parser completes (has basic metadata)
|
||||
if (task.pluginId === 'filename-parser') {
|
||||
this.updateVFSForFile(task.fileHash, task.filePath);
|
||||
}
|
||||
});
|
||||
|
||||
// When all tasks for a file complete
|
||||
this.taskScheduler.on('file:complete', (event: any) => {
|
||||
const { fileHash, filePath } = event;
|
||||
console.log(`[TaskScheduler] All plugins complete for ${filePath}`);
|
||||
|
||||
// Final VFS update with complete metadata
|
||||
this.updateVFSForFile(fileHash, filePath);
|
||||
|
||||
// Update unified state manager
|
||||
if (this.unifiedStateManager) {
|
||||
this.unifiedStateManager.completeHashProcessing(filePath, fileHash);
|
||||
@@ -232,21 +207,6 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Update VFS for a file with current metadata
|
||||
*/
|
||||
private updateVFSForFile(fileHash: string, filePath: string): void {
|
||||
const metadata = this.fileProcessor.getDatabase().get(filePath);
|
||||
if (metadata && metadata.cid_midhash256) {
|
||||
try {
|
||||
const virtualPath = this.metaDataToFolderStruct.renamingRule(metadata as any, '');
|
||||
this.virtualFileSystem.addFile(virtualPath, filePath, metadata);
|
||||
} catch (e) {
|
||||
// Renaming rule may fail for incomplete metadata
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
markDiscovered(filePath: string): void {
|
||||
// STATE 1: Mark as discovered when file is found (before queue)
|
||||
this.unifiedStateManager.addDiscovered(filePath);
|
||||
@@ -274,10 +234,32 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
// STEP 1: Compute midhash256 (file identifier)
|
||||
// Use WebDAV for file stats and hash computation
|
||||
const fileStats = await webdav.stat(filePath);
|
||||
if (!fileStats || !fileStats.exists) {
|
||||
console.error(`[WatchedFileProcessor] File not found via WebDAV: ${filePath}`);
|
||||
if (!fileStats.exists) {
|
||||
// Generate specific error message based on error type
|
||||
const errorType = fileStats.error || 'unknown';
|
||||
let errorMsg: string;
|
||||
switch (errorType) {
|
||||
case 'timeout':
|
||||
errorMsg = 'WebDAV request timeout';
|
||||
break;
|
||||
case 'not_configured':
|
||||
errorMsg = 'WebDAV client not configured';
|
||||
break;
|
||||
case 'not_found':
|
||||
errorMsg = 'File not found';
|
||||
break;
|
||||
case 'network_error':
|
||||
errorMsg = 'WebDAV network error';
|
||||
break;
|
||||
case 'http_error':
|
||||
errorMsg = `WebDAV HTTP error (${fileStats.statusCode})`;
|
||||
break;
|
||||
default:
|
||||
errorMsg = 'WebDAV error';
|
||||
}
|
||||
console.error(`[WatchedFileProcessor] ${errorMsg}: ${filePath}`);
|
||||
if (this.unifiedStateManager) {
|
||||
this.unifiedStateManager.completeLightProcessing(filePath, '', undefined, 'File not found');
|
||||
this.unifiedStateManager.completeLightProcessing(filePath, '', undefined, errorMsg);
|
||||
}
|
||||
return;
|
||||
}
|
||||
@@ -370,7 +352,7 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
|
||||
async finalize(): Promise<void> {
|
||||
try {
|
||||
console.log(`Virtual filesystem update start`);
|
||||
console.log(`Finalize processing start`);
|
||||
const start = performance.now();
|
||||
if (this.fileProcessor.getDatabase().size === 0) {
|
||||
console.log(`No files to process`);
|
||||
@@ -378,29 +360,21 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
}
|
||||
|
||||
// NOTE: Duplicate detection has been moved to meta-dup service
|
||||
// NOTE: VFS building has been moved to meta-fuse service
|
||||
|
||||
let startTime = performance.now();
|
||||
const startTime = performance.now();
|
||||
console.log(`Generating virtual structure for ${this.fileProcessor.getDatabase().size} files`);
|
||||
const virtualStructure = this.metaDataToFolderStruct.generateVirtualStructure(this.fileProcessor.getDatabase());
|
||||
this.metaDataToFolderStruct.generateVirtualStructure(this.fileProcessor.getDatabase());
|
||||
const virtualStructureTime = Math.ceil(performance.now() - startTime);
|
||||
console.log(`Virtual structure generation took ${virtualStructureTime}ms`);
|
||||
performanceMetrics.recordVirtualStructureGeneration(virtualStructureTime);
|
||||
|
||||
// Build VirtualFileSystem for FUSE/WebDAV API
|
||||
startTime = performance.now();
|
||||
this.virtualFileSystem.buildFromComputed(virtualStructure, this.fileProcessor.getDatabase());
|
||||
const vfsBuildTime = Math.ceil(performance.now() - startTime);
|
||||
console.log(`VirtualFileSystem build took ${vfsBuildTime}ms`);
|
||||
performanceMetrics.recordVFSUpdate(vfsBuildTime);
|
||||
const stats = this.virtualFileSystem.getStats();
|
||||
console.log(`VirtualFileSystem: ${stats.fileCount} files (${stats.metaFileCount} .meta), ${stats.directoryCount} directories, ${(stats.totalSize / 1024 / 1024).toFixed(2)} MB`);
|
||||
|
||||
if(this.jellyfinAPI.jellyApiAvailable()) {
|
||||
await this.jellyfinAPI.refreshLibrary();
|
||||
console.log(`Jellyfin refreshed`);
|
||||
}
|
||||
|
||||
console.log(`Virtual filesystem update took ${Math.ceil(performance.now() - start)}ms`);
|
||||
console.log(`Finalize processing took ${Math.ceil(performance.now() - start)}ms`);
|
||||
} catch (error) {
|
||||
console.error(error);
|
||||
}
|
||||
@@ -472,12 +446,12 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
const elapsed = Math.ceil(performance.now() - startTime);
|
||||
console.log(`[Cleanup] Removed ${removedCount} stale entries in ${elapsed}ms`);
|
||||
|
||||
// If we removed files, rebuild VFS to update structure
|
||||
// If we removed files, finalize processing and notify Stremio
|
||||
if (removedCount > 0) {
|
||||
console.log('[Cleanup] Rebuilding VFS after cleanup...');
|
||||
console.log('[Cleanup] Finalizing after cleanup...');
|
||||
await this.finalize();
|
||||
|
||||
// Notify Stremio addon to remove videos (incremental updates like FUSE)
|
||||
// Notify Stremio addon to remove videos
|
||||
if (removedHashIds.length > 0) {
|
||||
console.log(`[Cleanup] Notifying Stremio to remove ${removedHashIds.length} videos...`);
|
||||
for (const hashId of removedHashIds) {
|
||||
@@ -490,78 +464,7 @@ export class WatchedFileProcessor implements FileProcessorInterface {
|
||||
}
|
||||
|
||||
/**
|
||||
* Rebuild VirtualFileSystem from KV data
|
||||
* This is called on startup to restore the VFS from persistent storage
|
||||
*
|
||||
* Fast-load strategy: Load all metadata from KV WITHOUT disk validation.
|
||||
* Discovery will later validate file existence and clean up stale entries.
|
||||
*
|
||||
* Design Philosophy:
|
||||
* - KV = distributed metadata archive (permanent record)
|
||||
* - In-memory DB = current VFS state (only present files)
|
||||
* - Discovery = source of truth (validates and cleans)
|
||||
*/
|
||||
async rebuildVFSFromKV(): Promise<void> {
|
||||
const kvClient = this.getKVClient();
|
||||
if (!kvClient) {
|
||||
console.warn('[VFS Rebuild] KV client not available, skipping VFS rebuild');
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
console.log('[VFS Rebuild] Starting fast-load from KV...');
|
||||
const startTime = performance.now();
|
||||
|
||||
// Get all hash IDs from KV
|
||||
const hashIds = await kvClient.getAllHashIds();
|
||||
console.log(`[VFS Rebuild] Found ${hashIds.length} files in KV`);
|
||||
|
||||
if (hashIds.length === 0) {
|
||||
console.log('[VFS Rebuild] No files in KV, VFS remains empty');
|
||||
return;
|
||||
}
|
||||
|
||||
// Batch fetch all metadata in parallel (no disk validation)
|
||||
console.log('[VFS Rebuild] Fetching metadata in parallel...');
|
||||
const metadataList = await Promise.all(
|
||||
hashIds.map(async (hashId) => {
|
||||
try {
|
||||
return await kvClient.getMetadataFlat(hashId);
|
||||
} catch (error) {
|
||||
console.error(`[VFS Rebuild] Failed to load metadata for ${hashId}:`, error);
|
||||
return null;
|
||||
}
|
||||
})
|
||||
);
|
||||
|
||||
// Populate in-memory database without disk validation
|
||||
let loadedCount = 0;
|
||||
for (const metadata of metadataList) {
|
||||
if (metadata && metadata.filePath) {
|
||||
// Mark as unverified - discovery will validate later
|
||||
(metadata as ExtendedHashMeta)._lastVerified = 0;
|
||||
this.fileProcessor.getDatabase().set(metadata.filePath, metadata);
|
||||
loadedCount++;
|
||||
}
|
||||
}
|
||||
|
||||
console.log(`[VFS Rebuild] Loaded ${loadedCount}/${hashIds.length} files into memory (unverified)`);
|
||||
|
||||
// Build VFS immediately - may contain stale entries until discovery validates
|
||||
if (this.fileProcessor.getDatabase().size > 0) {
|
||||
await this.finalize();
|
||||
console.log(`[VFS Rebuild] VFS rebuilt in ${Math.ceil(performance.now() - startTime)}ms`);
|
||||
console.log('[VFS Rebuild] Discovery will validate file existence in background');
|
||||
} else {
|
||||
console.log('[VFS Rebuild] No files loaded, VFS remains empty');
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[VFS Rebuild] Error rebuilding VFS from KV:', error);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Notify Stremio addon to remove video (incremental update like FUSE)
|
||||
* Notify Stremio addon to remove video
|
||||
*/
|
||||
private async notifyStremioRemove(hashId: string): Promise<void> {
|
||||
try {
|
||||
|
||||
@@ -161,8 +161,30 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{
|
||||
}
|
||||
|
||||
const stats = await webdav.stat(filePath);
|
||||
if (!stats || !stats.exists) {
|
||||
throw new Error(`File not found via WebDAV: ${filePath}`);
|
||||
if (!stats.exists) {
|
||||
// Generate specific error message based on error type
|
||||
const errorType = stats.error || 'unknown';
|
||||
let errorMsg: string;
|
||||
switch (errorType) {
|
||||
case 'timeout':
|
||||
errorMsg = `WebDAV request timeout for: ${filePath}`;
|
||||
break;
|
||||
case 'not_configured':
|
||||
errorMsg = `WebDAV client not configured, cannot access: ${filePath}`;
|
||||
break;
|
||||
case 'not_found':
|
||||
errorMsg = `File not found via WebDAV: ${filePath}`;
|
||||
break;
|
||||
case 'network_error':
|
||||
errorMsg = `WebDAV network error for: ${filePath}`;
|
||||
break;
|
||||
case 'http_error':
|
||||
errorMsg = `WebDAV HTTP error (${stats.statusCode}) for: ${filePath}`;
|
||||
break;
|
||||
default:
|
||||
errorMsg = `WebDAV error for: ${filePath}`;
|
||||
}
|
||||
throw new Error(errorMsg);
|
||||
}
|
||||
|
||||
// STEP 1: Check cache for midhash256 FIRST (avoids network I/O on unchanged files)
|
||||
@@ -348,23 +370,46 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{
|
||||
this.unifiedStateManager.startLightProcessing(filePath);
|
||||
}
|
||||
|
||||
const stats = await webdav.stat(filePath);
|
||||
if (!stats || !stats.exists) {
|
||||
throw new Error(`File not found via WebDAV: ${filePath}`);
|
||||
}
|
||||
|
||||
// STEP 1: Use midhash256 from meta-core if provided, otherwise check cache/compute locally
|
||||
let midHash256: string | undefined;
|
||||
|
||||
if (providedMidHash256) {
|
||||
// midhash256 provided by meta-core - use directly (no file I/O needed)
|
||||
// midhash256 provided by meta-core - use directly (no WebDAV I/O needed!)
|
||||
// meta-core already verified the file exists when it computed the hash
|
||||
midHash256 = providedMidHash256;
|
||||
performanceMetrics.recordCacheHit('midhash256');
|
||||
|
||||
// Store in local cache for future lookups
|
||||
this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 });
|
||||
// Note: We skip cache storage here since we don't have file stats (size/mtime)
|
||||
// This is fine because meta-core will provide the hash again on rescan
|
||||
} else {
|
||||
// Fallback: Check cache or compute locally (legacy path)
|
||||
// Fallback: Need to access file via WebDAV (legacy path without meta-core events)
|
||||
const stats = await webdav.stat(filePath);
|
||||
if (!stats.exists) {
|
||||
// Generate specific error message based on error type
|
||||
const errorType = stats.error || 'unknown';
|
||||
let errorMsg: string;
|
||||
switch (errorType) {
|
||||
case 'timeout':
|
||||
errorMsg = `WebDAV request timeout for: ${filePath}`;
|
||||
break;
|
||||
case 'not_configured':
|
||||
errorMsg = `WebDAV client not configured, cannot access: ${filePath}`;
|
||||
break;
|
||||
case 'not_found':
|
||||
errorMsg = `File not found via WebDAV: ${filePath}`;
|
||||
break;
|
||||
case 'network_error':
|
||||
errorMsg = `WebDAV network error for: ${filePath}`;
|
||||
break;
|
||||
case 'http_error':
|
||||
errorMsg = `WebDAV HTTP error (${stats.statusCode}) for: ${filePath}`;
|
||||
break;
|
||||
default:
|
||||
errorMsg = `WebDAV error for: ${filePath}`;
|
||||
}
|
||||
throw new Error(errorMsg);
|
||||
}
|
||||
|
||||
// Check cache or compute locally
|
||||
const indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString());
|
||||
|
||||
if (indexLine && indexLine.cid_midhash256) {
|
||||
@@ -475,10 +520,8 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{
|
||||
this.unifiedStateManager.startHashProcessing(filePath);
|
||||
}
|
||||
|
||||
const stats = await webdav.stat(filePath);
|
||||
if (!stats || !stats.exists) {
|
||||
throw new Error(`File not found via WebDAV: ${filePath}`);
|
||||
}
|
||||
// Note: We skip WebDAV stat here - file existence was verified during light phase
|
||||
// If file was deleted, the container plugin will fail when accessing it
|
||||
const metadata = this.database.get(filePath);
|
||||
if (!metadata) {
|
||||
throw new Error('Metadata not found - light processing may have failed');
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import {FileAnalyzerInterface} from "../fileProcessor/FileAnalyzerInterface.js";
|
||||
import {UnifiedProcessingStateManager} from "../UnifiedProcessingStateManager.js";
|
||||
import type {IKVClient} from "../../kv/IKVClient.js";
|
||||
import {VirtualFileSystem} from "../../api/VirtualFileSystem.js";
|
||||
import {MetaDataToFolderStruct} from "../MetaDataToFolderStruct.js";
|
||||
import type {ContainerPluginScheduler} from "../../container-plugins/ContainerPluginScheduler.js";
|
||||
|
||||
@@ -53,11 +52,6 @@ export interface PipelineConfig {
|
||||
*/
|
||||
kvClient?: IKVClient;
|
||||
|
||||
/**
|
||||
* Virtual filesystem (for instant file visibility)
|
||||
*/
|
||||
virtualFileSystem: VirtualFileSystem;
|
||||
|
||||
/**
|
||||
* Metadata to folder structure converter (for virtual path generation)
|
||||
*/
|
||||
|
||||
@@ -162,26 +162,13 @@ export class StreamingPipeline {
|
||||
|
||||
await this.config.fileProcessor.processLightPhase(filePath, current, queueSize, midhash256);
|
||||
|
||||
// After light processing, add to VFS
|
||||
// After light processing, dispatch to plugins and queue background tasks
|
||||
const metadata = this.config.fileProcessor.getDatabase().get(filePath);
|
||||
if (metadata && metadata.cid_midhash256) {
|
||||
// Generate virtual path from metadata
|
||||
const virtualPath = this.config.metaDataToFolderStruct.renamingRule(metadata as any, filePath);
|
||||
|
||||
if (virtualPath) {
|
||||
// Add to VFS immediately (file appears with permanent midhash256 ID)
|
||||
this.config.virtualFileSystem.addFile(virtualPath, filePath, metadata);
|
||||
|
||||
// Notify Stremio addon to add video (incremental update like FUSE)
|
||||
if (metadata.cid_midhash256) {
|
||||
// Notify Stremio addon to add video
|
||||
this.notifyStremioAdd(metadata.cid_midhash256).catch(err => {
|
||||
// Silent fail - Stremio addon may not be running
|
||||
});
|
||||
}
|
||||
} else {
|
||||
// File cannot be added to VFS (e.g., unsupported type, missing title)
|
||||
console.log(`⚠ File skipped from VFS (no virtual path): ${filePath}`);
|
||||
}
|
||||
|
||||
// Dispatch tasks to container plugins (fire and forget)
|
||||
this.dispatchContainerPluginTasks(filePath, metadata.cid_midhash256, metadata);
|
||||
|
||||
@@ -59,12 +59,6 @@ export interface PerformanceMetricsData {
|
||||
// Per-plugin timing metrics
|
||||
pluginProcessingTimes: Record<string, PluginTimingStats>;
|
||||
|
||||
// VFS update metrics
|
||||
totalVFSUpdates: number;
|
||||
averageVFSUpdateTime: number;
|
||||
lastVFSUpdateTime: number;
|
||||
recentVFSUpdateTimes: MetricEntry[];
|
||||
|
||||
// Duplicate detection metrics
|
||||
averageDuplicateDetectionTime: number;
|
||||
lastDuplicateDetectionTime: number;
|
||||
@@ -121,12 +115,6 @@ export class PerformanceMetrics {
|
||||
private hashProcessingMin: number = Infinity;
|
||||
private hashProcessingMax: number = 0;
|
||||
|
||||
// VFS update tracking - running statistics
|
||||
private recentVFSUpdates: MetricEntry[] = [];
|
||||
private totalVFSUpdates: number = 0;
|
||||
private vfsUpdateSum: number = 0;
|
||||
private lastVFSUpdateTime: number = 0;
|
||||
|
||||
// Duplicate detection tracking - running statistics
|
||||
private totalDuplicatesFound: number = 0;
|
||||
private duplicateDetectionSum: number = 0;
|
||||
@@ -222,28 +210,6 @@ export class PerformanceMetrics {
|
||||
this.lastActivityTime = Date.now();
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a VFS update event
|
||||
*/
|
||||
recordVFSUpdate(timeMs: number): void {
|
||||
// Update running statistics
|
||||
this.vfsUpdateSum += timeMs;
|
||||
this.lastVFSUpdateTime = timeMs;
|
||||
this.totalVFSUpdates++;
|
||||
|
||||
// Keep recent list for display only
|
||||
this.recentVFSUpdates.push({
|
||||
timestamp: Date.now(),
|
||||
value: timeMs
|
||||
});
|
||||
|
||||
if (this.recentVFSUpdates.length > this.maxRecentEntries) {
|
||||
this.recentVFSUpdates.shift();
|
||||
}
|
||||
|
||||
this.lastActivityTime = Date.now();
|
||||
}
|
||||
|
||||
/**
|
||||
* Record a duplicate detection event
|
||||
*/
|
||||
@@ -447,14 +413,6 @@ export class PerformanceMetrics {
|
||||
return acc;
|
||||
}, {} as Record<string, PluginTimingStats>),
|
||||
|
||||
// VFS update metrics - O(1) from running statistics
|
||||
totalVFSUpdates: this.totalVFSUpdates,
|
||||
averageVFSUpdateTime: this.totalVFSUpdates > 0
|
||||
? this.vfsUpdateSum / this.totalVFSUpdates
|
||||
: 0,
|
||||
lastVFSUpdateTime: this.lastVFSUpdateTime,
|
||||
recentVFSUpdateTimes: this.recentVFSUpdates.slice(-10), // Last 10 updates
|
||||
|
||||
// Duplicate detection metrics - O(1) from running statistics
|
||||
averageDuplicateDetectionTime: this.duplicateDetectionCount > 0
|
||||
? this.duplicateDetectionSum / this.duplicateDetectionCount
|
||||
@@ -541,12 +499,6 @@ export class PerformanceMetrics {
|
||||
this.hashProcessingMin = Infinity;
|
||||
this.hashProcessingMax = 0;
|
||||
|
||||
// Reset VFS update metrics
|
||||
this.recentVFSUpdates = [];
|
||||
this.totalVFSUpdates = 0;
|
||||
this.vfsUpdateSum = 0;
|
||||
this.lastVFSUpdateTime = 0;
|
||||
|
||||
// Reset duplicate detection metrics
|
||||
this.totalDuplicatesFound = 0;
|
||||
this.duplicateDetectionSum = 0;
|
||||
|
||||
@@ -91,9 +91,6 @@ export interface ExtendedHashMeta extends HashMeta {
|
||||
|
||||
/** Source file path */
|
||||
filePath?: string;
|
||||
|
||||
/** Last verification timestamp (0 = unverified, used during VFS rebuild) */
|
||||
_lastVerified?: number;
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
|
||||
@@ -75,6 +75,11 @@ export function toWebdavUrl(filePath: string): string | null {
|
||||
return WEBDAV_URL + encodedPath;
|
||||
}
|
||||
|
||||
/**
|
||||
* Error types for WebDAV operations
|
||||
*/
|
||||
export type WebdavErrorType = 'not_found' | 'timeout' | 'not_configured' | 'network_error' | 'http_error';
|
||||
|
||||
/**
|
||||
* File stats returned from WebDAV HEAD request.
|
||||
*/
|
||||
@@ -82,19 +87,23 @@ export interface WebdavFileStats {
|
||||
size: number;
|
||||
mtime: Date;
|
||||
exists: boolean;
|
||||
/** Error type if the request failed */
|
||||
error?: WebdavErrorType;
|
||||
/** HTTP status code if applicable */
|
||||
statusCode?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get file stats via HTTP HEAD request to WebDAV.
|
||||
*
|
||||
* @param filePath - File path
|
||||
* @returns File stats or null on error
|
||||
* @returns File stats with error information if failed
|
||||
*/
|
||||
export async function stat(filePath: string): Promise<WebdavFileStats | null> {
|
||||
export async function stat(filePath: string): Promise<WebdavFileStats> {
|
||||
const url = toWebdavUrl(filePath);
|
||||
if (!url) {
|
||||
console.error('[webdav-client] Not configured, cannot stat file');
|
||||
return null;
|
||||
return { size: 0, mtime: new Date(0), exists: false, error: 'not_configured' };
|
||||
}
|
||||
|
||||
try {
|
||||
@@ -104,12 +113,12 @@ export async function stat(filePath: string): Promise<WebdavFileStats | null> {
|
||||
});
|
||||
|
||||
if (response.status === 404) {
|
||||
return { size: 0, mtime: new Date(0), exists: false };
|
||||
return { size: 0, mtime: new Date(0), exists: false, error: 'not_found', statusCode: 404 };
|
||||
}
|
||||
|
||||
if (!response.ok) {
|
||||
console.error(`[webdav-client] HEAD error for ${url}: ${response.status}`);
|
||||
return null;
|
||||
return { size: 0, mtime: new Date(0), exists: false, error: 'http_error', statusCode: response.status };
|
||||
}
|
||||
|
||||
const contentLength = response.headers.get('Content-Length');
|
||||
@@ -120,9 +129,20 @@ export async function stat(filePath: string): Promise<WebdavFileStats | null> {
|
||||
mtime: lastModified ? new Date(lastModified) : new Date(),
|
||||
exists: true
|
||||
};
|
||||
} catch (error) {
|
||||
} catch (error: any) {
|
||||
// Distinguish timeout errors from other network errors
|
||||
const isTimeout = error?.name === 'TimeoutError' ||
|
||||
error?.code === 'ABORT_ERR' ||
|
||||
error?.message?.includes('timeout') ||
|
||||
error?.message?.includes('aborted');
|
||||
|
||||
if (isTimeout) {
|
||||
console.error(`[webdav-client] HEAD timeout for ${url}`);
|
||||
return { size: 0, mtime: new Date(0), exists: false, error: 'timeout' };
|
||||
}
|
||||
|
||||
console.error(`[webdav-client] HEAD error for ${url}:`, error);
|
||||
return null;
|
||||
return { size: 0, mtime: new Date(0), exists: false, error: 'network_error' };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -134,7 +154,7 @@ export async function stat(filePath: string): Promise<WebdavFileStats | null> {
|
||||
*/
|
||||
export async function exists(filePath: string): Promise<boolean> {
|
||||
const stats = await stat(filePath);
|
||||
return stats?.exists ?? false;
|
||||
return stats.exists;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -54,33 +54,6 @@ function Monitor() {
|
||||
return () => clearInterval(interval);
|
||||
}, [fetchData]);
|
||||
|
||||
const triggerScan = async () => {
|
||||
try {
|
||||
await fetch('/api/scan/trigger', { method: 'POST' });
|
||||
fetchData();
|
||||
} catch (err) {
|
||||
console.error('Failed to trigger scan:', err);
|
||||
}
|
||||
};
|
||||
|
||||
const stopProcessing = async () => {
|
||||
try {
|
||||
await fetch('/api/processing/stop', { method: 'POST' });
|
||||
fetchData();
|
||||
} catch (err) {
|
||||
console.error('Failed to stop processing:', err);
|
||||
}
|
||||
};
|
||||
|
||||
const startProcessing = async () => {
|
||||
try {
|
||||
await fetch('/api/processing/start', { method: 'POST' });
|
||||
fetchData();
|
||||
} catch (err) {
|
||||
console.error('Failed to start processing:', err);
|
||||
}
|
||||
};
|
||||
|
||||
const retryFile = async (filePath: string) => {
|
||||
try {
|
||||
await fetch('/api/processing/retry', {
|
||||
@@ -103,25 +76,6 @@ function Monitor() {
|
||||
}
|
||||
};
|
||||
|
||||
const clearMetadata = async () => {
|
||||
if (!window.confirm('Are you sure you want to clear all metadata? This will delete all file metadata and trigger a full rescan.')) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const response = await fetch('/api/metadata/clear', { method: 'POST' });
|
||||
const data = await response.json();
|
||||
if (response.ok) {
|
||||
alert(`Cleared ${data.deletedCount} files. Rescan triggered.`);
|
||||
} else {
|
||||
alert(`Failed to clear metadata: ${data.message}`);
|
||||
}
|
||||
fetchData();
|
||||
} catch (err) {
|
||||
console.error('Failed to clear metadata:', err);
|
||||
alert('Failed to clear metadata');
|
||||
}
|
||||
};
|
||||
|
||||
// Calculate progress for 6-stage pipeline (including failed)
|
||||
// Discovered = files waiting for fast queue (validated but not yet processing)
|
||||
const getProgress = () => {
|
||||
@@ -193,23 +147,6 @@ function Monitor() {
|
||||
<div className="monitor-page">
|
||||
<div className="monitor-header">
|
||||
<h1>Processing Monitor</h1>
|
||||
<div className="header-actions">
|
||||
{status?.computed?.gateOpen === false ? (
|
||||
<button className="btn btn-success" onClick={startProcessing}>
|
||||
Start Processing
|
||||
</button>
|
||||
) : (
|
||||
<button className="btn btn-danger" onClick={stopProcessing}>
|
||||
Stop Processing
|
||||
</button>
|
||||
)}
|
||||
<button className="btn btn-primary" onClick={triggerScan}>
|
||||
Trigger Scan
|
||||
</button>
|
||||
<button className="btn btn-warning" onClick={clearMetadata}>
|
||||
Clear Metadata
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* Pipeline Progress Bar - 6 Stage (including failed) */}
|
||||
@@ -574,30 +511,6 @@ function Monitor() {
|
||||
font-size: 1.8rem;
|
||||
}
|
||||
|
||||
.header-actions {
|
||||
display: flex;
|
||||
gap: 12px;
|
||||
}
|
||||
|
||||
.btn-success {
|
||||
background: linear-gradient(135deg, #10b981, #059669);
|
||||
color: white;
|
||||
}
|
||||
|
||||
.btn-success:hover {
|
||||
background: linear-gradient(135deg, #059669, #047857);
|
||||
}
|
||||
|
||||
.btn-danger {
|
||||
background: linear-gradient(135deg, #ef4444, #dc2626);
|
||||
color: white;
|
||||
}
|
||||
|
||||
.btn-danger:hover {
|
||||
background: linear-gradient(135deg, #dc2626, #b91c1c);
|
||||
}
|
||||
|
||||
|
||||
/* Pipeline Progress Bar */
|
||||
.pipeline-section {
|
||||
margin-bottom: 24px;
|
||||
|
||||
@@ -21,7 +21,6 @@ function Plugins() {
|
||||
const [configEditing, setConfigEditing] = useState<Record<string, unknown> | null>(null);
|
||||
const [loading, setLoading] = useState(false);
|
||||
const [error, setError] = useState<string | null>(null);
|
||||
const [scanStatus, setScanStatus] = useState<string | null>(null);
|
||||
|
||||
// Container plugin management state
|
||||
const [showAddPluginModal, setShowAddPluginModal] = useState(false);
|
||||
@@ -115,18 +114,6 @@ function Plugins() {
|
||||
}
|
||||
};
|
||||
|
||||
const triggerScan = async () => {
|
||||
setScanStatus('Scanning...');
|
||||
try {
|
||||
await fetch('/api/scan/trigger', { method: 'POST' });
|
||||
setScanStatus('Scan triggered');
|
||||
setTimeout(() => setScanStatus(null), 3000);
|
||||
} catch (err) {
|
||||
console.error('Failed to trigger scan:', err);
|
||||
setScanStatus('Scan failed');
|
||||
}
|
||||
};
|
||||
|
||||
const rescanPlugins = async () => {
|
||||
setLoading(true);
|
||||
try {
|
||||
@@ -344,13 +331,6 @@ function Plugins() {
|
||||
>
|
||||
Rescan Plugins
|
||||
</button>
|
||||
<button
|
||||
className="btn btn-primary"
|
||||
onClick={triggerScan}
|
||||
disabled={loading}
|
||||
>
|
||||
{scanStatus || 'Trigger File Scan'}
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
|
||||
Reference in New Issue
Block a user