1226 lines
41 KiB
TypeScript
1226 lines
41 KiB
TypeScript
/**
|
|
* Redis Client for meta-fuse
|
|
* Reads metadata from Redis (written by meta-sort)
|
|
*
|
|
* Key features:
|
|
* - FILES_VOLUME path resolution (prepends /files/ to relative paths)
|
|
* - Parses meta-sort hash format into structured metadata
|
|
* - Redis pub/sub support for real-time updates
|
|
*/
|
|
|
|
import { Redis } from 'ioredis';
|
|
import { Logger } from 'tslog';
|
|
import * as os from 'os';
|
|
import type { IKVClient, KeyValuePair } from './IKVClient.js';
|
|
import { MetaCoreApiClient } from './MetaCoreApiClient.js';
|
|
|
|
const logger = new Logger({ name: 'RedisClient' });
|
|
|
|
/**
|
|
* Stream message from Redis Streams
|
|
*
|
|
* New meta:events format (from meta-core flat key architecture):
|
|
* - type: 'set' | 'del' - keyspace notification type
|
|
* - key: 'file:abc123/property' - the Redis key that changed
|
|
* - ts: timestamp
|
|
*
|
|
* Legacy file:events format (deprecated):
|
|
* - type: add, change, delete, rename, batch, reset, plugin:complete
|
|
* - path, size, midhash256, oldPath, payload fields
|
|
*/
|
|
export interface StreamMessage {
|
|
id: string;
|
|
type: 'set' | 'del' | 'add' | 'change' | 'delete' | 'rename' | 'batch' | 'reset' | 'plugin:complete';
|
|
// New meta:events fields (flat key architecture)
|
|
key?: string; // Redis key that changed: file:abc123/property
|
|
ts?: string; // Timestamp
|
|
// Legacy file:events fields (for backward compatibility)
|
|
path?: string;
|
|
size?: string;
|
|
midhash256?: string; // midhash256 CID computed by meta-core
|
|
oldPath?: string;
|
|
payload?: string;
|
|
timestamp: string;
|
|
}
|
|
|
|
/**
|
|
* Parsed meta:events key
|
|
* Extracts hashId and property from keys like 'file:abc123/property'
|
|
*/
|
|
export interface ParsedMetaEventKey {
|
|
hashId: string;
|
|
property: string;
|
|
}
|
|
|
|
/**
|
|
* Stream consumer callback
|
|
*/
|
|
export type StreamMessageHandler = (message: StreamMessage) => Promise<void>;
|
|
|
|
/**
|
|
* Full metadata structure including fields needed for virtual path computation
|
|
*/
|
|
export interface FileMetadata {
|
|
// Path information
|
|
sourcePath: string; // Full path: /files/media1/Movies/...
|
|
originalPath: string; // Relative path from meta-sort: media1/Movies/...
|
|
|
|
// File stats
|
|
size: number;
|
|
mtime: number;
|
|
ctime: number;
|
|
|
|
// Identification
|
|
hashId?: string; // midhash256 hash ID
|
|
|
|
// Title information (for folder organization)
|
|
title?: string;
|
|
originalTitle?: string;
|
|
fileName?: string;
|
|
|
|
// Series metadata
|
|
season?: number;
|
|
episode?: number;
|
|
extra?: boolean;
|
|
|
|
// Movie metadata
|
|
movieYear?: number;
|
|
year?: number;
|
|
|
|
// File type
|
|
fileType?: string; // 'video', 'subtitle', 'torrent', etc.
|
|
extension?: string;
|
|
|
|
// Version/variant info
|
|
version?: string;
|
|
subtitleLanguage?: string;
|
|
|
|
// Generic metadata container
|
|
[key: string]: unknown;
|
|
}
|
|
|
|
export interface RedisClientConfig {
|
|
url?: string;
|
|
prefix?: string;
|
|
filesVolume?: string;
|
|
reconnectInterval?: number;
|
|
|
|
/**
|
|
* meta-core HTTP API base URL. When set, read methods (getProperty,
|
|
* getProperties, getMetadataFlat, getAllHashIds, health) route through
|
|
* the /meta/{hash} HTTP family instead of touching Redis directly.
|
|
*
|
|
* This is the meta-fuse half of the api-mediated-access lockdown (see
|
|
* meta-core docs/api-mediated-access.md). Reads stay on Redis when
|
|
* left unset so tests / one-off scripts still work.
|
|
*/
|
|
metaCoreApiUrl?: string;
|
|
}
|
|
|
|
export class RedisClient implements Partial<IKVClient> {
|
|
private client: Redis | null = null;
|
|
private subscriber: Redis | null = null;
|
|
private config: Required<Omit<RedisClientConfig, 'metaCoreApiUrl'>> & Pick<RedisClientConfig, 'metaCoreApiUrl'>;
|
|
private isConnected = false;
|
|
private reconnectTimer: NodeJS.Timeout | null = null;
|
|
private subscriptions: Map<string, (message: string) => void> = new Map();
|
|
private apiClient: MetaCoreApiClient | null = null;
|
|
|
|
// Stream consumer state
|
|
private streamConsumerRunning = false;
|
|
private streamConsumerAbort: AbortController | null = null;
|
|
private consumerName: string;
|
|
|
|
constructor(config: RedisClientConfig = {}) {
|
|
this.config = {
|
|
url: config.url ?? process.env.REDIS_URL ?? 'redis://127.0.0.1:6379',
|
|
prefix: config.prefix ?? process.env.REDIS_PREFIX ?? '',
|
|
filesVolume: config.filesVolume ?? process.env.FILES_VOLUME ?? '/files',
|
|
reconnectInterval: config.reconnectInterval ?? 5000,
|
|
metaCoreApiUrl: config.metaCoreApiUrl,
|
|
};
|
|
|
|
if (this.config.metaCoreApiUrl) {
|
|
this.apiClient = new MetaCoreApiClient({ apiUrl: this.config.metaCoreApiUrl });
|
|
logger.info(`Reads routed through meta-core API at ${this.config.metaCoreApiUrl}`);
|
|
}
|
|
|
|
// Unique consumer name for this instance
|
|
this.consumerName = `${os.hostname()}-${process.pid}`;
|
|
}
|
|
|
|
/**
|
|
* Connect to Redis. In HTTP-only mode (no URL + apiClient set) this is
|
|
* a no-op — reads route through MetaCoreApiClient and stream
|
|
* consumption is owned by SSEEventClient at the index.ts layer.
|
|
*/
|
|
async connect(): Promise<void> {
|
|
if (this.client && this.isConnected) {
|
|
return;
|
|
}
|
|
|
|
if (this.apiClient && !this.config.url) {
|
|
logger.info('HTTP-only mode — no direct Redis connection');
|
|
return;
|
|
}
|
|
|
|
try {
|
|
this.client = new Redis(this.config.url, {
|
|
retryStrategy: (times: number) => {
|
|
if (times > 10) {
|
|
logger.warn('Max reconnection attempts reached');
|
|
return null;
|
|
}
|
|
return Math.min(times * 500, 5000);
|
|
},
|
|
maxRetriesPerRequest: 3,
|
|
});
|
|
|
|
this.client.on('connect', () => {
|
|
logger.info('Connected to Redis');
|
|
this.isConnected = true;
|
|
});
|
|
|
|
this.client.on('error', (err: Error) => {
|
|
logger.error('Redis error:', err.message);
|
|
this.isConnected = false;
|
|
});
|
|
|
|
this.client.on('close', () => {
|
|
logger.warn('Redis connection closed');
|
|
this.isConnected = false;
|
|
this.scheduleReconnect();
|
|
});
|
|
|
|
// Wait for connection
|
|
await new Promise<void>((resolve, reject) => {
|
|
const timeout = setTimeout(() => {
|
|
reject(new Error('Redis connection timeout'));
|
|
}, 10000);
|
|
|
|
this.client!.once('ready', () => {
|
|
clearTimeout(timeout);
|
|
this.isConnected = true;
|
|
resolve();
|
|
});
|
|
|
|
this.client!.once('error', (err: Error) => {
|
|
clearTimeout(timeout);
|
|
reject(err);
|
|
});
|
|
});
|
|
} catch (error) {
|
|
logger.error('Failed to connect to Redis:', error);
|
|
this.scheduleReconnect();
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Schedule reconnection attempt
|
|
*/
|
|
private scheduleReconnect(): void {
|
|
if (this.reconnectTimer) {
|
|
return;
|
|
}
|
|
|
|
this.reconnectTimer = setTimeout(async () => {
|
|
this.reconnectTimer = null;
|
|
try {
|
|
await this.connect();
|
|
} catch {
|
|
// Will retry via scheduleReconnect
|
|
}
|
|
}, this.config.reconnectInterval);
|
|
}
|
|
|
|
/**
|
|
* Disconnect from Redis
|
|
*/
|
|
async disconnect(): Promise<void> {
|
|
// Stop stream consumer first
|
|
this.stopStreamConsumer();
|
|
|
|
if (this.reconnectTimer) {
|
|
clearTimeout(this.reconnectTimer);
|
|
this.reconnectTimer = null;
|
|
}
|
|
|
|
if (this.subscriber) {
|
|
await this.subscriber.quit();
|
|
this.subscriber = null;
|
|
}
|
|
|
|
if (this.client) {
|
|
await this.client.quit();
|
|
this.client = null;
|
|
this.isConnected = false;
|
|
}
|
|
|
|
this.subscriptions.clear();
|
|
}
|
|
|
|
/**
|
|
* Check if the metadata backend is reachable.
|
|
*
|
|
* In HTTP-only mode (api-mediated-access PR D) there's no Redis socket;
|
|
* reachability is signalled by having an apiClient configured. The
|
|
* health badge is keyed off this; reporting `false` there would make
|
|
* the UI permanently show "Disconnected" even though the service is
|
|
* working fine over HTTP.
|
|
*/
|
|
get connected(): boolean {
|
|
if (this.apiClient && !this.config.url) {
|
|
return true;
|
|
}
|
|
return this.isConnected;
|
|
}
|
|
|
|
// ========================================================================
|
|
// Flat Key Methods (for meta-core flat key architecture)
|
|
// ========================================================================
|
|
|
|
/**
|
|
* Get a single property value from Redis using flat key format
|
|
* Key format: file:{hashId}/{property}
|
|
*
|
|
* @param hashId - The file hash ID
|
|
* @param property - The property name (can include / for nested props like 'titles/eng')
|
|
* @returns The property value or null if not found
|
|
*/
|
|
async getProperty(hashId: string, property: string): Promise<string | null> {
|
|
if (this.apiClient) {
|
|
try {
|
|
return await this.apiClient.getProperty(hashId, property);
|
|
} catch (error: any) {
|
|
logger.error(`HTTP getProperty failed for ${hashId}/${property}:`, error.message);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
if (!this.client || !this.isConnected) {
|
|
return null;
|
|
}
|
|
|
|
try {
|
|
const key = `${this.config.prefix}file:${hashId}/${property}`;
|
|
return await this.client.get(key);
|
|
} catch (error: any) {
|
|
logger.error(`Failed to get property ${property} for ${hashId}:`, error.message);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get multiple properties for a file using flat keys
|
|
* More efficient than multiple getProperty calls using MGET
|
|
*
|
|
* @param hashId - The file hash ID
|
|
* @param properties - Array of property names to fetch
|
|
* @returns Map of property name to value (null values excluded)
|
|
*/
|
|
async getProperties(hashId: string, properties: string[]): Promise<Map<string, string>> {
|
|
const result = new Map<string, string>();
|
|
|
|
if (!this.client || !this.isConnected || properties.length === 0) {
|
|
return result;
|
|
}
|
|
|
|
try {
|
|
const keys = properties.map(prop => `${this.config.prefix}file:${hashId}/${prop}`);
|
|
const values = await this.client.mget(keys);
|
|
|
|
for (let i = 0; i < properties.length; i++) {
|
|
const value = values[i];
|
|
if (value !== null) {
|
|
result.set(properties[i], value);
|
|
}
|
|
}
|
|
} catch (error: any) {
|
|
logger.error(`Failed to get properties for ${hashId}:`, error.message);
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
/**
|
|
* Parse a meta:events key to extract hashId and property
|
|
* Key format: file:{hashId}/{property}
|
|
*
|
|
* @param key - The Redis key from the event
|
|
* @returns Parsed key with hashId and property, or null if invalid format
|
|
*/
|
|
parseMetaEventKey(key: string): ParsedMetaEventKey | null {
|
|
// Remove prefix if present
|
|
let keyWithoutPrefix = key;
|
|
if (this.config.prefix && key.startsWith(this.config.prefix)) {
|
|
keyWithoutPrefix = key.slice(this.config.prefix.length);
|
|
}
|
|
|
|
// Key format: file:{hashId}/{property}
|
|
// Property can contain slashes (e.g., 'titles/eng')
|
|
const match = keyWithoutPrefix.match(/^file:([^/]+)\/(.+)$/);
|
|
if (!match) {
|
|
return null;
|
|
}
|
|
|
|
return {
|
|
hashId: match[1],
|
|
property: match[2],
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Check if a key is a file metadata key (vs other keys like indexes)
|
|
*
|
|
* @param key - The Redis key to check
|
|
* @returns True if this is a file metadata key
|
|
*/
|
|
isFileMetadataKey(key: string): boolean {
|
|
return this.parseMetaEventKey(key) !== null;
|
|
}
|
|
|
|
// ========================================================================
|
|
// Legacy HGETALL-based Methods (deprecated, kept for compatibility)
|
|
// ========================================================================
|
|
|
|
/**
|
|
* Get all file metadata from Redis
|
|
* @deprecated Use streaming state builder with flat keys instead
|
|
*/
|
|
async getAllFiles(): Promise<Map<string, FileMetadata>> {
|
|
const files = new Map<string, FileMetadata>();
|
|
|
|
if (!this.client || !this.isConnected) {
|
|
logger.warn('Redis not connected, returning empty file list');
|
|
return files;
|
|
}
|
|
|
|
try {
|
|
// Get all file keys using the index
|
|
const indexKey = `${this.config.prefix}file:__index__`;
|
|
const hashIds = await this.client.smembers(indexKey);
|
|
|
|
if (hashIds.length === 0) {
|
|
// Fallback: scan for file keys directly
|
|
const pattern = `${this.config.prefix}file:*`;
|
|
const keys = await this.scanKeys(pattern);
|
|
logger.debug(`Found ${keys.length} file keys in Redis (via scan)`);
|
|
|
|
for (const key of keys) {
|
|
if (key.includes('__index__')) continue;
|
|
|
|
try {
|
|
const data = await this.client.hgetall(key);
|
|
if (data && Object.keys(data).length > 0) {
|
|
const hashId = key.replace(`${this.config.prefix}file:`, '');
|
|
const metadata = this.parseMetadata(data, hashId);
|
|
files.set(metadata.sourcePath, metadata);
|
|
}
|
|
} catch (err) {
|
|
logger.warn(`Failed to get metadata for key ${key}:`, err);
|
|
}
|
|
}
|
|
} else {
|
|
logger.debug(`Found ${hashIds.length} files in index`);
|
|
|
|
// Fetch metadata for each hash ID
|
|
for (const hashId of hashIds) {
|
|
try {
|
|
const key = `${this.config.prefix}file:${hashId}`;
|
|
const data = await this.client.hgetall(key);
|
|
if (data && Object.keys(data).length > 0) {
|
|
const metadata = this.parseMetadata(data, hashId);
|
|
files.set(metadata.sourcePath, metadata);
|
|
}
|
|
} catch (err) {
|
|
logger.warn(`Failed to get metadata for hashId ${hashId}:`, err);
|
|
}
|
|
}
|
|
}
|
|
|
|
logger.info(`Loaded ${files.size} files from Redis`);
|
|
} catch (error) {
|
|
logger.error('Failed to get files from Redis:', error);
|
|
}
|
|
|
|
return files;
|
|
}
|
|
|
|
/**
|
|
* Get file metadata by hash ID
|
|
* @deprecated Use streaming state builder with flat keys instead
|
|
*/
|
|
async getFileByHashId(hashId: string): Promise<FileMetadata | null> {
|
|
if (!this.client || !this.isConnected) {
|
|
return null;
|
|
}
|
|
|
|
try {
|
|
const key = `${this.config.prefix}file:${hashId}`;
|
|
const data = await this.client.hgetall(key);
|
|
|
|
if (data && Object.keys(data).length > 0) {
|
|
return this.parseMetadata(data, hashId);
|
|
}
|
|
|
|
return null;
|
|
} catch (error) {
|
|
logger.error(`Failed to get file ${hashId}:`, error);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get all hash IDs from Redis index
|
|
*/
|
|
async getAllHashIds(): Promise<string[]> {
|
|
if (!this.client || !this.isConnected) {
|
|
return [];
|
|
}
|
|
|
|
try {
|
|
const indexKey = `${this.config.prefix}file:__index__`;
|
|
return await this.client.smembers(indexKey);
|
|
} catch (error) {
|
|
logger.error('Failed to get hash IDs:', error);
|
|
return [];
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Get count of files in Redis index
|
|
* More efficient than getAllHashIds() when you only need the count
|
|
*/
|
|
async getIndexCount(): Promise<number> {
|
|
if (!this.client || !this.isConnected) {
|
|
return 0;
|
|
}
|
|
|
|
try {
|
|
const indexKey = `${this.config.prefix}file:__index__`;
|
|
return await this.client.scard(indexKey);
|
|
} catch (error) {
|
|
logger.error('Failed to get index count:', error);
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Scan flat keys to build file metadata for all indexed files
|
|
* This is used as a fallback when the meta:events stream is empty or unavailable.
|
|
*
|
|
* Algorithm:
|
|
* 1. Get all hash IDs from the index
|
|
* 2. For each hash ID, fetch VFS-relevant properties using MGET
|
|
* 3. Return a Map of hashId to flat properties
|
|
*
|
|
* This replaces the broken getAllFiles() which used HGETALL on flat keys.
|
|
*
|
|
* @param vfsRelevantProperties - Set of property names that are VFS-relevant
|
|
* @returns Map of hashId to properties (only files with filePath)
|
|
*/
|
|
async scanFlatKeysForFiles(vfsRelevantProperties: Set<string>): Promise<Map<string, Record<string, string>>> {
|
|
const result = new Map<string, Record<string, string>>();
|
|
|
|
if (!this.client || !this.isConnected) {
|
|
logger.warn('Redis not connected, returning empty result');
|
|
return result;
|
|
}
|
|
|
|
try {
|
|
// Get all hash IDs from the index
|
|
const hashIds = await this.getAllHashIds();
|
|
logger.info(`Scanning flat keys for ${hashIds.length} files...`);
|
|
|
|
if (hashIds.length === 0) {
|
|
return result;
|
|
}
|
|
|
|
// Batch process hashIds to avoid overwhelming Redis
|
|
const batchSize = 100;
|
|
const propertiesArray = Array.from(vfsRelevantProperties);
|
|
|
|
for (let i = 0; i < hashIds.length; i += batchSize) {
|
|
const batch = hashIds.slice(i, i + batchSize);
|
|
|
|
// Process each hashId in the batch
|
|
const batchPromises = batch.map(async (hashId) => {
|
|
const props = await this.getProperties(hashId, propertiesArray);
|
|
|
|
// Only include files that have a filePath
|
|
if (props.has('filePath')) {
|
|
const propsObj: Record<string, string> = {};
|
|
for (const [key, value] of props) {
|
|
propsObj[key] = value;
|
|
}
|
|
return { hashId, props: propsObj };
|
|
}
|
|
return null;
|
|
});
|
|
|
|
const batchResults = await Promise.all(batchPromises);
|
|
|
|
for (const item of batchResults) {
|
|
if (item) {
|
|
result.set(item.hashId, item.props);
|
|
}
|
|
}
|
|
}
|
|
|
|
logger.info(`Flat key scan complete: ${result.size} files with filePath`);
|
|
return result;
|
|
} catch (error: any) {
|
|
logger.error('Failed to scan flat keys:', error.message);
|
|
return result;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Scan Redis keys matching pattern
|
|
*/
|
|
private async scanKeys(pattern: string): Promise<string[]> {
|
|
if (!this.client) return [];
|
|
|
|
const keys: string[] = [];
|
|
let cursor = '0';
|
|
|
|
do {
|
|
const [newCursor, batch] = await this.client.scan(cursor, 'MATCH', pattern, 'COUNT', 1000);
|
|
cursor = newCursor;
|
|
keys.push(...batch);
|
|
} while (cursor !== '0');
|
|
|
|
return keys;
|
|
}
|
|
|
|
/**
|
|
* Parse metadata from Redis hash
|
|
* Handles meta-sort format and resolves paths with FILES_VOLUME
|
|
*/
|
|
private parseMetadata(data: Record<string, string>, hashId?: string): FileMetadata {
|
|
// Get the original path (relative to FILES_VOLUME)
|
|
const originalPath = data.filePath ?? data.sourcePath ?? '';
|
|
|
|
// Resolve full source path by prepending FILES_VOLUME
|
|
let sourcePath = originalPath;
|
|
if (originalPath && !originalPath.startsWith('/')) {
|
|
// Relative path - prepend FILES_VOLUME
|
|
sourcePath = `${this.config.filesVolume}/${originalPath}`;
|
|
} else if (!originalPath.startsWith(this.config.filesVolume) && originalPath.startsWith('/')) {
|
|
// Absolute path but not in FILES_VOLUME - still prepend
|
|
sourcePath = `${this.config.filesVolume}${originalPath}`;
|
|
}
|
|
|
|
// Parse size
|
|
const size = parseInt(data.fileSize ?? data.size ?? data.sizeByte ?? '0', 10);
|
|
|
|
// Parse timestamps
|
|
let mtime = 0;
|
|
if (data.mtime) {
|
|
const parsed = Date.parse(data.mtime);
|
|
mtime = isNaN(parsed) ? parseFloat(data.mtime) : parsed;
|
|
}
|
|
|
|
// Parse season/episode (can be 0 for specials)
|
|
const season = data.season !== undefined ? parseInt(data.season, 10) : undefined;
|
|
const episode = data.episode !== undefined ? parseInt(data.episode, 10) : undefined;
|
|
|
|
// Parse year
|
|
const movieYear = data.movieYear ? parseInt(data.movieYear, 10) : undefined;
|
|
const year = data.year ? parseInt(data.year, 10) : movieYear;
|
|
|
|
return {
|
|
// Path information
|
|
sourcePath,
|
|
originalPath,
|
|
|
|
// File stats
|
|
size,
|
|
mtime,
|
|
ctime: data.ctime ? parseFloat(data.ctime) : mtime,
|
|
|
|
// Identification.
|
|
//
|
|
// `hashId` is always supplied by the caller — every `parseMetadata`
|
|
// call site derives it from the Redis key (`file:<hashId>/…`), so the
|
|
// fallbacks below it were already unreachable. The two that used to sit
|
|
// here (`data.cid_midhash256`, then `data.hashId`) were also the wrong
|
|
// shape: `cid_*` fields no longer exist on records, and meta-core
|
|
// rejects them at the write boundary. Kept as a single defensive `??`
|
|
// rather than a chain of dead ones. See METADATA_KEYS.md §14.13.
|
|
hashId: hashId ?? data.hashId,
|
|
|
|
// Title information
|
|
title: data.title,
|
|
originalTitle: data.originalTitle,
|
|
fileName: data.fileName,
|
|
|
|
// Series metadata
|
|
season,
|
|
episode,
|
|
extra: data.extra === 'true',
|
|
|
|
// Movie metadata
|
|
movieYear,
|
|
year,
|
|
|
|
// File type
|
|
fileType: data.fileType,
|
|
extension: data.extension,
|
|
|
|
// Version/variant
|
|
version: data.version,
|
|
subtitleLanguage: data.subtitleLanguage,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Subscribe to Redis pub/sub channel
|
|
*/
|
|
async subscribe(channel: string, callback: (message: string) => void): Promise<void> {
|
|
if (!this.client) {
|
|
throw new Error('Redis not connected');
|
|
}
|
|
|
|
// Create subscriber connection if not exists
|
|
if (!this.subscriber) {
|
|
this.subscriber = this.client.duplicate();
|
|
await new Promise<void>((resolve, reject) => {
|
|
const timeout = setTimeout(() => reject(new Error('Subscriber connection timeout')), 10000);
|
|
this.subscriber!.once('ready', () => {
|
|
clearTimeout(timeout);
|
|
resolve();
|
|
});
|
|
this.subscriber!.once('error', (err) => {
|
|
clearTimeout(timeout);
|
|
reject(err);
|
|
});
|
|
});
|
|
|
|
// Set up message handler
|
|
this.subscriber.on('message', (ch: string, message: string) => {
|
|
const handler = this.subscriptions.get(ch);
|
|
if (handler) {
|
|
try {
|
|
handler(message);
|
|
} catch (error: any) {
|
|
logger.error(`Error in subscription handler for ${ch}:`, error.message);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
// Store callback and subscribe
|
|
this.subscriptions.set(channel, callback);
|
|
await this.subscriber.subscribe(channel);
|
|
logger.info(`Subscribed to channel: ${channel}`);
|
|
}
|
|
|
|
/**
|
|
* Unsubscribe from Redis pub/sub channel
|
|
*/
|
|
async unsubscribe(channel: string): Promise<void> {
|
|
if (this.subscriber) {
|
|
await this.subscriber.unsubscribe(channel);
|
|
this.subscriptions.delete(channel);
|
|
logger.info(`Unsubscribed from channel: ${channel}`);
|
|
}
|
|
}
|
|
|
|
// ========================================================================
|
|
// Redis Streams Consumer Methods
|
|
// ========================================================================
|
|
|
|
/**
|
|
* Initialize stream consumer group
|
|
* Creates the consumer group at position 0 to read all historical events
|
|
*
|
|
* @param stream - Stream name (e.g., 'file:events')
|
|
* @param group - Consumer group name (e.g., 'meta-fuse-vfs')
|
|
*/
|
|
async initStreamConsumer(stream: string, group: string): Promise<void> {
|
|
if (!this.client) {
|
|
throw new Error('Redis not connected');
|
|
}
|
|
|
|
try {
|
|
// Create consumer group at position 0 (read all historical events)
|
|
// MKSTREAM creates the stream if it doesn't exist
|
|
await this.client.xgroup('CREATE', stream, group, '0', 'MKSTREAM');
|
|
logger.info(`Created consumer group '${group}' for stream '${stream}'`);
|
|
} catch (error: any) {
|
|
// BUSYGROUP means group already exists - that's fine
|
|
if (error.message?.includes('BUSYGROUP')) {
|
|
logger.debug(`Consumer group '${group}' already exists`);
|
|
} else {
|
|
throw error;
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Process pending entries from crashed consumers
|
|
* Uses XAUTOCLAIM to claim entries that have been idle too long
|
|
*
|
|
* @param stream - Stream name
|
|
* @param group - Consumer group name
|
|
* @param minIdleTime - Minimum idle time in ms before claiming (default: 30000)
|
|
* @param onMessage - Handler for each message
|
|
*/
|
|
async processPendingEntries(
|
|
stream: string,
|
|
group: string,
|
|
minIdleTime: number = 30000,
|
|
onMessage: StreamMessageHandler
|
|
): Promise<number> {
|
|
if (!this.client) {
|
|
return 0;
|
|
}
|
|
|
|
let processed = 0;
|
|
let cursor = '0-0';
|
|
|
|
try {
|
|
while (true) {
|
|
// XAUTOCLAIM claims idle entries and returns them
|
|
const result = await this.client.xautoclaim(
|
|
stream,
|
|
group,
|
|
this.consumerName,
|
|
minIdleTime,
|
|
cursor,
|
|
'COUNT',
|
|
100
|
|
) as [string, Array<[string, string[]]>, string[]];
|
|
|
|
const [nextCursor, entries] = result;
|
|
cursor = nextCursor;
|
|
|
|
if (!entries || entries.length === 0) {
|
|
break;
|
|
}
|
|
|
|
for (const [id, fields] of entries) {
|
|
try {
|
|
const message = this.parseStreamEntry(id, fields);
|
|
if (message) {
|
|
await onMessage(message);
|
|
// ACK the message after successful processing
|
|
await this.client.xack(stream, group, id);
|
|
processed++;
|
|
}
|
|
} catch (error: any) {
|
|
logger.error(`Error processing pending entry ${id}:`, error.message);
|
|
}
|
|
}
|
|
|
|
// If cursor is 0-0, we've processed all pending entries
|
|
if (cursor === '0-0') {
|
|
break;
|
|
}
|
|
}
|
|
|
|
if (processed > 0) {
|
|
logger.info(`Processed ${processed} pending stream entries`);
|
|
}
|
|
} catch (error: any) {
|
|
logger.error('Error processing pending entries:', error.message);
|
|
}
|
|
|
|
return processed;
|
|
}
|
|
|
|
/**
|
|
* Start stream consumer loop
|
|
* Reads messages using XREADGROUP and calls the handler
|
|
*
|
|
* @param stream - Stream name
|
|
* @param group - Consumer group name
|
|
* @param onMessage - Handler for each message
|
|
* @param blockMs - Block timeout in ms (default: 5000)
|
|
*/
|
|
async startStreamConsumer(
|
|
stream: string,
|
|
group: string,
|
|
onMessage: StreamMessageHandler,
|
|
blockMs: number = 5000
|
|
): Promise<void> {
|
|
if (!this.client) {
|
|
throw new Error('Redis not connected');
|
|
}
|
|
|
|
if (this.streamConsumerRunning) {
|
|
logger.warn('Stream consumer already running');
|
|
return;
|
|
}
|
|
|
|
this.streamConsumerRunning = true;
|
|
this.streamConsumerAbort = new AbortController();
|
|
|
|
logger.info(`Starting stream consumer: ${stream} (group: ${group}, consumer: ${this.consumerName})`);
|
|
|
|
// Consumer loop
|
|
while (this.streamConsumerRunning && !this.streamConsumerAbort.signal.aborted) {
|
|
try {
|
|
// XREADGROUP blocks waiting for new messages
|
|
// '>' means only read new messages (not already delivered to this consumer)
|
|
// Use call() to bypass TypeScript's strict argument checking for xreadgroup
|
|
const result = await (this.client.call(
|
|
'XREADGROUP',
|
|
'GROUP', group,
|
|
this.consumerName,
|
|
'BLOCK', blockMs,
|
|
'COUNT', 10,
|
|
'STREAMS', stream,
|
|
'>'
|
|
) as Promise<Array<[string, Array<[string, string[]]>]> | null>);
|
|
|
|
if (!result || result.length === 0) {
|
|
continue; // Timeout, loop again
|
|
}
|
|
|
|
// Process each stream's messages
|
|
for (const [streamName, entries] of result) {
|
|
for (const [id, fields] of entries) {
|
|
try {
|
|
const message = this.parseStreamEntry(id, fields);
|
|
if (message) {
|
|
await onMessage(message);
|
|
// ACK the message after successful processing
|
|
await this.client!.xack(stream, group, id);
|
|
}
|
|
} catch (error: any) {
|
|
logger.error(`Error processing stream entry ${id}:`, error.message);
|
|
// Don't ACK on error - message will be reprocessed
|
|
}
|
|
}
|
|
}
|
|
} catch (error: any) {
|
|
if (this.streamConsumerAbort?.signal.aborted) {
|
|
break;
|
|
}
|
|
logger.error('Stream consumer error:', error.message);
|
|
// Brief pause before retrying
|
|
await new Promise(resolve => setTimeout(resolve, 1000));
|
|
}
|
|
}
|
|
|
|
logger.info('Stream consumer stopped');
|
|
}
|
|
|
|
/**
|
|
* Stop the stream consumer loop
|
|
*/
|
|
stopStreamConsumer(): void {
|
|
if (!this.streamConsumerRunning) {
|
|
return;
|
|
}
|
|
|
|
logger.info('Stopping stream consumer...');
|
|
this.streamConsumerRunning = false;
|
|
this.streamConsumerAbort?.abort();
|
|
this.streamConsumerAbort = null;
|
|
}
|
|
|
|
/**
|
|
* Read stream entries from a specific position using XREAD (no consumer groups)
|
|
* Used for streaming bootstrap - replays all events from the beginning
|
|
*
|
|
* @param stream - Stream name
|
|
* @param fromId - Start position ('0' for beginning, '$' for end/new only)
|
|
* @param count - Maximum entries to read per call
|
|
* @param blockMs - Block timeout (0 for non-blocking)
|
|
* @returns Array of [id, message] tuples, empty if none
|
|
*/
|
|
async readStreamFrom(
|
|
stream: string,
|
|
fromId: string,
|
|
count: number = 100,
|
|
blockMs: number = 0
|
|
): Promise<Array<{ id: string; message: StreamMessage }>> {
|
|
if (!this.client || !this.isConnected) {
|
|
return [];
|
|
}
|
|
|
|
try {
|
|
let result: Array<[string, Array<[string, string[]]>]> | null;
|
|
|
|
if (blockMs > 0) {
|
|
result = await this.client.call(
|
|
'XREAD',
|
|
'BLOCK', blockMs,
|
|
'COUNT', count,
|
|
'STREAMS', stream,
|
|
fromId
|
|
) as typeof result;
|
|
} else {
|
|
result = await this.client.call(
|
|
'XREAD',
|
|
'COUNT', count,
|
|
'STREAMS', stream,
|
|
fromId
|
|
) as typeof result;
|
|
}
|
|
|
|
if (!result || result.length === 0) {
|
|
return [];
|
|
}
|
|
|
|
const entries: Array<{ id: string; message: StreamMessage }> = [];
|
|
|
|
for (const [_streamName, streamEntries] of result) {
|
|
for (const [id, fields] of streamEntries) {
|
|
const message = this.parseStreamEntry(id, fields);
|
|
if (message) {
|
|
entries.push({ id, message });
|
|
}
|
|
}
|
|
}
|
|
|
|
return entries;
|
|
} catch (error: any) {
|
|
logger.error('Error reading stream:', error.message);
|
|
return [];
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Read all stream entries from a position, calling handler for each
|
|
* Returns the last processed entry ID for resumption
|
|
*
|
|
* @param stream - Stream name
|
|
* @param fromId - Start position ('0' for beginning)
|
|
* @param onMessage - Handler for each message
|
|
* @param batchSize - Entries per read
|
|
* @returns Last processed entry ID, or fromId if no entries
|
|
*/
|
|
async replayStream(
|
|
stream: string,
|
|
fromId: string,
|
|
onMessage: StreamMessageHandler,
|
|
batchSize: number = 100
|
|
): Promise<string> {
|
|
let lastId = fromId;
|
|
let totalProcessed = 0;
|
|
const startTime = Date.now();
|
|
|
|
logger.info(`Replaying stream ${stream} from position ${fromId}...`);
|
|
|
|
while (true) {
|
|
const entries = await this.readStreamFrom(stream, lastId, batchSize, 0);
|
|
|
|
if (entries.length === 0) {
|
|
break;
|
|
}
|
|
|
|
for (const { id, message } of entries) {
|
|
try {
|
|
await onMessage(message);
|
|
lastId = id;
|
|
totalProcessed++;
|
|
} catch (error: any) {
|
|
logger.error(`Error processing stream entry ${id}:`, error.message);
|
|
// Continue with next entry
|
|
}
|
|
}
|
|
|
|
// If we got fewer than batchSize entries, we've reached the end
|
|
if (entries.length < batchSize) {
|
|
break;
|
|
}
|
|
}
|
|
|
|
const elapsed = Date.now() - startTime;
|
|
logger.info(`Stream replay complete: ${totalProcessed} entries in ${elapsed}ms, last ID: ${lastId}`);
|
|
|
|
return lastId;
|
|
}
|
|
|
|
/**
|
|
* Start a stream consumer that reads from a specific position using XREAD
|
|
* Does not use consumer groups - for simpler replay/live consumption
|
|
*
|
|
* @param stream - Stream name
|
|
* @param fromId - Start position ('0' for beginning, or last ID from replayStream)
|
|
* @param onMessage - Handler for each message
|
|
* @param blockMs - Block timeout for polling (default: 5000)
|
|
*/
|
|
async startSimpleStreamConsumer(
|
|
stream: string,
|
|
fromId: string,
|
|
onMessage: StreamMessageHandler,
|
|
blockMs: number = 5000
|
|
): Promise<void> {
|
|
if (!this.client) {
|
|
throw new Error('Redis not connected');
|
|
}
|
|
|
|
if (this.streamConsumerRunning) {
|
|
logger.warn('Stream consumer already running');
|
|
return;
|
|
}
|
|
|
|
this.streamConsumerRunning = true;
|
|
this.streamConsumerAbort = new AbortController();
|
|
|
|
let lastId = fromId;
|
|
logger.info(`Starting simple stream consumer: ${stream} from ${fromId}`);
|
|
|
|
while (this.streamConsumerRunning && !this.streamConsumerAbort.signal.aborted) {
|
|
try {
|
|
const entries = await this.readStreamFrom(stream, lastId, 10, blockMs);
|
|
|
|
for (const { id, message } of entries) {
|
|
try {
|
|
await onMessage(message);
|
|
lastId = id;
|
|
} catch (error: any) {
|
|
logger.error(`Error processing stream entry ${id}:`, error.message);
|
|
}
|
|
}
|
|
} catch (error: any) {
|
|
if (this.streamConsumerAbort?.signal.aborted) {
|
|
break;
|
|
}
|
|
logger.error('Stream consumer error:', error.message);
|
|
await new Promise(resolve => setTimeout(resolve, 1000));
|
|
}
|
|
}
|
|
|
|
logger.info('Simple stream consumer stopped');
|
|
}
|
|
|
|
/**
|
|
* Parse a stream entry into a StreamMessage
|
|
*
|
|
* Supports two formats:
|
|
* 1. New meta:events format: { type: 'set'|'del', key: 'file:abc123/prop', ts: '...' }
|
|
* 2. Legacy file:events format: { type: 'add'|'change'|..., path: '...', payload: '...' }
|
|
*/
|
|
private parseStreamEntry(id: string, fields: string[]): StreamMessage | null {
|
|
// Fields come as flat array: ['type', 'set', 'key', 'file:abc123/prop', 'ts', '123']
|
|
const fieldMap: Record<string, string> = {};
|
|
for (let i = 0; i < fields.length; i += 2) {
|
|
fieldMap[fields[i]] = fields[i + 1];
|
|
}
|
|
|
|
if (!fieldMap.type) {
|
|
logger.warn(`Invalid stream entry ${id}: missing type`);
|
|
return null;
|
|
}
|
|
|
|
// New meta:events format (set/del events with key field)
|
|
const metaEventTypes = ['set', 'del'];
|
|
if (metaEventTypes.includes(fieldMap.type)) {
|
|
if (!fieldMap.key) {
|
|
logger.warn(`Invalid stream entry ${id}: missing key for ${fieldMap.type} event`);
|
|
return null;
|
|
}
|
|
|
|
return {
|
|
id,
|
|
type: fieldMap.type as 'set' | 'del',
|
|
key: fieldMap.key,
|
|
ts: fieldMap.ts,
|
|
timestamp: fieldMap.ts || fieldMap.timestamp || '0',
|
|
};
|
|
}
|
|
|
|
// Legacy file:events format
|
|
// For direct file events (add/change/delete/rename), path is required
|
|
// For legacy events (batch/reset/plugin:complete), payload is required
|
|
const directEventTypes = ['add', 'change', 'delete', 'rename'];
|
|
const isDirectEvent = directEventTypes.includes(fieldMap.type);
|
|
|
|
if (isDirectEvent && !fieldMap.path) {
|
|
logger.warn(`Invalid stream entry ${id}: missing path for ${fieldMap.type} event`);
|
|
return null;
|
|
}
|
|
|
|
const legacyEventTypes = ['batch', 'reset', 'plugin:complete'];
|
|
if (legacyEventTypes.includes(fieldMap.type) && !fieldMap.payload) {
|
|
logger.warn(`Invalid stream entry ${id}: missing payload for ${fieldMap.type} event`);
|
|
return null;
|
|
}
|
|
|
|
return {
|
|
id,
|
|
type: fieldMap.type as StreamMessage['type'],
|
|
path: fieldMap.path,
|
|
size: fieldMap.size,
|
|
midhash256: fieldMap.midhash256,
|
|
oldPath: fieldMap.oldPath,
|
|
payload: fieldMap.payload,
|
|
timestamp: fieldMap.timestamp || '0',
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Get stats about Redis data
|
|
*/
|
|
async getStats(): Promise<{ totalKeys: number; fileCount: number; connected: boolean }> {
|
|
if (!this.client || !this.isConnected) {
|
|
return { totalKeys: 0, fileCount: 0, connected: false };
|
|
}
|
|
|
|
try {
|
|
// Try index first
|
|
const indexKey = `${this.config.prefix}file:__index__`;
|
|
const count = await this.client.scard(indexKey);
|
|
|
|
if (count > 0) {
|
|
return {
|
|
totalKeys: count,
|
|
fileCount: count,
|
|
connected: true,
|
|
};
|
|
}
|
|
|
|
// Fallback to scanning
|
|
const filePattern = `${this.config.prefix}file:*`;
|
|
const fileKeys = await this.scanKeys(filePattern);
|
|
const actualFileKeys = fileKeys.filter(k => !k.includes('__index__'));
|
|
|
|
return {
|
|
totalKeys: actualFileKeys.length,
|
|
fileCount: actualFileKeys.length,
|
|
connected: true,
|
|
};
|
|
} catch {
|
|
return { totalKeys: 0, fileCount: 0, connected: false };
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Health check
|
|
*/
|
|
async health(): Promise<boolean> {
|
|
if (!this.client || !this.isConnected) {
|
|
return false;
|
|
}
|
|
|
|
try {
|
|
const result = await this.client.ping();
|
|
return result === 'PONG';
|
|
} catch {
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Close connections
|
|
*/
|
|
async close(): Promise<void> {
|
|
await this.disconnect();
|
|
}
|
|
|
|
/**
|
|
* Get the raw ioredis client for direct access if needed
|
|
* Use with caution - prefer using the typed methods above
|
|
*/
|
|
getRawClient(): Redis | null {
|
|
return this.client;
|
|
}
|
|
}
|