src/server/jobs/heartbeat.ts

import { hostname } from 'node:os';
import { randomUUID } from 'node:crypto';
import type IORedis from 'ioredis';
import { heartbeatKey, queuePrefix } from './queue';

// 默认一个容器对应一个作用域；宿主机运行多个独立实例时可显式指定不同名称。
export const heartbeatScope = () =>
  process.env.WORKER_INSTANCE_ID?.trim() || hostname();
export const heartbeatRegistryKey = (scope?: string, prefix = queuePrefix()) =>
  `${heartbeatKey(prefix)}:active${scope ? `:${scope}` : ''}`;

export function createHeartbeat(
  redis: IORedis,
  scope = heartbeatScope(),
  ttlMs = 20000,
) {
  // 作用域可相同，登记身份不能相同；旧进程退出只能撤销自己的成员。
  const instanceId = `${scope}/${randomUUID()}`;
  const keys = [heartbeatRegistryKey(), heartbeatRegistryKey(scope)];
  return {
    instanceId,
    async renew() {
      // 使用 Redis 时间判断过期，避免不同 Worker 的本机时钟偏差互相误判。
      await redis.eval(
        `
        local time = redis.call('TIME')
        local now = time[1] * 1000 + math.floor(time[2] / 1000)
        for _, key in ipairs(KEYS) do
          redis.call('ZREMRANGEBYSCORE', key, '-inf', now)
          redis.call('ZADD', key, now + tonumber(ARGV[2]), ARGV[1])
          redis.call('PEXPIRE', key, tonumber(ARGV[2]) * 2)
        end
        return 1
      `,
        2,
        ...keys,
        instanceId,
        ttlMs,
      );
    },
    async remove() {
      await redis
        .multi()
        .zrem(keys[0], instanceId)
        .zrem(keys[1], instanceId)
        .exec();
    },
  };
}

export async function hasHeartbeat(
  redis: IORedis,
  scope?: string,
  prefix = queuePrefix(),
) {
  const count = await redis.eval(
    `
    local time = redis.call('TIME')
    local now = time[1] * 1000 + math.floor(time[2] / 1000)
    redis.call('ZREMRANGEBYSCORE', KEYS[1], '-inf', now)
    return redis.call('ZCARD', KEYS[1])
  `,
    1,
    heartbeatRegistryKey(scope, prefix),
  );
  return Number(count) > 0;
}