interface HeartbeatRedisClient {
  ping(): Promise<string>;
  set(_key: string, _value: string, _mode: 'EX', _ttlSeconds: number): Promise<unknown>;
  del(_key: string): Promise<number>;
  zadd(_key: string, _score: number, _member: string): Promise<number>;
  zrem(_key: string, _member: string): Promise<number>;
  quit(): Promise<unknown>;
}

interface WorkerHeartbeatOptions {
  redis: HeartbeatRedisClient;
  checkDatabase: () => Promise<unknown>;
  key: string;
  checkQueueReady?: () => Promise<unknown>;
  checkRuntimeReady?: () => Promise<unknown>;
  ttlSeconds?: number;
  now?: () => number;
  checkEventLoop?: () => Promise<void>;
}

const WORKER_HEARTBEAT_REGISTRY_KEY = 'worker:heartbeats';

export function getWorkerHeartbeatKey(
  workerId: string | undefined = process.env.WORKER_ID || process.env.HOSTNAME
): string {
  const normalizedWorkerId = workerId?.trim();
  if (!normalizedWorkerId) {
    throw new Error('WORKER_ID or HOSTNAME is required for worker heartbeat isolation');
  }
  return `worker:health:${normalizedWorkerId}`;
}

function checkEventLoopResponsiveness(maxDelayMs = 1000): Promise<void> {
  const startedAt = Date.now();

  return new Promise((resolve, reject) => {
    setTimeout(() => {
      const delay = Date.now() - startedAt;
      if (delay > maxDelayMs) {
        reject(new Error(`Worker event loop delay exceeded ${maxDelayMs}ms`));
        return;
      }
      resolve();
    }, 0);
  });
}

export class WorkerHeartbeat {
  private readonly redis: HeartbeatRedisClient;
  private readonly checkDatabase: () => Promise<unknown>;
  private readonly checkQueueReady: () => Promise<unknown>;
  private readonly checkRuntimeReady: () => Promise<unknown>;
  private readonly key: string;
  private readonly workerId: string;
  private readonly ttlSeconds: number;
  private readonly now: () => number;
  private readonly checkEventLoop: () => Promise<void>;
  private timer?: NodeJS.Timeout;
  private pulseInFlight?: Promise<void>;

  constructor(options: WorkerHeartbeatOptions) {
    this.redis = options.redis;
    this.checkDatabase = options.checkDatabase;
    this.checkQueueReady = options.checkQueueReady ?? (async () => undefined);
    this.checkRuntimeReady = options.checkRuntimeReady ?? (async () => undefined);
    this.key = options.key;
    this.workerId = options.key.slice(options.key.lastIndexOf(':') + 1);
    this.ttlSeconds = options.ttlSeconds ?? 15;
    this.now = options.now ?? Date.now;
    this.checkEventLoop = options.checkEventLoop ?? checkEventLoopResponsiveness;
  }

  async pulse(): Promise<void> {
    await this.checkEventLoop();
    await this.checkQueueReady();
    await this.checkRuntimeReady();
    await this.checkDatabase();
    await this.redis.ping();
    const now = this.now();
    await this.redis.zadd(WORKER_HEARTBEAT_REGISTRY_KEY, now, this.workerId);
    await this.redis.set(this.key, String(now), 'EX', this.ttlSeconds);
  }

  start(intervalMs: number, onError: (_error: unknown) => void): void {
    const runPulse = (): void => {
      if (this.pulseInFlight) {
        return;
      }

      this.pulseInFlight = this.pulse()
        .catch(onError)
        .finally(() => {
          this.pulseInFlight = undefined;
        });
    };

    runPulse();
    this.timer = setInterval(runPulse, intervalMs);
  }

  async stop(): Promise<void> {
    if (this.timer) {
      clearInterval(this.timer);
      this.timer = undefined;
    }

    await this.pulseInFlight;
    const cleanup = await Promise.allSettled([
      this.redis.del(this.key),
      this.redis.zrem(WORKER_HEARTBEAT_REGISTRY_KEY, this.workerId),
    ]);

    try {
      const failures = cleanup
        .filter((result): result is PromiseRejectedResult => result.status === 'rejected')
        .map((result) => result.reason);
      if (failures.length > 0) {
        throw new AggregateError(failures, 'Worker heartbeat cleanup failed');
      }
    } finally {
      await this.redis.quit();
    }
  }
}