gadget/gadget-tasks/src/services/lock.ts

// src/services/lock.ts
// Copyright (C) 2026 Rob Colbert <rob.colbert@openplatform.us>
// Licensed under the Apache License, Version 2.0

import os from "node:os";
import { Redis } from "ioredis";
import env from "../config/env.ts";
import { GadgetLog } from "@gadget/api";

const log = new GadgetLog({ name: "TaskLock", slug: "task-lock" });

const LOCK_KEY = "gadget:tasks:lock";
const LOCK_TTL_SECONDS = 60;
const HEARTBEAT_INTERVAL_MS = 30000; // 30 seconds

export interface TaskLockData {
  pid: number;
  hostname: string;
  startedAt: number;
  heartbeatAt: number;
}

class TaskLockService {
  private redis: Redis | null = null;
  private heartbeatTimer: ReturnType<typeof setInterval> | null = null;

  async connect(): Promise<void> {
    this.redis = new Redis({
      host: env.redis.host,
      port: env.redis.port,
      password: env.redis.password || undefined,
      keyPrefix: env.redis.keyPrefix,
      lazyConnect: true,
    });
    await this.redis.connect();
    log.info("connected to Redis", {
      host: env.redis.host,
      port: env.redis.port,
    });
  }

  /**
   * Attempt to acquire the singleton lock.
   * Returns true if acquired, false if another instance is running.
   */
  async acquire(): Promise<boolean> {
    if (!this.redis) {
      throw new Error("Redis connection not initialized");
    }

    const data: TaskLockData = {
      pid: process.pid,
      hostname: os.hostname(),
      startedAt: Date.now(),
      heartbeatAt: Date.now(),
    };

    // SET gadget:tasks:lock <data> EX 60 NX
    const result = await this.redis.set(
      LOCK_KEY,
      JSON.stringify(data),
      "EX",
      LOCK_TTL_SECONDS,
      "NX",
    );

    if (result === "OK") {
      this.startHeartbeat();
      return true;
    }

    // Lock exists — check if stale
    const existing = await this.redis.get(LOCK_KEY);
    if (existing) {
      const info = JSON.parse(existing) as TaskLockData;
      const age = Date.now() - info.heartbeatAt;
      if (age > LOCK_TTL_SECONDS * 1000) {
        // Stale — reclaim
        log.warn("stale lock detected, reclaiming", { existingPid: info.pid });
        await this.redis.del(LOCK_KEY);
        return this.acquire(); // retry
      }
      log.error("gadget-tasks is already running", {
        pid: info.pid,
        hostname: info.hostname,
        startedAt: new Date(info.startedAt).toISOString(),
      });
      return false;
    }

    // Lock disappeared — retry
    return this.acquire();
  }

  /**
   * Release the lock on shutdown.
   */
  async release(): Promise<void> {
    if (this.heartbeatTimer) {
      clearInterval(this.heartbeatTimer);
      this.heartbeatTimer = null;
    }
    if (this.redis) {
      try {
        await this.redis.del(LOCK_KEY);
      } catch {
        // best effort
      }
      try {
        await this.redis.quit();
      } catch {
        // best effort
      }
      this.redis = null;
    }
  }

  /**
   * Start heartbeat timer to refresh lock TTL.
   */
  private startHeartbeat(): void {
    this.heartbeatTimer = setInterval(async () => {
      if (!this.redis) return;
      try {
        const existing = await this.redis.get(LOCK_KEY);
        if (existing) {
          const data = JSON.parse(existing) as TaskLockData;
          data.heartbeatAt = Date.now();
          await this.redis.set(LOCK_KEY, JSON.stringify(data), "EX", LOCK_TTL_SECONDS);
        }
      } catch (error) {
        log.error("heartbeat failed", { error });
      }
    }, HEARTBEAT_INTERVAL_MS);
  }
}

export default new TaskLockService();