// One claimed job's lease: the heartbeat that renews it, and the fact no driver reports — that
// renewals stopped landing long enough for the queue to hand this job to another worker. A lease
// that lapses in silence is one job running twice with nothing in the log saying so, which is the
// hardest bug in a queue to see from the outside and the easiest to name from in here.

import type { Clock } from '@ultimat3/core';
import { logger, recordLeaseLost, renderThrowable } from '@ultimat3/core';
import { nowMs } from './clock';
import type { ClaimedJob, JobDriver, JobRecord } from './driver';
import { LeaseLostError } from './errors';
import type { JobIntrospection } from './introspection';
import { type IntervalScheduler, startRenewalTimer } from './renewal-timer';

export interface LeaseHeartbeatOptions {
  /**
   * `heartbeat` renews; `introspect.job` is read ONLY when a renewal is refused, to tell a cancel
   * (the designed stop) from a lease another worker now holds. A lease renews itself and settles
   * nothing.
   */
  readonly driver: Pick<JobDriver, 'heartbeat'> & {
    readonly introspect?: Pick<JobIntrospection, 'job'> | undefined;
  };
  readonly claimed: ClaimedJob;
  readonly visibilityTimeoutMs: number;
  readonly intervalMs: number;
  readonly workerId: string;
  readonly clock?: Clock;
  /** What every renewal runs on (`renewal-timer.ts`). Default: a real, unrefed interval. */
  readonly schedule?: IntervalScheduler;
}

/** The longest a refused renewal waits on the row read that names it a cancel or a loss. */
export const CLASSIFY_READ_MAX_MS = 5_000;

export interface LeaseHeartbeat {
  /** One renewal. The interval drives it; tests drive it instead of the timer. */
  renew(): Promise<void>;
  /** True once the lease lapsed: the queue may already have delivered this job again. */
  lost(): boolean;
  /**
   * Aborts when the lease is gone. The worker folds it into the `Ctx` it hands `executeJob`, so a
   * job cancelled from outside — or one whose lease lapsed and whose row another worker
   * re-claimed — stops at the next renewal instead of running to the end beside its replacement.
   * `steps.ts` refuses every write past it, which is what unwinds a body that ignores the signal.
   */
  readonly signal: AbortSignal;
  /** Stop renewing. Idempotent. */
  stop(): void;
}

/**
 * Start renewing the lease `claim()` bought, and say so out loud when renewal stops working.
 *
 * The lease is measured on THIS process's clock, not on `claimed.visibleAt`: that timestamp comes
 * from the driver's clock, and comparing the two makes every lease decision a function of clock
 * skew. Measured from the moment renewal started, the window can only ever be read as shorter than
 * it really is — reporting late is safe, reporting early would cry loss over a live lease.
 */
export function startLeaseHeartbeat(options: LeaseHeartbeatOptions): LeaseHeartbeat {
  const { claimed, visibilityTimeoutMs, workerId } = options;
  const now = (): number => nowMs(options.clock);
  let renewedAt = now();
  let renewing = false;
  let lost = false;
  const gone = new AbortController();

  const lapsed = (): boolean => now() - renewedAt >= visibilityTimeoutMs;

  /**
   * Reported once, then renewal stops. Once the window is gone this job is claimable by anyone,
   * so a further renewal would extend a lease this process no longer holds — and a counter that
   * ticked once per interval would count intervals, not jobs.
   */
  const reportLost = (error?: unknown, reason?: 'expired' | 'not-ours'): void => {
    // `stopped()` as well as `lost`, and it is the difference between a page and a fact: a clean
    // completion acks the row out of `running` and `worker-run.ts` stops the heartbeat, so a
    // renewal already in flight comes back `false` for a job that FINISHED. Reported, that is
    // `jobs.lease.lost` at error plus `recordLeaseLost(queue)` — the one signal meaning the queue
    // re-delivered a job this process was still running — raised for a run nobody re-delivered.
    if (lost || timer.stopped()) return;
    lost = true;
    timer.stop();
    logLost(error, reason);
    // Cancel LAST, so the loss is logged and counted before the body it unwinds starts throwing.
    gone.abort(new LeaseLostError({ job: claimed.name, jobId: claimed.id }));
  };

  /** `jobs.lease.lost` at error, counted: this job may now run twice. */
  const logLost = (error?: unknown, reason?: 'expired' | 'not-ours'): void => {
    logger.error('jobs.lease.lost', {
      workerId,
      job: claimed.name,
      jobId: claimed.id,
      queue: claimed.queue,
      attempt: claimed.attempt,
      visibilityTimeoutMs,
      reason: reason ?? 'expired',
      ...(error === undefined ? {} : { error: renderThrowable(error) }),
    });
    recordLeaseLost(claimed.queue);
  };

  /**
   * The row said "not yours" — the driver answered, so ownership is GONE whatever the cause: the
   * attempt is aborted now, before anything else is awaited. Only then is the row read once, to say
   * which: a cancel (`x jobs cancel`, an app's own stop) is the designed way a stop reaches a
   * running attempt — `jobs.lease.cancelled` at info, never counted, since `job_leases_lost_total`
   * means "this job may now run twice" and a cancel cannot. Anything else, an unreadable row
   * included, is `jobs.lease.lost`. Read after the abort, a stalled read delays a log line, never
   * the stop (a stalled read used to keep the attempt running until the window lapsed) — and only
   * for one renewal interval (at most `CLASSIFY_READ_MAX_MS`): a read still out then is unreadable,
   * so a real loss is never left uncounted behind a read that never answers.
   */
  const refused = async (): Promise<void> => {
    if (lost || timer.stopped()) return;
    lost = true;
    timer.stop();
    gone.abort(new LeaseLostError({ job: claimed.name, jobId: claimed.id }));
    const row = await boundedRead();
    if (row?.state !== 'cancelled') {
      logLost(undefined, 'not-ours');
      return;
    }
    logger.info('jobs.lease.cancelled', {
      workerId,
      job: claimed.name,
      jobId: claimed.id,
      queue: claimed.queue,
      attempt: claimed.attempt,
    });
  };

  /** `introspect.job`, or `undefined` once it rejects or outlasts its bound. */
  const boundedRead = async (): Promise<JobRecord | undefined> => {
    const read = options.driver.introspect?.job(claimed.id);
    if (read === undefined) return undefined;
    let bound: ReturnType<typeof setTimeout> | undefined;
    const outlasted = new Promise<undefined>((resolve) => {
      bound = setTimeout(resolve, Math.min(options.intervalMs, CLASSIFY_READ_MAX_MS), undefined);
    });
    try {
      return await Promise.race([read.catch((): undefined => undefined), outlasted]);
    } finally {
      clearTimeout(bound);
    }
  };

  const renew = async (): Promise<void> => {
    if (lost || timer.stopped()) return;
    // Expiry is decided BEFORE the driver is asked, because the failure that loses a lease most
    // quietly is the one that never answers: a heartbeat hung on a dead connection neither
    // resolves nor rejects, so a check that ran only on rejection would never run at all.
    if (lapsed()) {
      reportLost();
      return;
    }
    // One renewal in flight at a time. A driver slower than the interval would otherwise stack a
    // request per tick onto the connection that is already the thing failing.
    if (renewing) return;
    renewing = true;
    // Stamped BEFORE the call: the driver writes the new lease at the statement, so the window it
    // bought starts no later than this. Stamped at the reply instead, a slow answer would stretch
    // this process's view of the lease past the server's by the reply's latency — and a second
    // worker re-claims the row while this one still believes it holds it.
    const sentAt = now();
    try {
      const held = await options.driver.heartbeat(claimed.id, {
        visibilityTimeoutMs,
        workerId,
        claim: claimed.claim,
      });
      // Re-read AFTER the await, never only before it: the whole point of the flag is the answer
      // that lands past `stop()`. Every branch below decides something about a lease this process
      // may no longer be running under.
      if (timer.stopped()) return;
      // The driver answered, and it said the row is not ours. That is a DIFFERENT fact from an
      // expired window and the only one an operator can cause on purpose: `x jobs cancel` writes
      // a terminal state, and the renewal that misses it is what tells this attempt to stop. It
      // is not a failure of the queue, so it is reported once and the attempt is unwound.
      // `=== false`, not `!held`: `heartbeat` only recently began answering, and a driver from
      // before that — a hand-rolled one, a test double — resolves `undefined`. Reading that as a
      // lost lease would cancel every job on every renewal. Only an explicit "no" is a no.
      if (held === false) {
        await refused();
        return;
      }
      // Asked again AFTER the call, because a renewal that SUCCEEDS late is still late: an
      // event-loop stall or a driver that answered at the end of a connect timeout can land this
      // past the window it was renewing, and renewing there would restart the clock on
      // a lease the queue has already re-delivered — the loss hidden by the very call meant to
      // prevent it.
      if (lapsed()) {
        reportLost();
        return;
      }
      renewedAt = sentAt;
    } catch (error) {
      // One failed renewal is not a lost lease: there is a whole visibility window left to land
      // the next one, and the default interval gives three tries inside it. Say it at warn and
      // let the window decide.
      logger.warn('jobs.heartbeat.failed', {
        workerId,
        job: claimed.name,
        jobId: claimed.id,
        attempt: claimed.attempt,
        error: renderThrowable(error),
      });
      if (lapsed()) reportLost(error);
    } finally {
      renewing = false;
    }
  };

  const timer = startRenewalTimer(options.intervalMs, renew, options.schedule);

  return { renew, lost: () => lost, signal: gone.signal, stop: () => timer.stop() };
}