metastar/meta-dup
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

initial commit of meta-dup

Browse Source
This commit is contained in:
Pierre
2026-02-08 16:29:39 +01:00
parent b3a9adbf3c
commit 2baa9a6925
27 changed files with 6735 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/meta-dup-core/package.json
+28
View File
@@ -0,0 +1,28 @@
{
"name": "@meta-dup/core",
"version": "1.0.0",
"description": "Core duplicate detection service for meta-dup",
"main": "dist/index.js",
"type": "module",
"scripts": {
"build": "tsc",
"dev": "tsc --watch",
"start": "node dist/index.js",
"test": "vitest run",
"lint": "eslint src --ext ts"
},
"license": "MIT",
"dependencies": {
"@fastify/cors": "^11.1.0",
"@fastify/static": "^8.0.0",
"dotenv": "^16.4.1",
"fastify": "^5.6.1",
"ioredis": "^5.4.1",
"tslog": "^4.9.2"
},
"devDependencies": {
"@types/node": "^20.11.16",
"typescript": "^5.3.3",
"vitest": "^1.0.0"
}
}
packages/meta-dup-core/src/DuplicateIndex.ts
+334
View File
@@ -0,0 +1,334 @@
/**
* DuplicateIndex - Core duplicate detection logic
*
* Maintains in-memory indexes for:
* - Hash duplicates: files with the same SHA-256 hash (exact copies)
* - Title duplicates: files with the same parsed title (different releases)
*
* Updates in real-time via Redis Streams events from meta-sort.
*/
import { Logger } from 'tslog';
import type { FileMetadata, RedisClient } from './kv/RedisClient.js';
const logger = new Logger({ name: 'DuplicateIndex' });
/**
* Information about a file in a duplicate group
*/
export interface FileInfo {
hashId: string;
filePath: string;
title?: string;
sha256?: string;
sizeByte?: number;
year?: number;
season?: number;
episode?: number;
fileType?: string;
}
/**
* A group of duplicate files sharing the same key (hash or title)
*/
export interface DuplicateGroup {
key: string; // hash or title
files: FileInfo[]; // Array of file info
}
/**
* Statistics about duplicate detection
*/
export interface DuplicateStats {
hashGroupCount: number;
hashFileCount: number;
titleGroupCount: number;
titleFileCount: number;
totalFilesTracked: number;
lastUpdated: string;
}
/**
* Full duplicate data response
*/
export interface DuplicateData {
hashDuplicates: DuplicateGroup[];
titleDuplicates: DuplicateGroup[];
stats: DuplicateStats;
computedAt: string;
}
export class DuplicateIndex {
// Primary indexes: key -> Set of hashIds
private byHash: Map<string, Set<string>> = new Map();
private byTitle: Map<string, Set<string>> = new Map();
// File info storage: hashId -> FileInfo
private fileInfo: Map<string, FileInfo> = new Map();
// Track when last updated
private lastUpdated: Date = new Date();
constructor() {
logger.info('DuplicateIndex initialized');
}
/**
* Add or update a file in the index
*/
addFile(hashId: string, metadata: FileMetadata): void {
// First remove any existing entry to ensure clean state
this.removeFile(hashId);
// Extract relevant info
const info: FileInfo = {
hashId,
filePath: metadata.originalPath || metadata.sourcePath,
title: this.extractTitle(metadata),
sha256: metadata.sha256,
sizeByte: metadata.size,
year: metadata.year || metadata.movieYear,
season: metadata.season,
episode: metadata.episode,
fileType: metadata.fileType,
};
// Store file info
this.fileInfo.set(hashId, info);
// Index by SHA-256 hash (if available)
if (info.sha256) {
const hashSet = this.byHash.get(info.sha256) ?? new Set();
hashSet.add(hashId);
this.byHash.set(info.sha256, hashSet);
}
// Index by title (if available)
const titleKey = this.getTitleKey(metadata);
if (titleKey) {
const titleSet = this.byTitle.get(titleKey) ?? new Set();
titleSet.add(hashId);
this.byTitle.set(titleKey, titleSet);
}
this.lastUpdated = new Date();
}
/**
* Remove a file from the index
*/
removeFile(hashId: string): void {
const info = this.fileInfo.get(hashId);
if (!info) return;
// Remove from hash index
if (info.sha256) {
const hashSet = this.byHash.get(info.sha256);
if (hashSet) {
hashSet.delete(hashId);
if (hashSet.size === 0) {
this.byHash.delete(info.sha256);
}
}
}
// Remove from title index
if (info.title) {
const titleKey = info.title.toLowerCase();
const titleSet = this.byTitle.get(titleKey);
if (titleSet) {
titleSet.delete(hashId);
if (titleSet.size === 0) {
this.byTitle.delete(titleKey);
}
}
}
// Remove file info
this.fileInfo.delete(hashId);
this.lastUpdated = new Date();
}
/**
* Extract title from metadata for duplicate detection
*/
private extractTitle(metadata: FileMetadata): string | undefined {
// Prefer parsed title over filename
if (metadata.title) {
return metadata.title;
}
if (metadata.titles?.eng) {
return metadata.titles.eng;
}
if (metadata.originalTitle) {
return metadata.originalTitle;
}
return undefined;
}
/**
* Get normalized title key for indexing
* Includes year, season, episode for more precise matching
*/
private getTitleKey(metadata: FileMetadata): string | null {
const title = this.extractTitle(metadata);
if (!title) return null;
// Build a normalized key
let key = title.toLowerCase().trim();
// Add year if available (for movies)
if (metadata.year || metadata.movieYear) {
key += `:${metadata.year || metadata.movieYear}`;
}
// Add season/episode if available (for TV shows)
if (metadata.season !== undefined) {
key += `:s${metadata.season}`;
}
if (metadata.episode !== undefined) {
key += `:e${metadata.episode}`;
}
return key;
}
/**
* Get all hash-based duplicates (files with same SHA-256)
*/
getHashDuplicates(): DuplicateGroup[] {
const groups: DuplicateGroup[] = [];
for (const [hash, hashIds] of this.byHash) {
if (hashIds.size > 1) {
const files = this.getFileInfos(hashIds);
groups.push({
key: hash,
files: files.sort((a, b) => a.filePath.localeCompare(b.filePath)),
});
}
}
// Sort by number of duplicates (descending)
return groups.sort((a, b) => b.files.length - a.files.length);
}
/**
* Get all title-based duplicates (files with same parsed title)
*/
getTitleDuplicates(): DuplicateGroup[] {
const groups: DuplicateGroup[] = [];
for (const [title, hashIds] of this.byTitle) {
if (hashIds.size > 1) {
const files = this.getFileInfos(hashIds);
groups.push({
key: title,
files: files.sort((a, b) => a.filePath.localeCompare(b.filePath)),
});
}
}
// Sort by number of duplicates (descending)
return groups.sort((a, b) => b.files.length - a.files.length);
}
/**
* Get FileInfo for a set of hashIds
*/
private getFileInfos(hashIds: Set<string>): FileInfo[] {
const files: FileInfo[] = [];
for (const hashId of hashIds) {
const info = this.fileInfo.get(hashId);
if (info) {
files.push(info);
}
}
return files;
}
/**
* Get statistics about duplicates
*/
getStats(): DuplicateStats {
const hashDuplicates = this.getHashDuplicates();
const titleDuplicates = this.getTitleDuplicates();
return {
hashGroupCount: hashDuplicates.length,
hashFileCount: hashDuplicates.reduce((sum, g) => sum + g.files.length, 0),
titleGroupCount: titleDuplicates.length,
titleFileCount: titleDuplicates.reduce((sum, g) => sum + g.files.length, 0),
totalFilesTracked: this.fileInfo.size,
lastUpdated: this.lastUpdated.toISOString(),
};
}
/**
* Get full duplicate data
*/
getDuplicateData(): DuplicateData {
return {
hashDuplicates: this.getHashDuplicates(),
titleDuplicates: this.getTitleDuplicates(),
stats: this.getStats(),
computedAt: new Date().toISOString(),
};
}
/**
* Clear all indexes
*/
clear(): void {
this.byHash.clear();
this.byTitle.clear();
this.fileInfo.clear();
this.lastUpdated = new Date();
logger.info('DuplicateIndex cleared');
}
/**
* Rebuild index from all files in Redis
*/
async rebuildFromRedis(redisClient: RedisClient): Promise<void> {
logger.info('Rebuilding DuplicateIndex from Redis...');
// Clear existing data
this.clear();
// Load all files from Redis
const files = await redisClient.getAllFiles();
logger.info(`Loading ${files.size} files into DuplicateIndex...`);
// Add each file to the index
for (const [hashId, metadata] of files) {
this.addFile(hashId, metadata);
}
const stats = this.getStats();
logger.info(`DuplicateIndex rebuilt: ${stats.totalFilesTracked} files, ` +
`${stats.hashGroupCount} hash duplicates, ${stats.titleGroupCount} title duplicates`);
}
/**
* Handle file update from Redis
*/
async onFileUpdate(
hashId: string,
action: 'add' | 'update' | 'remove',
redisClient: RedisClient
): Promise<void> {
if (action === 'remove') {
this.removeFile(hashId);
logger.debug(`Removed file ${hashId} from index`);
} else {
// add or update - fetch latest metadata from Redis
const metadata = await redisClient.getFileByHashId(hashId);
if (metadata) {
this.addFile(hashId, metadata);
logger.debug(`${action === 'add' ? 'Added' : 'Updated'} file ${hashId} in index`);
}
}
}
}
packages/meta-dup-core/src/api/APIServer.ts
+228
View File
@@ -0,0 +1,228 @@
/**
* API Server for meta-dup
*
* REST API endpoints:
* - GET /health - Health check
* - GET /api/duplicates - Get all duplicates
* - GET /api/duplicates/hash - Hash duplicates only
* - GET /api/duplicates/title - Title duplicates only
* - GET /api/duplicates/stats - Statistics
* - POST /api/duplicates/rebuild - Force rebuild from Redis
*/
import Fastify, { FastifyInstance, FastifyRequest, FastifyReply } from 'fastify';
import cors from '@fastify/cors';
import fastifyStatic from '@fastify/static';
import { Logger } from 'tslog';
import { DuplicateIndex } from '../DuplicateIndex.js';
import { KVManager } from '../kv/KVManager.js';
import { RedisClient } from '../kv/RedisClient.js';
import { join, dirname } from 'path';
import { fileURLToPath } from 'url';
import { existsSync } from 'fs';
const logger = new Logger({ name: 'APIServer' });
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);
export interface APIServerConfig {
port: number;
host: string;
}
export class APIServer {
private app: FastifyInstance;
private config: APIServerConfig;
private duplicateIndex: DuplicateIndex;
private kvManager: KVManager;
private redisClient: RedisClient | null = null;
constructor(
duplicateIndex: DuplicateIndex,
config: APIServerConfig,
kvManager: KVManager
) {
this.duplicateIndex = duplicateIndex;
this.config = config;
this.kvManager = kvManager;
this.app = Fastify({
logger: false,
});
this.setupMiddleware();
this.setupRoutes();
}
/**
* Set Redis client (called after connection established)
*/
setRedisClient(client: RedisClient): void {
this.redisClient = client;
}
/**
* Setup middleware
*/
private setupMiddleware(): void {
// Enable CORS
this.app.register(cors, {
origin: true,
credentials: true,
});
// Serve static UI files if they exist
const uiDistPath = join(__dirname, '../../..', 'meta-dup-ui', 'dist');
if (existsSync(uiDistPath)) {
this.app.register(fastifyStatic, {
root: uiDistPath,
prefix: '/',
});
logger.info(`Serving UI from ${uiDistPath}`);
} else {
logger.info('UI dist not found, serving API only');
}
}
/**
* Setup routes
*/
private setupRoutes(): void {
// Health check
this.app.get('/health', async (request: FastifyRequest, reply: FastifyReply) => {
const isHealthy = await this.kvManager.isHealthy();
const stats = this.duplicateIndex.getStats();
return {
status: isHealthy ? 'ok' : 'degraded',
service: 'meta-dup',
redis: isHealthy,
stats: {
filesTracked: stats.totalFilesTracked,
hashDuplicates: stats.hashGroupCount,
titleDuplicates: stats.titleGroupCount,
},
timestamp: new Date().toISOString(),
};
});
// Get all duplicates
this.app.get('/api/duplicates', async (request: FastifyRequest, reply: FastifyReply) => {
return this.duplicateIndex.getDuplicateData();
});
// Get hash duplicates only
this.app.get('/api/duplicates/hash', async (request: FastifyRequest, reply: FastifyReply) => {
return {
duplicates: this.duplicateIndex.getHashDuplicates(),
computedAt: new Date().toISOString(),
};
});
// Get title duplicates only
this.app.get('/api/duplicates/title', async (request: FastifyRequest, reply: FastifyReply) => {
return {
duplicates: this.duplicateIndex.getTitleDuplicates(),
computedAt: new Date().toISOString(),
};
});
// Get statistics
this.app.get('/api/duplicates/stats', async (request: FastifyRequest, reply: FastifyReply) => {
return this.duplicateIndex.getStats();
});
// Force rebuild from Redis
this.app.post('/api/duplicates/rebuild', async (request: FastifyRequest, reply: FastifyReply) => {
if (!this.redisClient) {
reply.status(503);
return { error: 'Redis not connected' };
}
logger.info('Manual rebuild requested');
const startTime = Date.now();
await this.duplicateIndex.rebuildFromRedis(this.redisClient);
const elapsed = Date.now() - startTime;
return {
success: true,
message: 'Index rebuilt from Redis',
elapsedMs: elapsed,
stats: this.duplicateIndex.getStats(),
};
});
// Service discovery (for dashboard navigation)
this.app.get('/api/services', async (request: FastifyRequest, reply: FastifyReply) => {
const services: Array<{
name: string;
url: string;
api: string;
status: string;
role?: string;
}> = [];
try {
const serviceDiscovery = this.kvManager.getServiceDiscovery();
if (serviceDiscovery) {
const allServices = await serviceDiscovery.discoverAllServices();
for (const svc of allServices) {
services.push({
name: svc.name || 'Unknown',
url: svc.baseUrl || '',
api: svc.baseUrl || '',
status: svc.status || 'unknown',
role: svc.role,
});
}
}
} catch (error: any) {
logger.error('Error discovering services:', error);
}
return {
services,
current: 'meta-dup',
};
});
// Catch-all for SPA routing (serve index.html for non-API routes)
this.app.setNotFoundHandler(async (request: FastifyRequest, reply: FastifyReply) => {
const uiDistPath = join(__dirname, '../../..', 'meta-dup-ui', 'dist');
const indexPath = join(uiDistPath, 'index.html');
if (existsSync(indexPath) && !request.url.startsWith('/api/')) {
return reply.sendFile('index.html');
}
reply.status(404);
return { error: 'Not found' };
});
}
/**
* Start the server
*/
async start(): Promise<void> {
try {
await this.app.listen({
port: this.config.port,
host: this.config.host,
});
logger.info(`API server listening on ${this.config.host}:${this.config.port}`);
} catch (error) {
logger.error('Failed to start API server:', error);
throw error;
}
}
/**
* Stop the server
*/
async stop(): Promise<void> {
await this.app.close();
logger.info('API server stopped');
}
}
packages/meta-dup-core/src/index.ts
+268
View File
@@ -0,0 +1,268 @@
/**
* meta-dup-core entry point
*
* Standalone duplicate detection service that:
* - Consumes Redis Streams events from meta-sort
* - Detects duplicates by hash (SHA-256) and by title
* - Provides REST API and UI dashboard
*
* Uses Redis Streams for reliable event delivery from meta-sort.
*/
import 'dotenv/config';
import { Logger } from 'tslog';
import { KVManager } from './kv/KVManager.js';
import { RedisClient, StreamMessage } from './kv/RedisClient.js';
import { DuplicateIndex } from './DuplicateIndex.js';
import { APIServer } from './api/APIServer.js';
const logger = new Logger({ name: 'meta-dup' });
// Redis Streams for reliable event delivery from meta-core
const EVENTS_STREAM = 'file:events';
const CONSUMER_GROUP = 'meta-dup-consumer';
interface BatchUpdatePayload {
timestamp: number;
changes: Array<{
action: 'add' | 'update' | 'remove';
hashId: string;
}>;
}
interface ResetPayload {
timestamp: number;
action: 'reset';
}
/**
* Handle a stream message from Redis Streams
* Routes messages by type to appropriate handlers
*
* Supports both:
* - Direct file events from meta-core (add/change/delete/rename)
* - Legacy batch events (for backward compatibility)
*/
async function handleStreamMessage(
message: StreamMessage,
duplicateIndex: DuplicateIndex,
redisClient: RedisClient
): Promise<void> {
try {
switch (message.type) {
// Direct file events from meta-core
case 'add':
case 'change': {
// File added or changed - look up by path and update index
// Note: We need to wait for meta-sort to process the file first
// For now, just log - the duplicate index is rebuilt on startup
logger.debug(`File ${message.type}: ${message.path}`);
break;
}
case 'delete': {
// File deleted - would need to find by path and remove
logger.debug(`File deleted: ${message.path}`);
break;
}
case 'rename': {
// File renamed - treated as delete + add
logger.debug(`File renamed: ${message.oldPath} -> ${message.path}`);
break;
}
// Legacy batch events (for backward compatibility)
case 'batch': {
if (!message.payload) break;
const batch: BatchUpdatePayload = JSON.parse(message.payload);
logger.debug(`Processing batch update: ${batch.changes.length} changes`);
for (const change of batch.changes) {
await duplicateIndex.onFileUpdate(change.hashId, change.action, redisClient);
}
break;
}
case 'reset': {
if (!message.payload) break;
const reset: ResetPayload = JSON.parse(message.payload);
logger.info(`Processing reset event: ${reset.action}`);
await duplicateIndex.rebuildFromRedis(redisClient);
break;
}
case 'plugin:complete': {
// Plugin completion events - update the file's metadata
// This is when we get full hash for duplicate detection
if (message.payload) {
try {
const payload = JSON.parse(message.payload);
if (payload.fileHash && payload.pluginId === 'fullhash') {
// Full hash plugin completed - this is when we can detect duplicates
await duplicateIndex.onFileUpdate(payload.fileHash, 'update', redisClient);
}
} catch {
// Ignore parse errors
}
}
break;
}
default:
logger.warn(`Unknown stream message type: ${message.type}`);
}
} catch (error: any) {
logger.error(`Error processing stream message ${message.id}:`, error.message);
throw error; // Re-throw to prevent ACK
}
}
async function main(): Promise<void> {
logger.info('Starting meta-dup...');
logger.info(`Node.js ${process.version}`);
// Configuration
const config = {
metaCorePath: process.env.META_CORE_PATH ?? '/meta-core',
filesPath: process.env.FILES_VOLUME ?? '/files',
redisUrl: process.env.REDIS_URL,
redisPrefix: process.env.REDIS_PREFIX ?? '',
apiPort: parseInt(process.env.API_PORT ?? '3000', 10),
apiHost: process.env.API_HOST ?? '0.0.0.0',
baseUrl: process.env.BASE_URL,
serviceVersion: process.env.SERVICE_VERSION ?? '1.0.0',
};
logger.info(`Config: META_CORE_PATH=${config.metaCorePath}`);
// Initialize KV Manager (handles leader discovery)
const kvManager = new KVManager({
metaCorePath: config.metaCorePath,
filesPath: config.filesPath,
serviceName: 'meta-dup',
apiPort: config.apiPort,
baseUrl: config.baseUrl,
redisPrefix: config.redisPrefix,
});
let redisClient: RedisClient | null = null;
let duplicateIndex: DuplicateIndex | null = null;
let apiServer: APIServer | null = null;
// Graceful shutdown
const shutdown = async (signal: string): Promise<void> => {
logger.info(`Received ${signal}, shutting down...`);
try {
// Stop stream consumer first
if (redisClient) {
redisClient.stopStreamConsumer();
}
if (apiServer) await apiServer.stop();
await kvManager.stop();
logger.info('Shutdown complete');
process.exit(0);
} catch (error) {
logger.error('Error during shutdown:', error);
process.exit(1);
}
};
process.on('SIGTERM', () => shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT'));
// Handle ready event from KV manager
kvManager.onReady(async () => {
logger.info('KV Manager ready, initializing DuplicateIndex...');
try {
// Get Redis client from KV manager
redisClient = kvManager.getClient();
if (!redisClient) {
logger.error('Redis client not available');
return;
}
// Initialize DuplicateIndex
duplicateIndex = new DuplicateIndex();
// Rebuild index from all file:* keys in Redis
logger.info('Building initial duplicate index from Redis...');
await duplicateIndex.rebuildFromRedis(redisClient);
// Initialize Redis Streams consumer
logger.info(`Initializing stream consumer for ${EVENTS_STREAM}...`);
await redisClient.initStreamConsumer(EVENTS_STREAM, CONSUMER_GROUP);
// Process any pending entries from crashed consumers
await redisClient.processPendingEntries(
EVENTS_STREAM,
CONSUMER_GROUP,
30000, // 30 second idle threshold
async (message: StreamMessage) => {
await handleStreamMessage(message, duplicateIndex!, redisClient!);
}
);
// Start stream consumer in background
logger.info('Starting stream consumer...');
redisClient.startStreamConsumer(
EVENTS_STREAM,
CONSUMER_GROUP,
async (message: StreamMessage) => {
await handleStreamMessage(message, duplicateIndex!, redisClient!);
},
5000 // 5 second block timeout
).catch(error => {
logger.error('Stream consumer error:', error);
});
// Initialize API server
apiServer = new APIServer(duplicateIndex, {
port: config.apiPort,
host: config.apiHost,
}, kvManager);
// Set Redis client for rebuild endpoint
apiServer.setRedisClient(redisClient);
// Start API server
logger.info('Starting API server...');
await apiServer.start();
logger.info('meta-dup is ready!');
logger.info(`API: http://${config.apiHost}:${config.apiPort}`);
// Log initial stats
const stats = duplicateIndex.getStats();
logger.info(`Duplicates: ${stats.hashGroupCount} hash groups, ${stats.titleGroupCount} title groups`);
logger.info(`Total files tracked: ${stats.totalFilesTracked}`);
} catch (error) {
logger.error('Failed to initialize after KV ready:', error);
}
});
// Handle disconnect event
kvManager.onDisconnect(() => {
logger.warn('Redis disconnected, duplicate index will use cached data');
});
// Start KV Manager (initiates leader discovery)
try {
logger.info('Starting KV Manager...');
await kvManager.start();
// Wait for ready with timeout
await kvManager.waitForReady(60000);
} catch (error) {
logger.error('Failed to start meta-dup:', error);
process.exit(1);
}
}
main().catch((error) => {
logger.error('Fatal error:', error);
process.exit(1);
});
packages/meta-dup-core/src/kv/IKVClient.ts
+109
View File
@@ -0,0 +1,109 @@
/**
* KV Client Interface - Read-only abstraction for meta-dup
*
* meta-dup is a FOLLOWER service that only reads metadata from Redis.
* This interface provides the subset of operations needed for reading.
*/
export interface KeyValuePair {
key: string;
value: string;
}
export interface IKVClient {
/**
* Get a value by key (value is JSON deserialized)
*/
get(key: string): Promise<any | null>;
/**
* Count keys with a given prefix
*/
countKeysWithPrefix(prefix: string): Promise<number>;
/**
* Health check - verifies KV store is accessible
*/
health(): Promise<boolean>;
/**
* Get all key-value pairs with a given prefix
*/
getRange(prefix: string): Promise<KeyValuePair[]>;
/**
* Retrieve file metadata - reconstructs from individual property keys
*/
getMetadataFlat(hashId: string): Promise<any | null>;
/**
* Get all unique hash IDs (files) stored in KV
*/
getAllHashIds(): Promise<string[]>;
/**
* Subscribe to Redis pub/sub channel
*/
subscribe(channel: string, callback: (message: string) => void): Promise<void>;
/**
* Unsubscribe from Redis pub/sub channel
*/
unsubscribe(channel: string): Promise<void>;
/**
* Close connections and cleanup resources
*/
close(): Promise<void>;
}
/**
* Leader lock file content format (kv-leader.info)
* Written by meta-core leader, read by other services
*/
export interface LeaderLockInfo {
/** Hostname of the leader */
hostname: string;
/** Base URL for the leader service (e.g., http://localhost:8180) */
baseUrl: string;
/** meta-core API URL (port 9000) */
apiUrl: string;
/** Redis connection URL (e.g., redis://10.0.1.50:6379) */
redisUrl: string;
/** WebDAV URL for file access */
webdavUrl: string;
/** Timestamp when leadership was acquired */
timestamp: number;
/** Process ID of the leader */
pid: number;
}
/**
* Service registration info (simplified)
* Full URLs are obtained via the /urls API endpoint
*/
export interface ServiceInfo {
/** Service name (e.g., 'meta-sort', 'meta-fuse') */
name: string;
/** Hostname */
hostname: string;
/** Base URL for the service */
baseUrl: string;
/** Current status */
status: 'running' | 'stale' | 'stopped';
/** Last heartbeat timestamp (ISO string) */
lastHeartbeat: string;
/** Role for meta-core instances: 'leader', 'follower', or undefined for other services */
role?: string;
}
packages/meta-dup-core/src/kv/KVManager.ts
+359
View File
@@ -0,0 +1,359 @@
/**
* KV Manager for meta-dup (FOLLOWER-only)
*
* Simplified KV manager that:
* 1. Discovers the leader via LeaderClient (reads kv-leader.info)
* 2. Creates Redis client connection
* 3. Handles reconnection on leader failure
*
* Unlike meta-sort's KVManager, this version never spawns Redis -
* it only connects as a client. Leader election is handled by meta-core.
*/
import { hostname } from 'os';
import { Logger } from 'tslog';
import { LeaderClient } from './LeaderClient.js';
import { ServiceDiscovery } from './ServiceDiscovery.js';
import { RedisClient } from './RedisClient.js';
import type { LeaderLockInfo } from './IKVClient.js';
const logger = new Logger({ name: 'KVManager' });
interface KVManagerConfig {
/** Path to META_CORE_VOLUME (e.g., /meta-core) */
metaCorePath: string;
/** Path to FILES_VOLUME (e.g., /files) */
filesPath: string;
/** Service name (for logging) */
serviceName?: string;
/** HTTP API port */
apiPort?: number;
/** Override Redis URL (skip leader discovery if provided) */
redisUrl?: string;
/** Redis key prefix (default: 'meta-sort:') */
redisPrefix?: string;
/** Base URL for service discovery (overrides auto-detected URL) */
baseUrl?: string;
/** meta-core API URL (e.g., http://localhost:9000) */
metaCoreUrl?: string;
}
export class KVManager {
private config: Required<KVManagerConfig>;
private leaderClient: LeaderClient | null = null;
private serviceDiscovery: ServiceDiscovery | null = null;
private redisClient: RedisClient | null = null;
private isStarted = false;
private isShuttingDown = false;
// Event callbacks
private onReadyCallbacks: (() => void)[] = [];
private onDisconnectCallbacks: (() => void)[] = [];
constructor(config: KVManagerConfig) {
this.config = {
serviceName: 'meta-dup',
apiPort: 3000,
redisUrl: '',
redisPrefix: '',
baseUrl: '',
metaCoreUrl: '',
...config
};
}
/**
* Start the KV manager
*/
async start(): Promise<void> {
if (this.isStarted) {
logger.warn('KVManager already started');
return;
}
logger.info(`Starting KVManager for ${this.config.serviceName}...`);
// Initialize and start service discovery for registration and heartbeat
const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`;
this.serviceDiscovery = new ServiceDiscovery({
metaCorePath: this.config.metaCorePath,
serviceName: this.config.serviceName,
apiUrl
});
await this.serviceDiscovery.start();
// If direct Redis URL provided, skip leader discovery
if (this.config.redisUrl) {
logger.info(`Using direct Redis URL: ${this.config.redisUrl}`);
await this.connectToRedis(this.config.redisUrl);
this.isStarted = true;
return;
}
// Use leader client
this.leaderClient = new LeaderClient({
metaCorePath: this.config.metaCorePath,
metaCoreUrl: this.config.metaCoreUrl || undefined
});
// Set up leader change callback
this.leaderClient.onChange(async () => {
logger.info('Leader changed, reconnecting...');
await this.reconnect();
});
// Wait for leader and connect
try {
const leaderInfo = await this.leaderClient.waitForLeader(30000);
logger.info(`Connecting to Redis at ${leaderInfo.redisUrl}...`);
await this.connectToRedis(leaderInfo.redisUrl);
} catch (error: any) {
logger.error(`Failed to connect to leader: ${error.message}`);
throw error;
}
// Start watching for leader changes
this.leaderClient.startWatching();
this.isStarted = true;
}
/**
* Reconnect to Redis after leader change
*/
private async reconnect(): Promise<void> {
if (this.isShuttingDown || !this.leaderClient) {
return;
}
// Disconnect existing
await this.disconnectRedis();
this.notifyDisconnect();
// Wait for new leader and connect
try {
const leaderInfo = await this.leaderClient.waitForLeader(30000);
logger.info(`Reconnecting to Redis at ${leaderInfo.redisUrl}...`);
await this.connectToRedis(leaderInfo.redisUrl);
} catch (error: any) {
logger.error(`Failed to reconnect: ${error.message}`);
}
}
/**
* Stop the KV manager
*/
async stop(): Promise<void> {
if (!this.isStarted) return;
logger.info('Stopping KVManager...');
this.isShuttingDown = true;
// Stop leader client
if (this.leaderClient) {
this.leaderClient.close();
this.leaderClient = null;
}
// Stop service discovery (if it was started)
if (this.serviceDiscovery) {
await this.serviceDiscovery.stop();
this.serviceDiscovery = null;
}
// Disconnect Redis
await this.disconnectRedis();
this.isStarted = false;
logger.info('KVManager stopped');
}
/**
* Connect to Redis
*/
private async connectToRedis(url: string): Promise<void> {
try {
// Disconnect existing client if any
await this.disconnectRedis();
// Create new client
this.redisClient = new RedisClient({
url,
prefix: this.config.redisPrefix,
filesVolume: this.config.filesPath
});
await this.redisClient.connect();
logger.info('Connected to Redis');
// Notify ready
this.notifyReady();
} catch (error: any) {
logger.error(`Failed to connect to Redis: ${error.message}`);
this.redisClient = null;
}
}
/**
* Disconnect from Redis
*/
private async disconnectRedis(): Promise<void> {
if (this.redisClient) {
try {
await this.redisClient.disconnect();
} catch (error: any) {
logger.warn(`Error disconnecting Redis: ${error.message}`);
}
this.redisClient = null;
}
}
/**
* Notify ready callbacks
*/
private notifyReady(): void {
for (const callback of this.onReadyCallbacks) {
try {
callback();
} catch (error: any) {
logger.error(`Error in ready callback: ${error.message}`);
}
}
}
/**
* Notify disconnect callbacks
*/
private notifyDisconnect(): void {
for (const callback of this.onDisconnectCallbacks) {
try {
callback();
} catch (error: any) {
logger.error(`Error in disconnect callback: ${error.message}`);
}
}
}
// ========================================================================
// Event Registration
// ========================================================================
/**
* Register callback for when KV client is ready
*/
onReady(callback: () => void): this {
this.onReadyCallbacks.push(callback);
// If already ready, call immediately
if (this.redisClient?.connected) {
callback();
}
return this;
}
/**
* Register callback for disconnection
*/
onDisconnect(callback: () => void): this {
this.onDisconnectCallbacks.push(callback);
return this;
}
/**
* Wait for KV client to be ready
*/
async waitForReady(timeoutMs: number = 30000): Promise<void> {
if (this.redisClient?.connected) return;
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
reject(new Error(`KV client not ready after ${timeoutMs}ms`));
}, timeoutMs);
this.onReady(() => {
clearTimeout(timeout);
resolve();
});
});
}
// ========================================================================
// Getters
// ========================================================================
/**
* Get the Redis client (may be null if not connected)
*/
getClient(): RedisClient | null {
return this.redisClient;
}
/**
* Get the Redis client (throws if not connected)
*/
requireClient(): RedisClient {
if (!this.redisClient) {
throw new Error('Redis client not available. Ensure start() was called and leader is available.');
}
return this.redisClient;
}
/**
* Check if connected
*/
isConnected(): boolean {
return this.redisClient?.connected ?? false;
}
/**
* Health check
*/
async isHealthy(): Promise<boolean> {
if (!this.redisClient) return false;
try {
const stats = await this.redisClient.getStats();
return stats.connected;
} catch {
return false;
}
}
/**
* Get leader info
*/
getLeaderInfo(): LeaderLockInfo | null {
return this.leaderClient?.getCachedLeaderInfo() ?? null;
}
/**
* Get leader client
*/
getLeaderClient(): LeaderClient | null {
return this.leaderClient;
}
/**
* Get service discovery (for inter-service navigation)
*/
getServiceDiscovery(): ServiceDiscovery | null {
return this.serviceDiscovery;
}
/**
* Get configured paths
*/
getPaths() {
return {
metaCorePath: this.config.metaCorePath,
filesPath: this.config.filesPath
};
}
}
packages/meta-dup-core/src/kv/LeaderClient.ts
+279
View File
@@ -0,0 +1,279 @@
/**
* LeaderClient - Client for reading leader info from meta-core
*
* This replaces the old LeaderDiscovery.ts by delegating leader election
* to meta-core (Go sidecar) and just reading the results.
*
* Features:
* - Reads kv-leader.info file for leader URLs
* - Calls meta-core /urls API for current leader URLs
* - Watches for leader changes via file system watcher
*/
import { promises as fs } from 'fs';
import { watch, FSWatcher } from 'fs';
import { dirname } from 'path';
import { Logger } from 'tslog';
import type { LeaderLockInfo } from './IKVClient.js';
const logger = new Logger({ name: 'LeaderClient' });
export interface LeaderClientConfig {
/** Path to META_CORE_VOLUME (e.g., /meta-core) */
metaCorePath: string;
/** meta-core API URL (e.g., http://meta-core:9000 or http://localhost:9000) */
metaCoreUrl?: string;
}
export interface URLsResponse {
hostname: string;
baseUrl: string;
apiUrl: string;
redisUrl: string;
webdavUrl: string;
isLeader: boolean;
}
export class LeaderClient {
private config: LeaderClientConfig;
private infoFilePath: string;
private leaderInfo: LeaderLockInfo | null = null;
private watcher: FSWatcher | null = null;
private onChangeCallbacks: (() => void)[] = [];
// URL caching
private cachedUrls: URLsResponse | null = null;
private urlsCacheTime: number = 0;
private readonly urlsCacheTTL: number = 5000; // 5 seconds
constructor(config: LeaderClientConfig) {
this.config = config;
this.infoFilePath = `${config.metaCorePath}/locks/kv-leader.info`;
}
/**
* Read API URL from file (plain text format)
*/
private async getApiUrlFromFile(): Promise<string | null> {
try {
const content = await fs.readFile(this.infoFilePath, 'utf-8');
return content.trim() || null;
} catch (error: any) {
logger.error(`Failed to read API URL from file: ${error.message}`);
return null;
}
}
/**
* Fetch URLs from meta-core /urls API with caching
*/
private async fetchUrls(apiUrl: string): Promise<URLsResponse | null> {
// Check cache
const now = Date.now();
if (this.cachedUrls && (now - this.urlsCacheTime) < this.urlsCacheTTL) {
return this.cachedUrls;
}
try {
const response = await fetch(`${apiUrl}/urls`, {
method: 'GET',
headers: { 'Accept': 'application/json' },
signal: AbortSignal.timeout(5000)
});
if (!response.ok) {
logger.error(`Failed to fetch URLs: ${response.status} ${response.statusText}`);
return null;
}
this.cachedUrls = await response.json() as URLsResponse;
this.urlsCacheTime = now;
return this.cachedUrls;
} catch (error: any) {
logger.error(`Error calling /urls API: ${error.message}`);
return null;
}
}
/**
* Read leader info from file and /urls API
*/
async getLeaderInfo(): Promise<LeaderLockInfo | null> {
try {
// Read API URL from file (plain text)
const apiUrl = await this.getApiUrlFromFile();
if (!apiUrl) {
return null;
}
// Fetch full info from /urls API
const urls = await this.fetchUrls(apiUrl);
if (!urls) {
return null;
}
// Convert URLsResponse to LeaderLockInfo
this.leaderInfo = {
hostname: urls.hostname,
baseUrl: urls.baseUrl,
apiUrl: urls.apiUrl,
redisUrl: urls.redisUrl,
webdavUrl: urls.webdavUrl,
timestamp: Date.now(),
pid: 0 // Unknown for remote leader
};
return this.leaderInfo;
} catch (error: any) {
logger.error(`Failed to read leader info: ${error.message}`);
return null;
}
}
/**
* Get Redis URL from leader info
*/
async getRedisUrl(): Promise<string | null> {
const info = await this.getLeaderInfo();
return info?.redisUrl ?? null;
}
/**
* Get WebDAV URL from leader info
*/
async getWebdavUrl(): Promise<string | null> {
const info = await this.getLeaderInfo();
return info?.webdavUrl ?? null;
}
/**
* Get meta-core API URL from leader info
*/
async getApiUrl(): Promise<string | null> {
const info = await this.getLeaderInfo();
return info?.apiUrl ?? null;
}
/**
* Call meta-core /urls API to get current URLs
* Useful for getting URLs when you don't want to read the file directly
*/
async getUrls(): Promise<URLsResponse | null> {
// First try using configured metaCoreUrl
let apiUrl: string | null = this.config.metaCoreUrl ?? null;
// Fall back to reading from file
if (!apiUrl) {
apiUrl = await this.getApiUrlFromFile();
}
if (!apiUrl) {
logger.error('No meta-core API URL available');
return null;
}
return this.fetchUrls(apiUrl);
}
/**
* Wait for leader info to be available
*/
async waitForLeader(timeoutMs: number = 30000): Promise<LeaderLockInfo> {
const startTime = Date.now();
const pollInterval = 500;
while (Date.now() - startTime < timeoutMs) {
const info = await this.getLeaderInfo();
if (info) {
logger.info(`Leader found: ${info.hostname} at ${info.redisUrl}`);
return info;
}
logger.debug('Waiting for leader...');
await new Promise(resolve => setTimeout(resolve, pollInterval));
}
throw new Error(`No leader found within ${timeoutMs}ms`);
}
/**
* Start watching for leader changes
*/
startWatching(): void {
if (this.watcher) {
return; // Already watching
}
const lockDir = dirname(this.infoFilePath);
try {
this.watcher = watch(lockDir, (eventType, filename) => {
if (filename === 'kv-leader.info') {
logger.debug(`Lock file ${eventType}, invalidating cache...`);
// Invalidate cache to force fresh API call
this.cachedUrls = null;
this.urlsCacheTime = 0;
this.getLeaderInfo().then(() => {
this.notifyChange();
}).catch((err) => logger.error(`Error reloading leader info: ${err.message}`));
}
});
this.watcher.on('error', (error: any) => {
logger.warn(`File watcher error: ${error.message}`);
});
logger.info(`Watching for leader changes in ${lockDir}`);
} catch (error: any) {
logger.error(`Failed to start watching: ${error.message}`);
}
}
/**
* Stop watching for leader changes
*/
stopWatching(): void {
if (this.watcher) {
this.watcher.close();
this.watcher = null;
logger.debug('Stopped watching for leader changes');
}
}
/**
* Register callback for leader changes
*/
onChange(callback: () => void): this {
this.onChangeCallbacks.push(callback);
return this;
}
/**
* Notify all change callbacks
*/
private notifyChange(): void {
for (const callback of this.onChangeCallbacks) {
try {
callback();
} catch (error: any) {
logger.error(`Error in change callback: ${error.message}`);
}
}
}
/**
* Get cached leader info (without re-reading file)
*/
getCachedLeaderInfo(): LeaderLockInfo | null {
return this.leaderInfo;
}
/**
* Clean up resources
*/
close(): void {
this.stopWatching();
this.onChangeCallbacks = [];
}
}
packages/meta-dup-core/src/kv/RedisClient.ts
+733
View File
@@ -0,0 +1,733 @@
/**
* Redis Client for meta-dup
* 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 Streams 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';
const logger = new Logger({ name: 'RedisClient' });
/**
* Stream message from Redis Streams
* Supports both:
* - Direct file events from meta-core (add/change/delete/rename)
* - Legacy batch events from meta-sort (batch/reset/plugin:complete)
*/
export interface StreamMessage {
id: string;
type: 'add' | 'change' | 'delete' | 'rename' | 'batch' | 'reset' | 'plugin:complete';
// Direct file event fields (from meta-core)
path?: string;
size?: string;
partialHash?: string;
oldPath?: string;
// Legacy payload field (for backward compatibility)
payload?: string;
timestamp: string;
}
/**
* Stream consumer callback
*/
export type StreamMessageHandler = (message: StreamMessage) => Promise<void>;
/**
* File metadata structure for duplicate detection
*/
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
sha256?: string; // Full SHA-256 hash for duplicate detection
// Title information (for folder organization)
title?: string;
titles?: { eng?: string; [key: string]: string | undefined };
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;
}
export class RedisClient implements Partial<IKVClient> {
private client: Redis | null = null;
private subscriber: Redis | null = null;
private config: Required<RedisClientConfig>;
private isConnected = false;
private reconnectTimer: NodeJS.Timeout | null = null;
private subscriptions: Map<string, (message: string) => void> = new Map();
// 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,
};
// Unique consumer name for this instance
this.consumerName = `${os.hostname()}-${process.pid}`;
}
/**
* Connect to Redis
*/
async connect(): Promise<void> {
if (this.client && this.isConnected) {
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 connected
*/
get connected(): boolean {
return this.isConnected;
}
/**
* Get all file metadata from Redis
*/
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(hashId, 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(hashId, 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
*/
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 [];
}
}
/**
* 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 titles - can be nested object or simple string
let titles: { eng?: string; [key: string]: string | undefined } | undefined;
if (data['titles/eng']) {
titles = { eng: data['titles/eng'] };
}
// 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: hashId ?? data.cid_midhash256 ?? data.hashId,
sha256: data['cid_sha2-256'] ?? data.sha256,
// Title information
title: data.title,
titles,
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,
};
}
// ========================================================================
// 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., 'meta-sort:events')
* @param group - Consumer group name (e.g., 'meta-dup-consumer')
*/
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)
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;
}
/**
* Parse a stream entry into a StreamMessage
* Handles both direct file events (add/change/delete/rename) and legacy batch events
*/
private parseStreamEntry(id: string, fields: string[]): StreamMessage | null {
// Fields come as flat array: ['type', 'add', 'path', '...', 'timestamp', '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;
}
// 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;
}
if (!isDirectEvent && !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,
partialHash: fieldMap.partialHash,
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;
}
}
packages/meta-dup-core/src/kv/ServiceDiscovery.ts
+331
View File
@@ -0,0 +1,331 @@
/**
* Service Discovery using shared filesystem
*
* Each service registers itself in /meta-core/services/{service-name}-{hostname}.json
* with its base URL, status, and heartbeat timestamp.
*
* Services can discover each other by reading these JSON files.
* Stale detection: if lastHeartbeat > 60 seconds ago, service is considered stale.
*
* Each service must call start() to register itself and begin heartbeat updates.
* meta-core runs a cleanup process that removes services with stale heartbeats.
*/
import { promises as fs } from 'fs';
import { hostname } from 'os';
import { join } from 'path';
import { Logger } from 'tslog';
import type { ServiceInfo } from './IKVClient.js';
const logger = new Logger({ name: 'ServiceDiscovery' });
interface ServiceDiscoveryConfig {
/** Path to META_CORE_VOLUME (e.g., /meta-core) */
metaCorePath: string;
/** Service name */
serviceName: string;
/** Base URL for the service */
apiUrl: string;
/** Heartbeat interval in ms (default: 30000) */
heartbeatInterval?: number;
/** Stale threshold in ms (default: 60000) */
staleThreshold?: number;
}
export class ServiceDiscovery {
private config: ServiceDiscoveryConfig;
private servicesDir: string;
private serviceFilePath: string;
private heartbeatTimer: NodeJS.Timeout | null = null;
private isShuttingDown = false;
private isStarted = false;
private currentHostname: string;
constructor(config: ServiceDiscoveryConfig) {
this.config = {
heartbeatInterval: 30000,
staleThreshold: 60000,
...config
};
this.currentHostname = hostname();
this.servicesDir = `${this.config.metaCorePath}/services`;
// Use hostname-based file naming
this.serviceFilePath = `${this.servicesDir}/${this.config.serviceName}-${this.currentHostname}.json`;
}
/**
* Ensure services directory exists
*/
private async ensureServicesDir(): Promise<void> {
await fs.mkdir(this.servicesDir, { recursive: true });
}
/**
* Build service info for this service
*/
private buildServiceInfo(status: ServiceInfo['status']): ServiceInfo {
return {
name: this.config.serviceName,
hostname: this.currentHostname,
baseUrl: this.config.apiUrl,
status,
lastHeartbeat: new Date().toISOString()
};
}
/**
* Register this service by writing service info to the shared filesystem.
* meta-core's cleanup process monitors these files and removes stale entries.
*/
async register(): Promise<void> {
await this.ensureServicesDir();
const info = this.buildServiceInfo('running');
await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2));
logger.info(`Registered ${this.config.serviceName}-${this.currentHostname}`);
}
/**
* Update service status
*/
async updateStatus(status: ServiceInfo['status']): Promise<void> {
try {
const content = await fs.readFile(this.serviceFilePath, 'utf-8');
const info = JSON.parse(content) as ServiceInfo;
info.status = status;
info.lastHeartbeat = new Date().toISOString();
await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2));
} catch (error) {
logger.error('Failed to update status:', error);
}
}
/**
* Send heartbeat (update lastHeartbeat timestamp)
*/
async heartbeat(): Promise<void> {
try {
const content = await fs.readFile(this.serviceFilePath, 'utf-8');
const info = JSON.parse(content) as ServiceInfo;
info.lastHeartbeat = new Date().toISOString();
await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2));
} catch (error: any) {
// If file was deleted, re-register
if (error.code === 'ENOENT') {
await this.register();
} else {
logger.error('Heartbeat failed:', error);
}
}
}
/**
* Start heartbeat loop
*/
startHeartbeat(): void {
this.heartbeatTimer = setInterval(async () => {
if (this.isShuttingDown) return;
await this.heartbeat();
}, this.config.heartbeatInterval);
}
/**
* Stop heartbeat loop
*/
stopHeartbeat(): void {
if (this.heartbeatTimer) {
clearInterval(this.heartbeatTimer);
this.heartbeatTimer = null;
}
}
/**
* Unregister this service (on shutdown)
*/
async unregister(): Promise<void> {
this.isShuttingDown = true;
this.stopHeartbeat();
try {
// Remove the service file
await fs.unlink(this.serviceFilePath);
logger.info(`Unregistered ${this.config.serviceName}-${this.currentHostname}`);
} catch (error: any) {
if (error.code !== 'ENOENT') {
logger.error('Failed to unregister:', error);
}
}
}
/**
* Check if a service info is stale
*/
private isStale(info: ServiceInfo): boolean {
const lastHeartbeat = new Date(info.lastHeartbeat).getTime();
return Date.now() - lastHeartbeat > this.config.staleThreshold!;
}
// ========================================================================
// Discovery Methods
// ========================================================================
/**
* Discover a service by name
* Looks for files matching pattern: {name}-*.json (hostname-based naming)
*/
async discoverService(name: string): Promise<ServiceInfo | null> {
try {
// First try exact match for backward compatibility
const exactPath = join(this.servicesDir, `${name}.json`);
try {
const content = await fs.readFile(exactPath, 'utf-8');
const service = JSON.parse(content) as ServiceInfo;
if (this.isStale(service)) {
service.status = 'stale';
}
return service.status === 'running' ? service : null;
} catch {
// Try hostname-based files
}
// Search for hostname-based files: name-*.json
const files = await fs.readdir(this.servicesDir);
const matchingFiles = files.filter(f =>
f.startsWith(`${name}-`) && f.endsWith('.json')
);
// Return first valid (non-stale) service
for (const file of matchingFiles) {
const filePath = join(this.servicesDir, file);
const content = await fs.readFile(filePath, 'utf-8');
const service = JSON.parse(content) as ServiceInfo;
if (this.isStale(service)) {
service.status = 'stale';
continue; // Skip stale services
}
if (service.status === 'running') {
return service;
}
}
return null;
} catch (error: any) {
if (error.code === 'ENOENT') {
return null; // Service not registered
}
throw error;
}
}
/**
* Discover all registered services
*/
async discoverAllServices(): Promise<ServiceInfo[]> {
try {
const files = await fs.readdir(this.servicesDir);
const services: ServiceInfo[] = [];
for (const file of files) {
if (!file.endsWith('.json')) continue;
const filePath = join(this.servicesDir, file);
try {
const content = await fs.readFile(filePath, 'utf-8');
const service = JSON.parse(content) as ServiceInfo;
if (this.isStale(service)) {
service.status = 'stale';
}
services.push(service);
} catch (error) {
logger.error(`Failed to read ${file}:`, error);
}
}
return services;
} catch (error: any) {
if (error.code === 'ENOENT') {
return []; // Services directory doesn't exist yet
}
throw error;
}
}
/**
* Check if a service is healthy
*/
async isServiceHealthy(name: string): Promise<boolean> {
const service = await this.discoverService(name);
if (!service) return false;
// Try to ping the service's health endpoint
const url = `${service.baseUrl}/health`;
try {
const response = await fetch(url, {
signal: AbortSignal.timeout(5000)
});
return response.ok;
} catch {
return false;
}
}
// ========================================================================
// Lifecycle
// ========================================================================
/**
* Full startup sequence: register service and start heartbeat loop.
*/
async start(): Promise<void> {
if (this.isStarted) return;
await this.register();
this.startHeartbeat();
this.isStarted = true;
logger.info(`Started ${this.config.serviceName} at ${this.config.apiUrl}`);
}
/**
* Full shutdown sequence
*/
async stop(): Promise<void> {
if (!this.isStarted) return;
await this.unregister();
this.isStarted = false;
}
// ========================================================================
// Getters
// ========================================================================
getServiceFilePath(): string {
return this.serviceFilePath;
}
getServicesDir(): string {
return this.servicesDir;
}
getIsStarted(): boolean {
return this.isStarted;
}
}
packages/meta-dup-core/tsconfig.json
+20
View File
@@ -0,0 +1,20 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "NodeNext",
"moduleResolution": "NodeNext",
"lib": ["ES2022"],
"outDir": "./dist",
"rootDir": "./src",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"declaration": true,
"declarationMap": true,
"sourceMap": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist"]
}