initial commit
This commit is contained in:
Pierre
@@ -0,0 +1,113 @@
|
||||
/**
|
||||
* KV Client Interface - Read-only abstraction for meta-fuse
|
||||
*
|
||||
* meta-fuse 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
|
||||
*/
|
||||
export interface LeaderLockInfo {
|
||||
/** Hostname of the leader */
|
||||
host: string;
|
||||
|
||||
/** Redis connection URL (e.g., redis://10.0.1.50:6379) */
|
||||
api: string;
|
||||
|
||||
/** HTTP API URL for the leader service */
|
||||
http: string;
|
||||
|
||||
/** Timestamp when leadership was acquired */
|
||||
timestamp: number;
|
||||
|
||||
/** Process ID of the leader */
|
||||
pid: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Service registration info stored in JSON file
|
||||
*/
|
||||
export interface ServiceInfo {
|
||||
/** Service name */
|
||||
name: string;
|
||||
|
||||
/** Service version */
|
||||
version: string;
|
||||
|
||||
/** HTTP API URL (e.g., http://10.0.1.50:3000) */
|
||||
api: string;
|
||||
|
||||
/** Service status */
|
||||
status: 'starting' | 'running' | 'stopping' | 'stopped';
|
||||
|
||||
/** Process ID */
|
||||
pid: number;
|
||||
|
||||
/** Hostname */
|
||||
hostname: string;
|
||||
|
||||
/** Service start time (ISO string) */
|
||||
startedAt: string;
|
||||
|
||||
/** Last heartbeat time (ISO string) */
|
||||
lastHeartbeat: string;
|
||||
|
||||
/** Service capabilities */
|
||||
capabilities: string[];
|
||||
|
||||
/** Named endpoints */
|
||||
endpoints: Record<string, string>;
|
||||
}
|
||||
@@ -0,0 +1,338 @@
|
||||
/**
|
||||
* KV Manager for meta-fuse (FOLLOWER-only)
|
||||
*
|
||||
* Simplified KV manager that:
|
||||
* 1. Discovers the leader via lock file
|
||||
* 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.
|
||||
*/
|
||||
|
||||
import { Logger } from 'tslog';
|
||||
import { LeaderDiscovery } from './LeaderDiscovery.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;
|
||||
|
||||
/** Service version */
|
||||
version?: 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;
|
||||
|
||||
/** Service capabilities */
|
||||
capabilities?: string[];
|
||||
}
|
||||
|
||||
export class KVManager {
|
||||
private config: Required<KVManagerConfig>;
|
||||
private leaderDiscovery: LeaderDiscovery | 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-fuse',
|
||||
version: '1.0.0',
|
||||
apiPort: 3000,
|
||||
redisUrl: '',
|
||||
redisPrefix: 'meta-sort:',
|
||||
baseUrl: '',
|
||||
capabilities: ['read', 'vfs'],
|
||||
...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 service discovery
|
||||
const apiUrl = this.config.baseUrl || `http://localhost:${this.config.apiPort}`;
|
||||
this.serviceDiscovery = new ServiceDiscovery({
|
||||
metaCorePath: this.config.metaCorePath,
|
||||
serviceName: this.config.serviceName,
|
||||
version: this.config.version,
|
||||
apiUrl,
|
||||
baseUrl: this.config.baseUrl || undefined,
|
||||
capabilities: this.config.capabilities,
|
||||
endpoints: {
|
||||
health: '/api/fuse/health',
|
||||
dashboard: '/',
|
||||
stats: '/api/fuse/stats',
|
||||
}
|
||||
});
|
||||
|
||||
// Start service discovery
|
||||
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 discovery
|
||||
this.leaderDiscovery = new LeaderDiscovery({
|
||||
metaCorePath: this.config.metaCorePath
|
||||
});
|
||||
|
||||
// Set up leader discovery callbacks
|
||||
this.leaderDiscovery.onLeaderFound(async (info: LeaderLockInfo) => {
|
||||
logger.info(`Connecting to leader at ${info.api}...`);
|
||||
await this.connectToRedis(info.api);
|
||||
});
|
||||
|
||||
this.leaderDiscovery.onLeaderLost(async () => {
|
||||
logger.warn('Leader lost, disconnecting...');
|
||||
await this.disconnectRedis();
|
||||
this.notifyDisconnect();
|
||||
});
|
||||
|
||||
// Start leader discovery
|
||||
await this.leaderDiscovery.start();
|
||||
|
||||
this.isStarted = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the KV manager
|
||||
*/
|
||||
async stop(): Promise<void> {
|
||||
if (!this.isStarted) return;
|
||||
|
||||
logger.info('Stopping KVManager...');
|
||||
this.isShuttingDown = true;
|
||||
|
||||
// Stop service discovery
|
||||
if (this.serviceDiscovery) {
|
||||
await this.serviceDiscovery.stop();
|
||||
this.serviceDiscovery = null;
|
||||
}
|
||||
|
||||
// Stop leader discovery
|
||||
if (this.leaderDiscovery) {
|
||||
await this.leaderDiscovery.stop();
|
||||
this.leaderDiscovery = 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 (if using leader discovery)
|
||||
*/
|
||||
getLeaderInfo(): LeaderLockInfo | null {
|
||||
return this.leaderDiscovery?.getLeaderInfo() ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,276 @@
|
||||
/**
|
||||
* Leader Discovery for meta-fuse (FOLLOWER-only)
|
||||
*
|
||||
* This is a simplified version of leader election that only discovers
|
||||
* the existing leader. meta-fuse never spawns Redis - it only connects
|
||||
* to the leader's Redis instance.
|
||||
*
|
||||
* Architecture:
|
||||
* - Leader holds flock on META_CORE_PATH/locks/kv-leader.lock
|
||||
* - Leader writes info to META_CORE_PATH/locks/kv-leader.info
|
||||
* - This class reads the info file to discover the leader
|
||||
* - Watches for leader changes via health checks
|
||||
* - Triggers reconnection callbacks on leader failure/change
|
||||
*/
|
||||
|
||||
import { promises as fs, watch } from 'fs';
|
||||
import { dirname } from 'path';
|
||||
import { Logger } from 'tslog';
|
||||
import type { LeaderLockInfo } from './IKVClient.js';
|
||||
|
||||
const logger = new Logger({ name: 'LeaderDiscovery' });
|
||||
|
||||
interface LeaderDiscoveryConfig {
|
||||
/** Path to META_CORE_VOLUME (e.g., /meta-core) */
|
||||
metaCorePath: string;
|
||||
|
||||
/** Health check interval in ms (default: 5000) */
|
||||
healthCheckInterval?: number;
|
||||
|
||||
/** Max consecutive failures before marking leader as lost (default: 3) */
|
||||
maxFailures?: number;
|
||||
}
|
||||
|
||||
export class LeaderDiscovery {
|
||||
private config: Required<LeaderDiscoveryConfig>;
|
||||
private lockFilePath: string;
|
||||
private leaderInfo: LeaderLockInfo | null = null;
|
||||
private healthCheckTimer: NodeJS.Timeout | null = null;
|
||||
private fileWatcher: ReturnType<typeof watch> | null = null;
|
||||
private isStarted = false;
|
||||
private consecutiveFailures = 0;
|
||||
|
||||
// Event callbacks
|
||||
private onLeaderFoundCallback?: (info: LeaderLockInfo) => void;
|
||||
private onLeaderLostCallback?: () => void;
|
||||
|
||||
constructor(config: LeaderDiscoveryConfig) {
|
||||
this.config = {
|
||||
healthCheckInterval: 5000,
|
||||
maxFailures: 3,
|
||||
...config
|
||||
};
|
||||
|
||||
this.lockFilePath = `${this.config.metaCorePath}/locks/kv-leader.info`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Start watching for leader
|
||||
*/
|
||||
async start(): Promise<void> {
|
||||
if (this.isStarted) {
|
||||
logger.warn('LeaderDiscovery already started');
|
||||
return;
|
||||
}
|
||||
|
||||
logger.info(`Starting leader discovery, lock file: ${this.lockFilePath}`);
|
||||
|
||||
// Ensure lock directory exists (for file watching)
|
||||
await this.ensureLockDir();
|
||||
|
||||
// Try to read existing leader info
|
||||
await this.checkForLeader();
|
||||
|
||||
// Start file watcher for immediate detection of leader changes
|
||||
this.startFileWatcher();
|
||||
|
||||
// Start health check for ongoing monitoring
|
||||
this.startHealthCheck();
|
||||
|
||||
this.isStarted = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop watching
|
||||
*/
|
||||
async stop(): Promise<void> {
|
||||
if (!this.isStarted) return;
|
||||
|
||||
logger.info('Stopping leader discovery');
|
||||
|
||||
if (this.healthCheckTimer) {
|
||||
clearInterval(this.healthCheckTimer);
|
||||
this.healthCheckTimer = null;
|
||||
}
|
||||
|
||||
if (this.fileWatcher) {
|
||||
this.fileWatcher.close();
|
||||
this.fileWatcher = null;
|
||||
}
|
||||
|
||||
this.isStarted = false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ensure lock directory exists
|
||||
*/
|
||||
private async ensureLockDir(): Promise<void> {
|
||||
const lockDir = dirname(this.lockFilePath);
|
||||
try {
|
||||
await fs.mkdir(lockDir, { recursive: true });
|
||||
} catch (error: any) {
|
||||
if (error.code !== 'EEXIST') {
|
||||
logger.warn(`Could not create lock directory: ${error.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check for leader by reading lock file
|
||||
*/
|
||||
private async checkForLeader(): Promise<void> {
|
||||
try {
|
||||
const content = await fs.readFile(this.lockFilePath, 'utf-8');
|
||||
const info = JSON.parse(content) as LeaderLockInfo;
|
||||
|
||||
// Validate leader info
|
||||
if (info && info.api && info.http) {
|
||||
const isNew = !this.leaderInfo || this.leaderInfo.api !== info.api;
|
||||
this.leaderInfo = info;
|
||||
this.consecutiveFailures = 0;
|
||||
|
||||
if (isNew) {
|
||||
logger.info(`Leader found: ${info.host} at ${info.api}`);
|
||||
this.onLeaderFoundCallback?.(info);
|
||||
}
|
||||
}
|
||||
} catch (error: any) {
|
||||
if (error.code === 'ENOENT') {
|
||||
// Lock file doesn't exist - no leader yet
|
||||
if (this.leaderInfo) {
|
||||
logger.warn('Lock file removed, leader lost');
|
||||
this.handleLeaderLost();
|
||||
}
|
||||
} else {
|
||||
logger.warn(`Error reading lock file: ${error.message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if current leader is still healthy
|
||||
*/
|
||||
private async isLeaderHealthy(): Promise<boolean> {
|
||||
if (!this.leaderInfo) return false;
|
||||
|
||||
try {
|
||||
const response = await fetch(`${this.leaderInfo.http}/health`, {
|
||||
signal: AbortSignal.timeout(5000)
|
||||
});
|
||||
return response.ok;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle leader being lost
|
||||
*/
|
||||
private handleLeaderLost(): void {
|
||||
this.leaderInfo = null;
|
||||
this.onLeaderLostCallback?.();
|
||||
}
|
||||
|
||||
/**
|
||||
* Start watching lock file for changes
|
||||
*/
|
||||
private startFileWatcher(): void {
|
||||
try {
|
||||
const lockDir = dirname(this.lockFilePath);
|
||||
this.fileWatcher = watch(lockDir, async (eventType, filename) => {
|
||||
if (filename === 'kv-leader.info') {
|
||||
logger.debug(`Lock file ${eventType}, checking for leader...`);
|
||||
await this.checkForLeader();
|
||||
}
|
||||
});
|
||||
|
||||
this.fileWatcher.on('error', (error) => {
|
||||
logger.warn(`File watcher error: ${error.message}`);
|
||||
});
|
||||
} catch (error: any) {
|
||||
logger.warn(`Could not start file watcher: ${error.message}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Start health check loop
|
||||
*/
|
||||
private startHealthCheck(): void {
|
||||
this.healthCheckTimer = setInterval(async () => {
|
||||
if (!this.leaderInfo) {
|
||||
// No leader known, try to find one
|
||||
await this.checkForLeader();
|
||||
return;
|
||||
}
|
||||
|
||||
// Check if current leader is healthy
|
||||
const healthy = await this.isLeaderHealthy();
|
||||
|
||||
if (healthy) {
|
||||
this.consecutiveFailures = 0;
|
||||
} else {
|
||||
this.consecutiveFailures++;
|
||||
logger.warn(`Leader health check failed (${this.consecutiveFailures}/${this.config.maxFailures})`);
|
||||
|
||||
if (this.consecutiveFailures >= this.config.maxFailures) {
|
||||
logger.warn('Leader appears dead, marking as lost');
|
||||
this.handleLeaderLost();
|
||||
|
||||
// Try to find new leader
|
||||
await this.checkForLeader();
|
||||
}
|
||||
}
|
||||
}, this.config.healthCheckInterval);
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Event Registration
|
||||
// ========================================================================
|
||||
|
||||
/**
|
||||
* Register callback for when leader is found/changed
|
||||
*/
|
||||
onLeaderFound(callback: (info: LeaderLockInfo) => void): this {
|
||||
this.onLeaderFoundCallback = callback;
|
||||
|
||||
// If we already have leader info, call immediately
|
||||
if (this.leaderInfo) {
|
||||
callback(this.leaderInfo);
|
||||
}
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Register callback for when leader is lost
|
||||
*/
|
||||
onLeaderLost(callback: () => void): this {
|
||||
this.onLeaderLostCallback = callback;
|
||||
return this;
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Getters
|
||||
// ========================================================================
|
||||
|
||||
/**
|
||||
* Get current leader info
|
||||
*/
|
||||
getLeaderInfo(): LeaderLockInfo | null {
|
||||
return this.leaderInfo;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get Redis URL of current leader
|
||||
*/
|
||||
getRedisUrl(): string | null {
|
||||
return this.leaderInfo?.api ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a leader is known
|
||||
*/
|
||||
hasLeader(): boolean {
|
||||
return this.leaderInfo !== null;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,511 @@
|
||||
/**
|
||||
* 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 type { IKVClient, KeyValuePair } from './IKVClient.js';
|
||||
|
||||
const logger = new Logger({ name: 'RedisClient' });
|
||||
|
||||
/**
|
||||
* 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;
|
||||
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();
|
||||
|
||||
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 ?? 'meta-sort:',
|
||||
filesVolume: config.filesVolume ?? process.env.FILES_VOLUME ?? '/files',
|
||||
reconnectInterval: config.reconnectInterval ?? 5000,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 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> {
|
||||
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(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
|
||||
*/
|
||||
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,
|
||||
|
||||
// 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,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* 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}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,306 @@
|
||||
/**
|
||||
* Service Discovery using shared filesystem
|
||||
*
|
||||
* Each service registers itself in /meta-core/services/{service-name}.json
|
||||
* with its API endpoint, status, and heartbeat timestamp.
|
||||
*
|
||||
* Services can discover each other by reading these JSON files.
|
||||
* Stale detection: if lastHeartbeat > 60 seconds ago, service is considered dead.
|
||||
*/
|
||||
|
||||
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;
|
||||
|
||||
/** Service version */
|
||||
version: string;
|
||||
|
||||
/** HTTP API URL (can be overridden by baseUrl) */
|
||||
apiUrl?: string;
|
||||
|
||||
/** Base URL for service discovery (overrides auto-detected URL) */
|
||||
baseUrl?: string;
|
||||
|
||||
/** Service capabilities */
|
||||
capabilities?: string[];
|
||||
|
||||
/** Named endpoints */
|
||||
endpoints?: Record<string, 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;
|
||||
|
||||
constructor(config: ServiceDiscoveryConfig) {
|
||||
this.config = {
|
||||
heartbeatInterval: 30000,
|
||||
staleThreshold: 60000,
|
||||
capabilities: [],
|
||||
endpoints: {},
|
||||
...config
|
||||
};
|
||||
|
||||
this.servicesDir = `${this.config.metaCorePath}/services`;
|
||||
this.serviceFilePath = `${this.servicesDir}/${this.config.serviceName}.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,
|
||||
version: this.config.version,
|
||||
api: this.config.baseUrl || this.config.apiUrl || '',
|
||||
status,
|
||||
pid: process.pid,
|
||||
hostname: hostname(),
|
||||
startedAt: new Date().toISOString(),
|
||||
lastHeartbeat: new Date().toISOString(),
|
||||
capabilities: this.config.capabilities || [],
|
||||
endpoints: this.config.endpoints || {}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Register this service
|
||||
*/
|
||||
async register(): Promise<void> {
|
||||
await this.ensureServicesDir();
|
||||
|
||||
const info = this.buildServiceInfo('starting');
|
||||
await fs.writeFile(this.serviceFilePath, JSON.stringify(info, null, 2));
|
||||
|
||||
logger.info(`Registered ${this.config.serviceName}`);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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();
|
||||
await this.updateStatus('running');
|
||||
} 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 {
|
||||
// Update status to stopped before removing
|
||||
await this.updateStatus('stopped');
|
||||
|
||||
logger.info(`Unregistered ${this.config.serviceName}`);
|
||||
} catch (error) {
|
||||
logger.error('Failed to unregister:', error);
|
||||
}
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Discovery Methods
|
||||
// ========================================================================
|
||||
|
||||
/**
|
||||
* Discover a service by name
|
||||
*/
|
||||
async discoverService(name: string): Promise<ServiceInfo | null> {
|
||||
const servicePath = join(this.servicesDir, `${name}.json`);
|
||||
|
||||
try {
|
||||
const content = await fs.readFile(servicePath, 'utf-8');
|
||||
const service = JSON.parse(content) as ServiceInfo;
|
||||
|
||||
// Check if service is alive (heartbeat within threshold)
|
||||
const lastHeartbeat = new Date(service.lastHeartbeat).getTime();
|
||||
const isAlive = Date.now() - lastHeartbeat < this.config.staleThreshold!;
|
||||
|
||||
if (!isAlive) {
|
||||
logger.warn(`Service ${name} appears stale (last heartbeat: ${service.lastHeartbeat})`);
|
||||
return null;
|
||||
}
|
||||
|
||||
return service;
|
||||
} 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 name = file.replace('.json', '');
|
||||
const service = await this.discoverService(name);
|
||||
|
||||
if (service && service.status === 'running') {
|
||||
services.push(service);
|
||||
}
|
||||
}
|
||||
|
||||
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 healthEndpoint = service.endpoints.health || '/health';
|
||||
const url = `${service.api}${healthEndpoint}`;
|
||||
|
||||
try {
|
||||
const response = await fetch(url, {
|
||||
signal: AbortSignal.timeout(5000)
|
||||
});
|
||||
return response.ok;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Lifecycle
|
||||
// ========================================================================
|
||||
|
||||
/**
|
||||
* Full startup sequence
|
||||
*/
|
||||
async start(): Promise<void> {
|
||||
if (this.isStarted) return;
|
||||
|
||||
await this.register();
|
||||
await this.updateStatus('running');
|
||||
this.startHeartbeat();
|
||||
this.isStarted = true;
|
||||
|
||||
logger.info(`Started ${this.config.serviceName} at ${this.config.baseUrl || 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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user