meta-fuse/docs/streaming-architecture.md

Streaming Architecture

Overview

This document describes the streaming event processing system used by meta-fuse to maintain VFS state. The architecture uses Redis Streams for reliable, ordered event delivery with replay capability.

Core Components

StreamingStateBuilder

File: src/vfs/StreamingStateBuilder.ts

The StreamingStateBuilder is the central component for processing stream events and maintaining file metadata state.

interface StreamingStateBuilderConfig {
    /** Redis client for fetching property values */
    redisClient: RedisClient;
    /** Renaming rules config for extracting VFS-relevant properties */
    rulesConfig: RenamingConfig;
    /** VFS update callback */
    vfsCallback?: VFSUpdateCallback;
    /** FILES_VOLUME path prefix */
    filesPath?: string;
}

Responsibilities:

• Process events from meta:events stream
• Filter events by VFS-relevant properties
• Fetch property values from Redis
• Maintain in-memory state of all files
• Notify VFS when files are complete or change

RulesPropertyExtractor

File: src/vfs/RulesPropertyExtractor.ts

Analyzes renaming rules configuration to determine which metadata properties are needed for VFS path computation.

class RulesPropertyExtractor {
    // Extract all VFS-relevant properties from rules config
    extractVfsProperties(config: RenamingConfig): Set<string>;

    // Get core properties that are always required
    getCoreProperties(): Set<string>;
}

Core Properties (always required):

• filePath - Source file location
• size, fileSize - File size
• mtime, ctime - Timestamps
• fileName, extension - Filename components

Rule-Derived Properties (extracted from templates and conditions):

• Template variables: {title}, {season:pad2}, {episode:pad2}
• Condition fields: fileType, year, mediaType

VFSUpdateCallback

File: src/vfs/StreamingStateBuilder.ts:32-37

Interface for VFS to receive updates from StreamingStateBuilder:

interface VFSUpdateCallback {
    // Called when a property value changes
    onPropertyChange(hashId: string, property: string, value: string): void;

    // Called when a property is deleted
    onPropertyDelete(hashId: string, property: string): void;

    // Called when a file is completely removed (filePath deleted)
    onFileDelete(hashId: string): void;

    // Called when a file becomes complete (has filePath) or updates
    onFileComplete(hashId: string, metadata: Record<string, string>): void;
}

---

Event Processing Pipeline

Stream Message Format

File: src/kv/RedisClient.ts:30-43

interface StreamMessage {
    id: string;           // Stream entry ID (e.g., "1703808000000-0")
    type: 'set' | 'del';  // Operation type
    key?: string;         // Redis key (e.g., "file:abc123/title")
    ts?: string;          // Timestamp
    timestamp: string;    // Fallback timestamp
}

5-Step Event Processing

┌─────────────────────────────────────────────────────────────────┐
│  Step 1: Receive Stream Message                                 │
│  ────────────────────────────────────────────────────────────── │
│  {id: "1703808000-0", type: "set", key: "file:abc123/title"}   │
└───────────────────────────────┬─────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│  Step 2: Parse Key                                              │
│  ────────────────────────────────────────────────────────────── │
│  parseMetaEventKey("file:abc123/title")                        │
│  → { hashId: "abc123", property: "title" }                     │
└───────────────────────────────┬─────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│  Step 3: Check VFS Relevance                                    │
│  ────────────────────────────────────────────────────────────── │
│  isVfsRelevantProperty("title", vfsRelevantProps)              │
│  → true (if "title" is in renaming rules)                      │
│  → false → skip, increment propertiesSkipped                   │
└───────────────────────────────┬─────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│  Step 4: Fetch Property Value                                   │
│  ────────────────────────────────────────────────────────────── │
│  await redisClient.getProperty("abc123", "title")              │
│  → "Inception"                                                  │
└───────────────────────────────┬─────────────────────────────────┘
                                │
                                ▼
┌─────────────────────────────────────────────────────────────────┐
│  Step 5: Update State & Notify VFS                              │
│  ────────────────────────────────────────────────────────────── │
│  updateProperty("abc123", "title", "Inception")                │
│  vfsCallback.onPropertyChange("abc123", "title", "Inception")  │
│                                                                 │
│  If file just became complete (now has filePath):              │
│    vfsCallback.onFileComplete("abc123", metadata)              │
└─────────────────────────────────────────────────────────────────┘

Code Implementation

// StreamingStateBuilder.ts:163-206
private async handlePropertySet(hashId: string, property: string): Promise<void> {
    // Step 3: Check if this property is VFS-relevant
    if (!isVfsRelevantProperty(property, this.vfsRelevantProps)) {
        this.stats.propertiesSkipped++;
        return;
    }

    // Step 4: Fetch the property value from Redis
    const value = await this.redisClient.getProperty(hashId, property);
    if (value === null) {
        return; // Property was set then deleted before we could fetch
    }

    this.stats.propertiesFetched++;

    // Check if file was complete BEFORE updating state
    const wasComplete = this.isFileComplete(hashId);

    // Step 5: Update internal state
    this.updateProperty(hashId, property, value);

    // Notify VFS of changes
    if (this.vfsCallback) {
        this.vfsCallback.onPropertyChange(hashId, property, value);

        // If file just became complete (has filePath now), notify VFS
        if (!wasComplete && this.isFileComplete(hashId)) {
            this.stats.filesComplete++;
            const metadata = this.getFileMetadata(hashId);
            if (metadata) {
                this.vfsCallback.onFileComplete(hashId, metadata);
            }
        }
    }
}

---

Bootstrap vs Live Phases

Bootstrap Phase

When: Application startup

Goal: Rebuild complete VFS state from event history

// index.ts:158-174
// Bootstrap: Replay meta:events stream from position 0
const lastId = await redisClient.replayStream(
    EVENTS_STREAM,
    '0',                    // Start from beginning
    async (message: StreamMessage) => {
        await stateBuilder.processEvent(message);
    },
    100                     // Batch size
);

Characteristics:

• Reads from stream position 0 (all history)
• Non-blocking batch reads (no XREADGROUP)
• Processes all events sequentially
• Returns last processed event ID

Live Phase

When: After bootstrap completes

Goal: Process new events in real-time

// index.ts:177-187
redisClient.startSimpleStreamConsumer(
    EVENTS_STREAM,
    lastId,                 // Resume from bootstrap position
    async (message: StreamMessage) => {
        await stateBuilder.processEvent(message);
    },
    5000                    // 5 second block timeout
);

Characteristics:

• Resumes from last bootstrap position
• Uses blocking reads (XREAD BLOCK)
• Processes events as they arrive
• Runs until shutdown

---

Memory Efficiency

State Structure

type FileState = Map<string, string>;      // property → value
type FilesState = Map<string, FileState>;  // hashId → properties

Memory Per File

Component	Size
hashId (64 char)	~128 bytes
Map overhead	~56 bytes
5-10 properties (avg)	~200-400 bytes
Total per file	~400-600 bytes

Comparison with Full Load

Approach	Memory per File	10K Files
Full HGETALL (all props)	~2-5 KB	20-50 MB
Streaming (VFS props only)	~500 bytes	5 MB
Savings	75-90%	15-45 MB

Why It Works

1. Property Filtering: Only fetch properties used in renaming rules
2. Lazy Loading: Properties fetched on-demand from events
3. No Duplication: Values stored once in state map
4. Garbage Collection: File state removed when file deleted

---

Property Filtering Details

Extracting VFS-Relevant Properties

// RulesPropertyExtractor.ts:132-155
extractVfsProperties(config: RenamingConfig): Set<string> {
    const properties = new Set<string>(CORE_PROPERTIES);

    // Extract from all enabled rules
    for (const rule of config.rules) {
        if (!rule.enabled) continue;

        // From template: {title|originalTitle}, {season:pad2}
        const templateVars = templateEngine.extractVariables(rule.template);
        properties.add(...templateVars);

        // From conditions: { field: 'fileType', value: 'video' }
        const conditionFields = extractConditionFields(rule.conditions);
        properties.add(...conditionFields);
    }

    return properties;
}

Property Path Matching

Handles both exact and nested property matching:

// RulesPropertyExtractor.ts:85-111
function isVfsRelevantProperty(property: string, vfsProperties: Set<string>): boolean {
    const normalized = normalizePropertyPath(property);

    // Direct match
    if (vfsProperties.has(normalized)) {
        return true;
    }

    // Nested property match
    // e.g., if we track 'titles', then 'titles/eng' matches
    for (const vfsProp of vfsProperties) {
        if (normalized.startsWith(vfsProp + '.')) {
            return true;
        }
        if (vfsProp.startsWith(normalized + '.')) {
            return true;
        }
    }

    return false;
}

---

Statistics and Monitoring

StreamingStateBuilder Stats

// StreamingStateBuilder.ts:68-74
private stats = {
    eventsProcessed: 0,     // Total events received
    propertiesFetched: 0,   // Properties fetched from Redis
    propertiesSkipped: 0,   // Non-VFS-relevant properties skipped
    filesComplete: 0,       // Files that became complete
    lastEventId: '0',       // Last processed event ID
};

Accessing Stats

const stats = stateBuilder.getStats();
// Returns:
{
    eventsProcessed: 15234,
    propertiesFetched: 8934,
    propertiesSkipped: 6300,
    filesComplete: 5123,
    lastEventId: "1703808000000-0",
    fileCount: 5123,        // Complete files (have filePath)
    stateSize: 5234         // Total files in state
}

Interpreting Stats

Metric	Meaning
eventsProcessed	Total stream events consumed
propertiesFetched	VFS-relevant property GET calls
propertiesSkipped	Non-relevant events filtered
filesComplete	Files added to VFS
fileCount	Current files in VFS
Fetch ratio	propertiesFetched / eventsProcessed (lower is better)

---

Error Handling

Event Processing Errors

// StreamingStateBuilder.ts:123-136
async processEvent(message: StreamMessage): Promise<void> {
    this.stats.eventsProcessed++;
    this.stats.lastEventId = message.id;

    if ((message.type === 'set' || message.type === 'del') && message.key) {
        await this.processMetaEvent(message);
        return;
    }

    // Ignore legacy event types - not supported in flat key architecture
    logger.debug(`Ignoring legacy event type: ${message.type}`);
}

Property Fetch Failures

// StreamingStateBuilder.ts:171-177
const value = await this.redisClient.getProperty(hashId, property);
if (value === null) {
    // Property was set then deleted before we could fetch it
    // Or it's a different data type - skip silently
    return;
}

Stream Consumer Recovery

// index.ts:185-187
.catch(error => {
    logger.error('Stream consumer error:', error);
});

On error, the consumer loop continues retrying with brief pauses.

---

Configuration

StreamingStateBuilder Configuration

const stateBuilder = new StreamingStateBuilder({
    redisClient,
    rulesConfig: vfs.getRulesConfig(),
    vfsCallback: vfs,
    filesPath: '/files',
});

Dynamic Rules Update

When renaming rules change:

// StreamingStateBuilder.ts:99-102
updateRulesConfig(rulesConfig: RenamingConfig): void {
    this.vfsRelevantProps = this.propertyExtractor.extractVfsProperties(rulesConfig);
    logger.info(`Updated VFS-relevant properties: ${this.vfsRelevantProps.size}`);
}

---

Related Documentation

• VFS Rebuild Architecture - High-level architecture overview
• API Reference - REST API documentation
• Redis Key Format - Flat key storage format