import {
  createRedisClient,
  type Callback,
  type Redis,
  type RedisOptions,
  type Result,
} from "@internal/redis";
import {
  Attributes,
  context,
  getMeter,
  Meter,
  ObservableResult,
  propagation,
  SEMATTRS_MESSAGE_ID,
  SEMATTRS_MESSAGING_OPERATION,
  SEMATTRS_MESSAGING_SYSTEM,
  Span,
  SpanKind,
  SpanOptions,
  Tracer,
  ValueType,
} from "@internal/tracing";
import { promiseWithResolvers, tryCatch } from "@trigger.dev/core";
import { Logger, LogLevel } from "@trigger.dev/core/logger";
import { calculateNextRetryDelay, flattenAttributes } from "@trigger.dev/core/v3";
import { type RetryOptions } from "@trigger.dev/core/v3/schemas";
import { Decimal } from "@trigger.dev/database";
import { CronSchema, Worker, type WorkerConcurrencyOptions } from "@trigger.dev/redis-worker";
import { nanoid } from "nanoid";
import { Readable } from "node:stream";
import { setInterval, setTimeout } from "node:timers/promises";
import { z } from "zod";
import {
  attributesFromAuthenticatedEnv,
  MinimalAuthenticatedEnvironment,
} from "../shared/index.js";
import {
  InputPayload,
  OutputPayload,
  OutputPayloadV2,
  RunQueueKeyProducer,
  RunQueueKeyProducerEnvironment,
  RunQueueSelectionStrategy,
} from "./types.js";
import { WorkerQueueResolver } from "./workerQueueResolver.js";

const SemanticAttributes = {
  QUEUE: "runqueue.queue",
  WORKER_QUEUE: "runqueue.workerQueue",
  MASTER_QUEUE_SHARD: "runqueue.masterQueueShard",
  CONSUMER_ID: "runqueue.consumerId",
  RUN_ID: "runqueue.runId",
  RESULT_COUNT: "runqueue.resultCount",
  CONCURRENCY_KEY: "runqueue.concurrencyKey",
  ORG_ID: "runqueue.orgId",
};

export type RunQueueOptions = {
  name: string;
  tracer: Tracer;
  redis: RedisOptions;
  defaultEnvConcurrency: number;
  defaultEnvConcurrencyBurstFactor?: number;
  windowSize?: number;
  keys: RunQueueKeyProducer;
  queueSelectionStrategy: RunQueueSelectionStrategy;
  verbose?: boolean;
  logger?: Logger;
  logLevel?: LogLevel;
  retryOptions?: RetryOptions;
  shardCount?: number;
  masterQueueConsumersDisabled?: boolean;
  masterQueueConsumersIntervalMs?: number;
  masterQueueCooloffPeriodMs?: number;
  masterQueueCooloffCountThreshold?: number;
  masterQueueConsumerDequeueCount?: number;
  processWorkerQueueDebounceMs?: number;
  /**
   * TTL (seconds) applied to the per-base-queue lengthCounter/runningCounter on
   * lazy-init. Bounds the maximum window for any drift accumulated during a
   * rolling-deploy v1/v2 overlap. INCR/DECR do NOT extend the TTL, so the
   * counter expires this long after init regardless of activity, and the next
   * CK operation re-anchors from ckIndex. Default: 86400 (24h).
   */
  counterTtlSeconds?: number;
  workerOptions?: {
    pollIntervalMs?: number;
    immediatePollIntervalMs?: number;
    shutdownTimeoutMs?: number;
    concurrency?: WorkerConcurrencyOptions;
    disabled?: boolean;
  };
  meter?: Meter;
  dequeueBlockingTimeoutSeconds?: number;
  concurrencySweeper?: {
    scanSchedule?: string;
    scanJitterInMs?: number;
    processMarkedSchedule?: string;
    processMarkedJitterInMs?: number;
    callback: ConcurrencySweeperCallback;
  };
  /** TTL system for automatic run expiration */
  ttlSystem?: {
    /** Number of shards for TTL sorted sets (default: same as queue shards) */
    shardCount?: number;
    /** How often to poll each shard for expired runs (ms, default: 1000) */
    pollIntervalMs?: number;
    /** Max number of runs to expire per poll per shard (default: 100) */
    batchSize?: number;
    /** Whether TTL consumers (polling loops) are disabled on this instance (default: false) */
    consumersDisabled?: boolean;
    /** Key suffix for TTL worker's queue sorted set (relative to RunQueue keyPrefix) */
    workerQueueSuffix: string;
    /** Key suffix for TTL worker's items hash (relative to RunQueue keyPrefix) */
    workerItemsSuffix: string;
    /** Visibility timeout for TTL worker jobs (ms, default: 30000) */
    visibilityTimeoutMs?: number;
  };
};

export interface ConcurrencySweeperCallback {
  (runIds: string[]): Promise<Array<{ id: string; orgId: string }>>;
}

type DequeuedMessage = {
  messageId: string;
  messageScore: string;
  message: OutputPayload;
  workerQueueLength?: number;
};

type MarkedRun = {
  orgId: string;
  messageId: string;
  score: number;
};

const defaultRetrySettings = {
  maxAttempts: 12,
  factor: 2,
  minTimeoutInMs: 1_000,
  maxTimeoutInMs: 3_600_000,
  randomize: true,
};

const workerCatalog = {
  processQueueForWorkerQueue: {
    schema: z.object({
      queueKey: z.string(),
      environmentId: z.string(),
    }),
    visibilityTimeoutMs: 30_000,
  },
  scanConcurrencySets: {
    schema: CronSchema,
    visibilityTimeoutMs: 60_000 * 5,
    cron: "*/10 * * * *",
    jitterInMs: 60_000,
    retry: {
      maxAttempts: 1,
    },
  },
  processMarkedRuns: {
    schema: CronSchema,
    visibilityTimeoutMs: 60_000 * 5,
    cron: "*/5 * * * *",
    jitterInMs: 30_000,
    retry: {
      maxAttempts: 1,
    },
  },
};

type QueueCooloffState =
  | {
      _tag: "normal";
      consecutiveFailures: number;
    }
  | {
      _tag: "cooloff";
      cooloffExpiresAt: number;
    };

/**
 * RunQueue – the queue that's used to process runs
 */
export class RunQueue {
  private retryOptions: RetryOptions;
  private subscriber: Redis;
  private luaDebugSubscriber: Redis;
  private logger: Logger;
  public redis: Redis;
  public keys: RunQueueKeyProducer;
  private queueSelectionStrategy: RunQueueSelectionStrategy;
  private shardCount: number;
  private counterTtlSeconds: number;
  private abortController: AbortController;
  private worker: Worker<typeof workerCatalog>;
  private workerQueueResolver: WorkerQueueResolver;
  private _observableWorkerQueues: Set<string> = new Set();
  private _meter: Meter;
  private _queueCooloffStates: Map<string, QueueCooloffState> = new Map();

  constructor(public readonly options: RunQueueOptions) {
    this.shardCount = options.shardCount ?? 2;
    this.counterTtlSeconds = options.counterTtlSeconds ?? 86400;
    this.retryOptions = options.retryOptions ?? defaultRetrySettings;
    this.redis = createRedisClient(options.redis, {
      onError: (error) => {
        this.logger.error(`RunQueue redis client error:`, {
          error,
          keyPrefix: options.redis.keyPrefix,
        });
      },
    });
    this.logger = options.logger ?? new Logger("RunQueue", options.logLevel ?? "info");

    this.workerQueueResolver = new WorkerQueueResolver({ logger: this.logger });
    this._meter = options.meter ?? getMeter("run-queue");

    const workerQueueObservableGauge = this._meter.createObservableGauge(
      "runqueue.workerQueue.length",
      {
        description: "The number of messages in the worker queue",
        unit: "messages",
        valueType: ValueType.INT,
      }
    );

    const masterQueueObservableGauge = this._meter.createObservableGauge(
      "runqueue.masterQueue.length",
      {
        description: "The number of queues in the master queue shard",
        unit: "queues",
        valueType: ValueType.INT,
      }
    );

    workerQueueObservableGauge.addCallback(this.#updateWorkerQueueLength.bind(this));
    masterQueueObservableGauge.addCallback(this.#updateMasterQueueLength.bind(this));

    this.abortController = new AbortController();

    this.keys = options.keys;
    this.queueSelectionStrategy = options.queueSelectionStrategy;

    this.subscriber = this.redis.duplicate();
    this.luaDebugSubscriber = this.redis.duplicate();

    this.worker = new Worker({
      name: "run-queue-worker",
      redisOptions: {
        ...options.redis,
        keyPrefix: `${options.redis.keyPrefix}:worker`,
      },
      catalog: {
        ...workerCatalog,
        scanConcurrencySets: {
          ...workerCatalog.scanConcurrencySets,
          cron: options.concurrencySweeper?.scanSchedule ?? workerCatalog.scanConcurrencySets.cron,
          jitterInMs:
            options.concurrencySweeper?.scanJitterInMs ??
            workerCatalog.scanConcurrencySets.jitterInMs,
        },
        processMarkedRuns: {
          ...workerCatalog.processMarkedRuns,
          cron:
            options.concurrencySweeper?.processMarkedSchedule ??
            workerCatalog.processMarkedRuns.cron,
          jitterInMs:
            options.concurrencySweeper?.processMarkedJitterInMs ??
            workerCatalog.processMarkedRuns.jitterInMs,
        },
      },
      concurrency: options.workerOptions?.concurrency,
      pollIntervalMs: options.workerOptions?.pollIntervalMs ?? 1000,
      immediatePollIntervalMs: options.workerOptions?.immediatePollIntervalMs ?? 100,
      shutdownTimeoutMs: options.workerOptions?.shutdownTimeoutMs ?? 10_000,
      logger: new Logger("RunQueueWorker", options.logLevel ?? "info"),
      jobs: {
        processQueueForWorkerQueue: async (job) => {
          await this.#processQueueForWorkerQueue(job.payload.queueKey, job.payload.environmentId);
        },
        scanConcurrencySets: async (job) => {
          await this.scanConcurrencySets();
        },
        processMarkedRuns: async (job) => {
          await this.processMarkedRuns();
        },
      },
    });

    if (!options.workerOptions?.disabled) {
      this.worker.start();
    }

    this.#setupSubscriber();
    this.#setupLuaLogSubscriber();
    this.#startMasterQueueConsumers();
    this.#startTtlConsumers();
    this.#registerCommands();
  }

  get name() {
    return this.options.name;
  }

  get tracer() {
    return this.options.tracer;
  }

  get meter() {
    return this._meter;
  }

  public async registerObservableWorkerQueue(workerQueue: string) {
    this._observableWorkerQueues.add(workerQueue);
  }

  /**
   * Replaces the full set of worker queues observed by the `runqueue.workerQueue.length`
   * gauge. Used by a periodic observer that derives the set from the current worker
   * groups, so the observed set stays correct (and prunes queues that no longer exist)
   * independent of dequeue activity.
   */
  public setObservableWorkerQueues(workerQueues: string[]) {
    this._observableWorkerQueues = new Set(workerQueues);
  }

  async #updateWorkerQueueLength(observableResult: ObservableResult<Attributes>) {
    for (const workerQueue of this._observableWorkerQueues) {
      const workerQueueLength = await this.redis.llen(this.keys.workerQueueKey(workerQueue));

      observableResult.observe(workerQueueLength, {
        [SemanticAttributes.WORKER_QUEUE]: workerQueue,
      });
    }
  }

  async #updateMasterQueueLength(observableResult: ObservableResult<Attributes>) {
    for (let shard = 0; shard < this.shardCount; shard++) {
      const masterQueueKey = this.keys.masterQueueKeyForShard(shard);
      const masterQueueLength = await this.redis.zcard(masterQueueKey);

      observableResult.observe(masterQueueLength, {
        [SemanticAttributes.MASTER_QUEUE_SHARD]: shard.toString(),
      });
    }
  }

  public async updateQueueConcurrencyLimits(
    env: MinimalAuthenticatedEnvironment,
    queue: string,
    concurrency: number
  ) {
    return this.redis.set(this.keys.queueConcurrencyLimitKey(env, queue), concurrency);
  }

  public async removeQueueConcurrencyLimits(env: MinimalAuthenticatedEnvironment, queue: string) {
    return this.redis.del(this.keys.queueConcurrencyLimitKey(env, queue));
  }

  public async getQueueConcurrencyLimit(env: MinimalAuthenticatedEnvironment, queue: string) {
    const result = await this.redis.get(this.keys.queueConcurrencyLimitKey(env, queue));

    return result ? Number(result) : undefined;
  }

  public async updateEnvConcurrencyLimits(env: MinimalAuthenticatedEnvironment) {
    await this.#callUpdateEnvironmentConcurrencyLimits({
      envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env),
      envConcurrencyLimit: env.maximumConcurrencyLimit,
      envConcurrencyLimitBurstFactorKey: this.keys.envConcurrencyLimitBurstFactorKey(env),
      envConcurrencyLimitBurstFactor: env.concurrencyLimitBurstFactor.toString(),
    });
  }

  public async getEnvConcurrencyLimit(env: MinimalAuthenticatedEnvironment) {
    const result = await this.redis.get(this.keys.envConcurrencyLimitKey(env));

    return result ? Number(result) : this.options.defaultEnvConcurrency;
  }

  public async getEnvConcurrencyLimitWithBurstFactor(env: MinimalAuthenticatedEnvironment) {
    const result = await this.redis.get(this.keys.envConcurrencyLimitBurstFactorKey(env));

    const burstFactor = result
      ? Number(result)
      : (this.options.defaultEnvConcurrencyBurstFactor ?? 1);

    const limit = await this.getEnvConcurrencyLimit(env);

    return Math.floor(limit * burstFactor);
  }

  public async getEnvConcurrencyBurstFactor(env: MinimalAuthenticatedEnvironment) {
    const result = await this.redis.get(this.keys.envConcurrencyLimitBurstFactorKey(env));

    const burstFactor = result
      ? Number(result)
      : (this.options.defaultEnvConcurrencyBurstFactor ?? 1);

    return burstFactor;
  }

  public async getCurrentConcurrencyOfEnvironment(env: MinimalAuthenticatedEnvironment) {
    return this.redis.smembers(this.keys.envCurrentConcurrencyKey(env));
  }

  public async getCurrentConcurrencyOfQueue(env: MinimalAuthenticatedEnvironment, queue: string) {
    return this.redis.smembers(this.keys.queueCurrentConcurrencyKey(env, queue));
  }

  public async lengthOfQueue(
    env: MinimalAuthenticatedEnvironment,
    queue: string,
    concurrencyKey?: string
  ) {
    // Per-variant length when caller specifies a concurrency key
    if (concurrencyKey) {
      return this.redis.zcard(this.keys.queueKey(env, queue, concurrencyKey));
    }

    // Aggregate base-queue length = ZCARD(base) + GET(lengthCounter).
    // The counter is non-existent for queues that have never had a CK enqueue —
    // in that case it returns null which we treat as 0 and the base ZCARD is
    // the whole truth.
    const baseKey = this.keys.queueKey(env, queue);
    const lengthCounterKey = this.keys.queueLengthCounterKey(env, queue);

    const pipeline = this.redis.pipeline();
    pipeline.zcard(baseKey);
    pipeline.get(lengthCounterKey);
    const results = await pipeline.exec();

    if (!results) {
      return 0;
    }
    const [baseErr, baseVal] = results[0];
    const [ctrErr, ctrVal] = results[1];
    const baseCount = baseErr || baseVal == null ? 0 : (baseVal as number);
    const ctrCount = ctrErr || ctrVal == null ? 0 : Number(ctrVal);
    return baseCount + ctrCount;
  }

  public async lengthOfQueueAvailableMessages(
    env: MinimalAuthenticatedEnvironment,
    queue: string,
    currentTime: Date = new Date(),
    concurrencyKey?: string
  ) {
    return this.redis.zcount(
      this.keys.queueKey(env, queue, concurrencyKey),
      "-inf",
      String(currentTime.getTime())
    );
  }

  public async lengthOfEnvQueue(env: MinimalAuthenticatedEnvironment) {
    return this.redis.zcard(this.keys.envQueueKey(env));
  }

  public async lengthOfDeadLetterQueue(env: MinimalAuthenticatedEnvironment) {
    return this.redis.zcard(this.keys.deadLetterQueueKey(env));
  }

  public async messageInDeadLetterQueue(env: MinimalAuthenticatedEnvironment, messageId: string) {
    const result = await this.redis.zscore(this.keys.deadLetterQueueKey(env), messageId);
    return !!result;
  }

  public async redriveMessage(env: MinimalAuthenticatedEnvironment, messageId: string) {
    // Publish redrive message
    await this.redis.publish(
      "rq:redrive",
      JSON.stringify({
        runId: messageId,
        orgId: env.organization.id,
        envId: env.id,
        projectId: env.project.id,
      })
    );
  }

  public async oldestMessageInQueue(
    env: MinimalAuthenticatedEnvironment,
    queue: string,
    concurrencyKey?: string
  ) {
    // Get the "score" of the sorted set to get the oldest message score
    const result = await this.redis.zrange(
      this.keys.queueKey(env, queue, concurrencyKey),
      0,
      0,
      "WITHSCORES"
    );

    if (result.length === 0) {
      return;
    }

    const score = Number(result[1]);

    return score;
  }

  public async currentConcurrencyOfQueue(
    env: MinimalAuthenticatedEnvironment,
    queue: string,
    concurrencyKey?: string
  ) {
    return this.redis.scard(this.keys.queueCurrentConcurrencyKey(env, queue, concurrencyKey));
  }

  public async currentDequeuedOfQueue(
    env: MinimalAuthenticatedEnvironment,
    queue: string,
    concurrencyKey?: string
  ) {
    return this.redis.scard(this.keys.queueCurrentDequeuedKey(env, queue, concurrencyKey));
  }

  public async currentConcurrencyOfQueues(
    env: MinimalAuthenticatedEnvironment,
    queues: string[]
  ): Promise<Record<string, number>> {
    // For each queue, SCARD(base:currentDequeued) + GET(runningCounter). Missing
    // counter is treated as 0 so non-CK queues just see the base SCARD.
    const pipeline = this.redis.pipeline();
    queues.forEach((queue) => {
      pipeline.scard(this.keys.queueCurrentDequeuedKey(env, queue));
      pipeline.get(this.keys.queueRunningCounterKey(env, queue));
    });

    const results = await pipeline.exec();

    const empty = (): Record<string, number> =>
      queues.reduce(
        (acc, queue) => {
          acc[queue] = 0;
          return acc;
        },
        {} as Record<string, number>
      );

    if (!results) return empty();

    return queues.reduce(
      (acc, queue, index) => {
        const [baseErr, baseVal] = results[index * 2];
        const [ctrErr, ctrVal] = results[index * 2 + 1];
        const baseCount = baseErr || baseVal == null ? 0 : (baseVal as number);
        const ctrCount = ctrErr || ctrVal == null ? 0 : Number(ctrVal);
        acc[queue] = baseCount + ctrCount;
        return acc;
      },
      {} as Record<string, number>
    );
  }

  public async lengthOfQueues(
    env: MinimalAuthenticatedEnvironment,
    queues: string[]
  ): Promise<Record<string, number>> {
    // For each queue, ZCARD(base) + GET(lengthCounter). Missing counter is
    // treated as 0 so non-CK queues just see the base ZCARD.
    const pipeline = this.redis.pipeline();
    queues.forEach((queue) => {
      pipeline.zcard(this.keys.queueKey(env, queue));
      pipeline.get(this.keys.queueLengthCounterKey(env, queue));
    });

    const results = await pipeline.exec();

    const empty = (): Record<string, number> =>
      queues.reduce(
        (acc, queue) => {
          acc[queue] = 0;
          return acc;
        },
        {} as Record<string, number>
      );

    if (!results) return empty();

    return queues.reduce(
      (acc, queue, index) => {
        const [baseErr, baseVal] = results[index * 2];
        const [ctrErr, ctrVal] = results[index * 2 + 1];
        const baseCount = baseErr || baseVal == null ? 0 : (baseVal as number);
        const ctrCount = ctrErr || ctrVal == null ? 0 : Number(ctrVal);
        acc[queue] = baseCount + ctrCount;
        return acc;
      },
      {} as Record<string, number>
    );
  }

  /**
   * Get the current concurrency of the environment
   * @param env - The environment to get the current concurrency of
   * @returns The current concurrency of the environment
   */
  public async currentConcurrencyOfEnvironment(env: MinimalAuthenticatedEnvironment) {
    // We are now using the envCurrentDequeuedKey to track the current "displayed" concurrency of the environment
    // Operationally, we use the envCurrentConcurrencyKey to track the actual concurrency of the environment
    // The currentDequeuedKey is incremented when a message is dequeued from the worker queue,
    // wherease the currentConcurrencyKey is incremented when a message is dequeued from the message queue and put into the worker queue
    return this.redis.scard(this.keys.envCurrentDequeuedKey(env));
  }

  /**
   * Get the operational current concurrency of the environment
   * @param env - The environment to get the current concurrency of
   * @returns The current concurrency of the environment
   */
  public async operationalCurrentConcurrencyOfEnvironment(env: MinimalAuthenticatedEnvironment) {
    return this.redis.scard(this.keys.envCurrentConcurrencyKey(env));
  }

  public async messageExists(orgId: string, messageId: string) {
    return this.redis.exists(this.keys.messageKey(orgId, messageId));
  }

  public async readMessage(orgId: string, messageId: string) {
    return this.readMessageFromKey(this.keys.messageKey(orgId, messageId));
  }

  public async readMessageFromKey(messageKey: string) {
    return this.#trace(
      "readMessageFromKey",
      async (span) => {
        const rawMessage = await this.redis.get(messageKey);

        if (!rawMessage) {
          return;
        }

        const [error, message] = parseRawMessage(rawMessage);

        if (error) {
          this.logger.error(`[${this.name}] Failed to parse message`, {
            messageKey,
            error,
            service: this.name,
            message: message ?? rawMessage,
          });
        }

        if (message) {
          span.setAttributes({
            [SemanticAttributes.QUEUE]: message.queue,
            [SemanticAttributes.RUN_ID]: message.runId,
            [SemanticAttributes.CONCURRENCY_KEY]: message.concurrencyKey,
            [SemanticAttributes.WORKER_QUEUE]: this.#getWorkerQueueFromMessage(message),
          });
        }

        return message;
      },
      {
        attributes: {
          [SEMATTRS_MESSAGING_OPERATION]: "receive",
          [SEMATTRS_MESSAGING_SYSTEM]: "runqueue",
        },
      }
    );
  }

  /**
   * Enqueue a run message for execution.
   *
   * The Lua script atomically decides between two paths:
   *
   * **Fast path** (when `enableFastPath` is true and conditions are met):
   * Skips the queue sorted set and pushes directly to the worker queue (Redis list).
   * This eliminates the ~500ms debounce delay from the processQueueForWorkerQueue job.
   * Conditions: no available messages in the queue, env concurrency available, queue concurrency available.
   *
   * **Slow path** (default, or when fast-path conditions aren't met):
   * Adds to the queue sorted set, then a debounced processQueueForWorkerQueue job moves it
   * to the worker queue after checking concurrency limits.
   *
   * Both paths are atomic within their respective Lua scripts. The fast path claims concurrency
   * slots (SADD to currentConcurrency sets) identically to how the dequeue Lua does, so
   * ack/nack/release operations work the same regardless of which path was taken.
   *
   * @param enableFastPath - Gated per WorkerInstanceGroup. Dev environments always true.
   *   Production regions opt in via the `enableFastPath` flag on WorkerInstanceGroup.
   */
  public async enqueueMessage({
    env,
    message,
    workerQueue,
    skipDequeueProcessing = false,
    enableFastPath = false,
  }: {
    env: MinimalAuthenticatedEnvironment;
    message: InputPayload;
    workerQueue: string;
    skipDequeueProcessing?: boolean;
    /** When true, the Lua script may push directly to the worker queue if concurrency is available. */
    enableFastPath?: boolean;
  }) {
    return await this.#trace(
      "enqueueMessage",
      async (span) => {
        const { runId, concurrencyKey } = message;

        const queueKey = this.keys.queueKey(env, message.queue, concurrencyKey);

        // For CK queues, use wildcard dedup so all CKs share one worker queue processing job
        const dedupQueueKey = concurrencyKey ? this.keys.toCkWildcard(queueKey) : queueKey;

        propagation.inject(context.active(), message);

        span.setAttributes({
          [SemanticAttributes.QUEUE]: queueKey,
          [SemanticAttributes.RUN_ID]: runId,
          [SemanticAttributes.CONCURRENCY_KEY]: concurrencyKey,
          [SemanticAttributes.WORKER_QUEUE]: workerQueue,
        });

        const messagePayload: OutputPayloadV2 = {
          ...message,
          version: "2",
          queue: queueKey,
          workerQueue,
          attempt: 0,
        };

        // Pass TTL info to enqueue so it can be added atomically
        const ttlInfo =
          message.ttlExpiresAt && this.options.ttlSystem
            ? {
                ttlExpiresAt: message.ttlExpiresAt,
                ttlQueueKey: this.keys.ttlQueueKeyForShard(this.#getTtlShardForQueue(queueKey)),
                ttlMember: `${queueKey}|${message.runId}|${message.orgId}`,
              }
            : undefined;

        // Enqueue the message. The Lua script atomically decides whether to fast-path
        // (push directly to worker queue) or slow-path (add to queue sorted set).
        const fastPathTaken = await this.#callEnqueueMessage(
          messagePayload,
          ttlInfo,
          enableFastPath
        );

        span.setAttribute("fastPath", fastPathTaken);

        if (!fastPathTaken && !skipDequeueProcessing) {
          // Slow path: schedule the dequeue job to move the message from queue to worker queue
          await this.worker.enqueueOnce({
            id: dedupQueueKey, // dedupe by environment and base queue (CK wildcard for CK queues)
            job: "processQueueForWorkerQueue",
            payload: {
              queueKey: dedupQueueKey,
              environmentId: env.id,
            },
            // Add a small delay to dedupe messages so at most one of these will processed,
            // every 500ms per queue, concurrency key, and environment
            availableAt: new Date(Date.now() + (this.options.processWorkerQueueDebounceMs ?? 500)), // 500ms from now
          });
        }
      },
      {
        kind: SpanKind.PRODUCER,
        attributes: {
          [SEMATTRS_MESSAGING_OPERATION]: "publish",
          [SEMATTRS_MESSAGE_ID]: message.runId,
          [SEMATTRS_MESSAGING_SYSTEM]: "runqueue",
          ...attributesFromAuthenticatedEnv(env),
        },
      }
    );
  }

  /**
   * Dequeue messages from the worker queue
   */
  public async dequeueMessageFromWorkerQueue(
    consumerId: string,
    workerQueue: string,
    options?: {
      blockingPop?: boolean;
      blockingPopTimeoutSeconds?: number;
    }
  ): Promise<DequeuedMessage | undefined> {
    return this.#trace(
      "dequeueMessageFromWorkerQueue",
      async (span) => {
        const dequeuedMessage = await this.#callDequeueMessageFromWorkerQueue({
          workerQueue,
          blockingPop: options?.blockingPop ?? true,
          blockingPopTimeoutSeconds:
            options?.blockingPopTimeoutSeconds ?? this.options.dequeueBlockingTimeoutSeconds ?? 10,
        });

        if (!dequeuedMessage) {
          return;
        }

        span.setAttributes({
          [SemanticAttributes.QUEUE]: dequeuedMessage.message.queue,
          [SemanticAttributes.RUN_ID]: dequeuedMessage.messageId,
          [SemanticAttributes.CONCURRENCY_KEY]: dequeuedMessage.message.concurrencyKey,
          ...flattenAttributes(dequeuedMessage.message, "message"),
        });

        return dequeuedMessage;
      },
      {
        kind: SpanKind.CONSUMER,
        attributes: {
          [SEMATTRS_MESSAGING_OPERATION]: "receive",
          [SEMATTRS_MESSAGING_SYSTEM]: "runqueue",
          [SemanticAttributes.WORKER_QUEUE]: workerQueue,
          [SemanticAttributes.CONSUMER_ID]: consumerId,
        },
      }
    );
  }

  /**
   * Acknowledge a message, which will:
   * - remove all data from the queue
   * - release all concurrency
   * This is done when the run is in a final state.
   * @param messageId
   */
  public async acknowledgeMessage(
    orgId: string,
    messageId: string,
    options?: { skipDequeueProcessing?: boolean; removeFromWorkerQueue?: boolean }
  ) {
    return this.#trace(
      "acknowledgeMessage",
      async (span) => {
        const message = await this.readMessage(orgId, messageId);

        if (!message) {
          // Message not found, it may have already been acknowledged
          return;
        }

        span.setAttributes({
          [SemanticAttributes.QUEUE]: message.queue,
          [SemanticAttributes.ORG_ID]: message.orgId,
          [SemanticAttributes.RUN_ID]: messageId,
          [SemanticAttributes.CONCURRENCY_KEY]: message.concurrencyKey,
        });

        if (!options?.skipDequeueProcessing) {
          // For CK queues, use wildcard dedup so all CKs share one worker queue processing job
          const dedupQueueKey = message.concurrencyKey
            ? this.keys.toCkWildcard(message.queue)
            : message.queue;

          // This will move the message to the worker queue so it can be dequeued
          await this.worker.enqueueOnce({
            id: dedupQueueKey, // dedupe by environment and base queue
            job: "processQueueForWorkerQueue",
            payload: {
              queueKey: dedupQueueKey,
              environmentId: message.environmentId,
            },
            // Add a small delay to dedupe messages so at most one of these will processed,
            // every 500ms per queue, concurrency key, and environment
            availableAt: new Date(Date.now() + (this.options.processWorkerQueueDebounceMs ?? 500)), // 500ms from now
          });
        }

        await this.#callAcknowledgeMessage({
          message,
          removeFromWorkerQueue: options?.removeFromWorkerQueue,
        });
      },
      {
        kind: SpanKind.CONSUMER,
        attributes: {
          [SEMATTRS_MESSAGING_OPERATION]: "ack",
          [SEMATTRS_MESSAGE_ID]: messageId,
          [SEMATTRS_MESSAGING_SYSTEM]: "runqueue",
        },
      }
    );
  }

  /**
   * Negative acknowledge a message, which will requeue the message (with an optional future date).
    If you pass no date it will get reattempted with exponential backoff.
   */
  public async nackMessage({
    orgId,
    messageId,
    retryAt,
    incrementAttemptCount = true,
    skipDequeueProcessing = false,
  }: {
    orgId: string;
    messageId: string;
    retryAt?: number;
    incrementAttemptCount?: boolean;
    skipDequeueProcessing?: boolean;
  }) {
    return this.#trace(
      "nackMessage",
      async (span) => {
        const maxAttempts = this.retryOptions.maxAttempts ?? defaultRetrySettings.maxAttempts;

        const message = await this.readMessage(orgId, messageId);
        if (!message) {
          this.logger.log(`[${this.name}].nackMessage() message not found`, {
            orgId,
            messageId,
            maxAttempts,
            retryAt,
            service: this.name,
          });
          return;
        }

        span.setAttributes({
          [SemanticAttributes.QUEUE]: message.queue,
          [SemanticAttributes.RUN_ID]: messageId,
          [SemanticAttributes.CONCURRENCY_KEY]: message.concurrencyKey,
          [SemanticAttributes.WORKER_QUEUE]: this.#getWorkerQueueFromMessage(message),
        });

        if (incrementAttemptCount) {
          message.attempt = message.attempt + 1;
          if (message.attempt >= maxAttempts) {
            await this.#callMoveToDeadLetterQueue({ message });
            return false;
          }
        }

        if (!skipDequeueProcessing) {
          // For CK queues, use wildcard dedup so all CKs share one worker queue processing job
          const dedupQueueKey = message.concurrencyKey
            ? this.keys.toCkWildcard(message.queue)
            : message.queue;

          // This will move the message to the worker queue so it can be dequeued
          await this.worker.enqueueOnce({
            id: dedupQueueKey, // dedupe by environment and base queue
            job: "processQueueForWorkerQueue",
            payload: {
              queueKey: dedupQueueKey,
              environmentId: message.environmentId,
            },
            // Add a small delay to dedupe messages so at most one of these will processed,
            // every 500ms per queue, concurrency key, and environment
            availableAt: new Date(Date.now() + (this.options.processWorkerQueueDebounceMs ?? 500)), // 500ms from now
          });
        }

        await this.#callNackMessage({ message, retryAt });

        return true;
      },
      {
        kind: SpanKind.CONSUMER,
        attributes: {
          [SEMATTRS_MESSAGING_OPERATION]: "nack",
          [SEMATTRS_MESSAGE_ID]: messageId,
          [SEMATTRS_MESSAGING_SYSTEM]: "runqueue",
        },
      }
    );
  }

  /**
   * Release all concurrency for a message, including environment and queue concurrency
   */
  public async releaseAllConcurrency(orgId: string, messageId: string) {
    return this.#trace(
      "releaseAllConcurrency",
      async (span) => {
        const message = await this.readMessage(orgId, messageId);

        if (!message) {
          this.logger.log(`[${this.name}].releaseAllConcurrency() message not found`, {
            messageId,
            service: this.name,
          });
          return;
        }

        span.setAttributes({
          [SemanticAttributes.QUEUE]: message.queue,
          [SemanticAttributes.ORG_ID]: message.orgId,
          [SemanticAttributes.RUN_ID]: messageId,
          [SemanticAttributes.CONCURRENCY_KEY]: message.concurrencyKey,
        });

        // CK queues route through the Tracked variant so the base-queue
        // runningCounter stays in sync. Non-CK queues keep the original
        // release path — no counter to maintain.
        if (message.concurrencyKey) {
          return this.redis.releaseConcurrencyTracked(
            this.keys.queueCurrentConcurrencyKeyFromQueue(message.queue),
            this.keys.envCurrentConcurrencyKeyFromQueue(message.queue),
            this.keys.queueCurrentDequeuedKeyFromQueue(message.queue),
            this.keys.envCurrentDequeuedKeyFromQueue(message.queue),
            this.keys.queueRunningCounterKeyFromQueue(message.queue),
            this.keys.ckIndexKeyFromQueue(message.queue),
            messageId,
            this.options.redis.keyPrefix ?? "",
            String(this.counterTtlSeconds)
          );
        }

        return this.redis.releaseConcurrency(
          this.keys.queueCurrentConcurrencyKeyFromQueue(message.queue),
          this.keys.envCurrentConcurrencyKeyFromQueue(message.queue),
          this.keys.queueCurrentDequeuedKeyFromQueue(message.queue),
          this.keys.envCurrentDequeuedKeyFromQueue(message.queue),
          messageId
        );
      },
      {
        kind: SpanKind.CONSUMER,
        attributes: {
          [SEMATTRS_MESSAGING_OPERATION]: "releaseAllConcurrency",
          [SEMATTRS_MESSAGE_ID]: messageId,
          [SEMATTRS_MESSAGING_SYSTEM]: "runqueue",
        },
      }
    );
  }

  public async removeEnvironmentQueuesFromMasterQueue(
    runtimeEnvironmentId: string,
    organizationId: string,
    projectId: string
  ) {
    // Calculate the master queue shard for this environment
    const masterQueue = this.keys.masterQueueKeyForEnvironment(
      runtimeEnvironmentId,
      this.shardCount
    );

    // Use scanStream to find all matching members
    const stream = this.redis.zscanStream(masterQueue, {
      match: this.keys.queueKey(organizationId, projectId, "*", "*"),
      count: 100,
    });

    return new Promise<void>((resolve, reject) => {
      const matchingQueues: string[] = [];

      stream.on("data", (resultKeys) => {
        // zscanStream returns [member1, score1, member2, score2, ...]
        // We only want the members (even indices)
        for (let i = 0; i < resultKeys.length; i += 2) {
          matchingQueues.push(resultKeys[i]);
        }
      });

      stream.on("end", async () => {
        if (matchingQueues.length > 0) {
          await this.redis.zrem(masterQueue, matchingQueues);
        }
        resolve();
      });

      stream.on("error", (err) => reject(err));
    });
  }

  public async clearMessageFromConcurrencySets(params: {
    runId: string;
    orgId: string;
    queue: string;
    env: RunQueueKeyProducerEnvironment;
  }) {
    return this.#callClearMessageFromConcurrencySets(params);
  }

  async quit() {
    this.abortController.abort();

    await Promise.all([
      this.subscriber.unsubscribe(),
      this.luaDebugSubscriber.unsubscribe(),
      this.subscriber.quit(),
      this.luaDebugSubscriber.quit(),
      this.worker.stop(),
    ]);

    await this.redis.quit();
  }

  /**
   * Peek all messages on a worker queue (useful for tests or debugging)
   */
  async peekAllOnWorkerQueue(workerQueue: string) {
    const workerQueueKey = this.keys.workerQueueKey(workerQueue);
    return await this.redis.lrange(workerQueueKey, 0, -1);
  }

  /**
   * Create a scan stream for queue current concurrency keys
   */
  public currentConcurrencyScanStream(
    count: number = 10,
    onEnd?: () => void,
    onError?: (error: Error) => void
  ): { stream: Readable; redis: Redis } {
    const pattern = this.keys.currentConcurrencySetKeyScanPattern();
    const stream = this.redis.scanStream({
      match: pattern,
      count,
      type: "set",
    });

    if (onEnd) {
      stream.on("end", onEnd);
    }

    if (onError) {
      stream.on("error", onError);
    }

    return {
      stream,
      redis: this.redis,
    };
  }

  private async handleRedriveMessage(channel: string, message: string) {
    try {
      const { runId, envId, projectId, orgId } = JSON.parse(message) as any;
      if (
        typeof orgId !== "string" ||
        typeof runId !== "string" ||
        typeof envId !== "string" ||
        typeof projectId !== "string"
      ) {
        this.logger.error(
          "handleRedriveMessage: invalid message format: runId, envId, projectId and orgId must be strings",
          { message, channel }
        );
        return;
      }

      const data = await this.readMessage(orgId, runId);

      if (!data) {
        this.logger.error(`handleRedriveMessage: couldn't read message`, { orgId, runId, channel });
        return;
      }

      await this.enqueueMessage({
        env: {
          id: data.environmentId,
          type: data.environmentType,
          //this isn't used in enqueueMessage
          maximumConcurrencyLimit: -1,
          concurrencyLimitBurstFactor: new Decimal(1.0),
          project: {
            id: data.projectId,
          },
          organization: {
            id: data.orgId,
          },
        },
        message: {
          ...data,
          attempt: 0,
        },
        workerQueue: this.#getWorkerQueueFromMessage(data),
      });

      //remove from the dlq
      const result = await this.redis.zrem(
        this.keys.deadLetterQueueKey({ envId, orgId, projectId }),
        runId
      );

      if (result === 0) {
        this.logger.error(`handleRedriveMessage: couldn't remove message from dlq`, {
          orgId,
          runId,
          channel,
        });
        return;
      }

      this.logger.log(`handleRedriveMessage: redrived item ${runId} from Dead Letter Queue`);
    } catch (error) {
      this.logger.error("Error processing redrive message", { error, message });
    }
  }

  async #trace<T>(
    name: string,
    fn: (span: Span) => Promise<T>,
    options?: SpanOptions & { sampleRate?: number }
  ): Promise<T> {
    return this.tracer.startActiveSpan(
      name,
      {
        ...options,
        attributes: {
          ...options?.attributes,
        },
      },
      async (span) => {
        try {
          return await fn(span);
        } catch (e) {
          if (e instanceof Error) {
            span.recordException(e);
          } else {
            span.recordException(new Error(String(e)));
          }

          throw e;
        } finally {
          span.end();
        }
      }
    );
  }

  async #setupSubscriber() {
    const channel = `${this.options.name}:redrive`;
    this.subscriber.subscribe(channel, (err) => {
      if (err) {
        this.logger.error(`Failed to subscribe to ${channel}`, { error: err });
      } else {
        this.logger.log(`Subscribed to ${channel}`);
      }
    });

    this.subscriber.on("message", this.handleRedriveMessage.bind(this));
  }

  /**
   * Debug lua scripts by publishing to this channel
   *
   * @example
   *
   * ```lua
   * redis.call("PUBLISH", "runqueue:lua:debug", "workerQueueKey: " .. workerQueueKey .. " messageKeyValue -> " .. tostring(messageKeyValue))
   * ```
   */
  async #setupLuaLogSubscriber() {
    this.luaDebugSubscriber.subscribe("runqueue:lua:debug", (err) => {
      if (err) {
        this.logger.error(`Failed to subscribe to runqueue:lua:debug`, { error: err });
      } else {
        this.logger.log(`Subscribed to runqueue:lua:debug`);
      }
    });

    this.luaDebugSubscriber.on("message", (_channel, msg) => {
      this.logger.debug("runqueue lua debug", { msg });
    });
  }

  #startMasterQueueConsumers() {
    if (this.options.masterQueueConsumersDisabled) {
      this.logger.debug("Master queue consumers disabled");

      return;
    }

    for (let i = 0; i < this.shardCount; i++) {
      this.logger.debug(`Starting master queue consumer ${i}`);
      // We will start a consumer for each shard
      this.#startMasterQueueConsumer(i).catch((err) => {
        this.logger.error(`Failed to start master queue consumer ${i}`, { error: err });
      });
    }

    this.logger.debug(`Started ${this.shardCount} master queue consumers`);
  }

  async #startMasterQueueConsumer(shard: number) {
    let lastProcessedAt = Date.now();
    let processedCount = 0;

    const consumerId = nanoid();

    try {
      for await (const _ of setInterval(this.options.masterQueueConsumersIntervalMs ?? 500, null, {
        signal: this.abortController.signal,
      })) {
        this.logger.verbose(`Processing master queue shard ${shard}`, {
          processedCount,
          lastProcessedAt,
          service: this.name,
          shard,
          consumerId,
        });

        const now = performance.now();

        const [error, results] = await tryCatch(
          this.#processMasterQueueShard(
            shard,
            consumerId,
            this.options.masterQueueConsumerDequeueCount ?? 10
          )
        );

        if (error) {
          this.logger.error(`Failed to process master queue shard ${shard}`, {
            error,
            service: this.name,
            shard,
            consumerId,
          });

          continue;
        }

        const duration = performance.now() - now;

        this.logger.verbose(`Processed master queue shard ${shard} in ${duration}ms`, {
          processedCount,
          lastProcessedAt,
          service: this.name,
          shard,
          duration,
          results,
          consumerId,
        });

        processedCount++;
        lastProcessedAt = Date.now();
      }
    } catch (error) {
      if (error instanceof Error && error.name !== "AbortError") {
        throw error;
      }

      this.logger.debug(`Master queue consumer ${shard} stopped`, {
        service: this.name,
        shard,
        processedCount,
        lastProcessedAt,
      });
    }
  }

  // TTL System Methods

  #startTtlConsumers() {
    if (!this.options.ttlSystem) {
      this.logger.debug("TTL system disabled (no ttlSystem config)");
      return;
    }

    if (this.options.ttlSystem.consumersDisabled) {
      this.logger.debug("TTL consumers disabled on this instance");
      return;
    }

    const shardCount = this.options.ttlSystem.shardCount ?? this.shardCount;

    for (let i = 0; i < shardCount; i++) {
      this.logger.debug(`Starting TTL consumer ${i}`);
      this.#startTtlConsumer(i).catch((err) => {
        this.logger.error(`Failed to start TTL consumer ${i}`, { error: err });
      });
    }

    this.logger.debug(`Started ${shardCount} TTL consumers`);
  }

  async #startTtlConsumer(shard: number) {
    if (!this.options.ttlSystem) {
      return;
    }

    const pollIntervalMs = this.options.ttlSystem.pollIntervalMs ?? 1000;
    const batchSize = this.options.ttlSystem.batchSize ?? 100;
    let processedCount = 0;

    try {
      for await (const _ of setInterval(pollIntervalMs, null, {
        signal: this.abortController.signal,
      })) {
        const now = Date.now();

        const [error, expiredRuns] = await tryCatch(this.#expireTtlRuns(shard, now, batchSize));

        if (error) {
          this.logger.error(`Failed to expire TTL runs for shard ${shard}`, {
            error,
            service: this.name,
            shard,
          });
          continue;
        }

        if (expiredRuns.length > 0) {
          this.logger.debug(`Expired ${expiredRuns.length} TTL runs in shard ${shard}`, {
            service: this.name,
            shard,
            count: expiredRuns.length,
          });
          processedCount += expiredRuns.length;
        }
      }
    } catch (error) {
      if (error instanceof Error && error.name !== "AbortError") {
        throw error;
      }

      this.logger.debug(`TTL consumer ${shard} stopped`, {
        service: this.name,
        shard,
        processedCount,
      });
    }
  }

  /**
   * Atomically expire TTL runs: removes from TTL set, acknowledges from normal queue,
   * and enqueues each run to the TTL worker for DB updates.
   */
  async #expireTtlRuns(
    shard: number,
    now: number,
    batchSize: number
  ): Promise<Array<{ queueKey: string; runId: string; orgId: string }>> {
    const ttlSystem = this.options.ttlSystem;
    if (!ttlSystem) {
      return [];
    }

    const shardCount = ttlSystem.shardCount ?? this.shardCount;
    const ttlQueueKey = this.keys.ttlQueueKeyForShard(shard);
    const keyPrefix = this.options.redis.keyPrefix ?? "";
    const workerQueueKey = keyPrefix + ttlSystem.workerQueueSuffix;
    const workerItemsKey = keyPrefix + ttlSystem.workerItemsSuffix;
    const visibilityTimeoutMs = (ttlSystem.visibilityTimeoutMs ?? 30_000).toString();

    // Atomically get and remove expired runs from TTL set, ack them from normal queues, and enqueue to TTL worker
    const results = await this.redis.expireTtlRunsTracked(
      ttlQueueKey,
      keyPrefix,
      now.toString(),
      batchSize.toString(),
      shardCount.toString(),
      workerQueueKey,
      workerItemsKey,
      visibilityTimeoutMs
    );

    if (!results || results.length === 0) {
      return [];
    }

    // Parse the results: each item is "queueKey|runId|orgId"
    const expiredRuns = results.map((member: string) => {
      const [queueKey, runId, orgId] = member.split("|");
      return { queueKey, runId, orgId };
    });

    // Rebalance master queues for all affected queues.
    // Group by master queue key (derived from environment) since different queues
    // may belong to different master queue shards.
    const queuesByMasterKey = new Map<string, string[]>();

    for (const { queueKey } of expiredRuns) {
      const envId = this.keys.envIdFromQueue(queueKey);
      const masterQueueKey = this.keys.masterQueueKeyForEnvironment(envId, this.shardCount);

      const queues = queuesByMasterKey.get(masterQueueKey) ?? [];
      queues.push(queueKey);
      queuesByMasterKey.set(masterQueueKey, queues);
    }

    if (queuesByMasterKey.size > 0) {
      const pipeline = this.redis.pipeline();
      const keyPrefix = this.options.redis.keyPrefix ?? "";

      for (const [masterQueueKey, queueNames] of queuesByMasterKey) {
        // For CK queues, skip the legacy rebalance — the CK index was already
        // updated inside the Lua script, and migrateLegacyMasterQueues would
        // re-add the concrete :ck:bar member to the master queue.
        // Only rebalance non-CK queues here.
        const nonCkQueues = queueNames.filter((q) => !q.includes(":ck:"));
        if (nonCkQueues.length > 0) {
          const uniqueQueueNames = [...new Set(nonCkQueues)];
          pipeline.migrateLegacyMasterQueues(masterQueueKey, keyPrefix, ...uniqueQueueNames);
        }
      }

      await pipeline.exec();
    }

    return expiredRuns;
  }

  /**
   * Get the TTL shard for a queue key
   */
  #getTtlShardForQueue(queueKey: string): number {
    const { envId } = this.keys.descriptorFromQueue(queueKey);
    const shardCount = this.options.ttlSystem?.shardCount ?? this.shardCount;
    return this.keys.masterQueueShardForEnvironment(envId, shardCount);
  }

  async migrateLegacyMasterQueue(legacyMasterQueue: string) {
    const legacyMasterQueueKey = this.keys.legacyMasterQueueKey(legacyMasterQueue);

    this.logger.debug("Migrating legacy master queue", {
      legacyMasterQueueKey,
      service: this.name,
    });

    // Get all items from the legacy master queue
    const queueNames = await this.redis.zrange(legacyMasterQueueKey, 0, -1);

    this.logger.debug("Found items in legacy master queue", {
      queueNames,
      service: this.name,
    });

    // We need to group the items by the new masterQueueKey, so we need to extract out the environmentId from the queue name and calculate the shard
    const queuesByMasterQueueKey = new Map<string, string[]>();

    for (const queueName of queueNames) {
      const environmentId = this.keys.envIdFromQueue(queueName);
      const shard = this.keys.masterQueueShardForEnvironment(environmentId, this.shardCount);
      const masterQueueKey = this.keys.masterQueueKeyForShard(shard);
      queuesByMasterQueueKey.set(masterQueueKey, [
        ...(queuesByMasterQueueKey.get(masterQueueKey) ?? []),
        queueName,
      ]);
    }

    this.logger.debug("Grouping items by new master queue key", {
      queuesByMasterQueueKey: Object.fromEntries(queuesByMasterQueueKey.entries()),
      service: this.name,
    });

    const pipeline = this.redis.pipeline();

    for (const [masterQueueKey, queueNames] of queuesByMasterQueueKey) {
      pipeline.migrateLegacyMasterQueues(
        masterQueueKey,
        this.options.redis.keyPrefix ?? "",
        ...queueNames
      );
    }

    await pipeline.exec();

    this.logger.debug("Migrated legacy master queue", {
      legacyMasterQueueKey,
      service: this.name,
    });
  }

  // This is used for test purposes only
  async processMasterQueueForEnvironment(environmentId: string, maxCount: number = 10) {
    const shard = this.keys.masterQueueShardForEnvironment(environmentId, this.shardCount);

    return this.#processMasterQueueShard(shard, environmentId, maxCount);
  }

  // Test-only: directly dequeue from a queue (CK or non-CK) without enqueuing to worker queue
  async testDequeueFromMasterQueue(
    shard: number,
    environmentId: string,
    maxCount: number = 10
  ): Promise<DequeuedMessage[]> {
    const masterQueueKey = this.keys.masterQueueKeyForShard(shard);
    const envQueues = await this.queueSelectionStrategy.distributeFairQueuesFromParentQueue(
      masterQueueKey,
      environmentId
    );

    const allMessages: DequeuedMessage[] = [];

    for (const env of envQueues) {
      for (const queue of env.queues) {
        const messages = this.keys.isCkWildcard(queue)
          ? await this.#callDequeueMessagesFromCkQueue({
              ckWildcardQueue: queue,
              shard,
              maxCount,
            })
          : await this.#callDequeueMessagesFromQueue({
              messageQueue: queue,
              shard,
              maxCount,
            });
        allMessages.push(...messages);
      }
    }

    return allMessages;
  }

  async #processMasterQueueShard(shard: number, consumerId: string, maxCount: number = 10) {
    return this.#trace(
      "processMasterQueueShard",
      async (span) => {
        const masterQueueKey = this.keys.masterQueueKeyForShard(shard);

        const envQueues = await this.queueSelectionStrategy.distributeFairQueuesFromParentQueue(
          masterQueueKey,
          consumerId
        );

        span.setAttribute("master_queue_key", masterQueueKey);
        span.setAttribute("environment_count", envQueues.length);
        span.setAttribute("max_count", maxCount);

        if (envQueues.length === 0) {
          return [];
        }

        let attemptedEnvs = 0;
        let attemptedQueues = 0;
        let messagesDequeued = 0;
        let cooloffPeriodCount = 0;

        for (const env of envQueues) {
          attemptedEnvs++;

          for (const queue of env.queues) {
            const cooloffState = this._queueCooloffStates.get(queue) ?? {
              _tag: "normal",
              consecutiveFailures: 0,
            };

            if (cooloffState._tag === "cooloff") {
              if (cooloffState.cooloffExpiresAt > Date.now()) {
                cooloffPeriodCount++;
                continue;
              } else {
                this._queueCooloffStates.delete(queue);
              }
            }

            attemptedQueues++;

            // Attempt to dequeue from this queue
            // Dispatch CK wildcard queues to the CK-aware dequeue path
            const dequeuePromise = this.keys.isCkWildcard(queue)
              ? this.#callDequeueMessagesFromCkQueue({
                  ckWildcardQueue: queue,
                  shard,
                  maxCount,
                })
              : this.#callDequeueMessagesFromQueue({
                  messageQueue: queue,
                  shard,
                  maxCount,
                });

            const [error, messages] = await tryCatch(dequeuePromise);

            if (error) {
              this.logger.error(
                `[processMasterQueueShard][${this.name}] Failed to dequeue from queue`,
                {
                  error,
                  queue,
                }
              );

              continue;
            }

            if (messages.length > 0) {
              // Reset cooloff state on successful dequeue
              this._queueCooloffStates.delete(queue);
            }

            if (messages.length === 0) {
              if (cooloffState._tag === "normal") {
                const cooloffCountThreshold = Math.max(
                  1,
                  this.options.masterQueueCooloffCountThreshold ?? 10
                ); // Don't let the cooloff count be less than 1

                this.logger.debug("Evaluating cooloff state", {
                  queue,
                  cooloffState,
                  cooloffCountThreshold,
                  consecutiveFailures: cooloffState.consecutiveFailures,
                });

                if (cooloffState.consecutiveFailures >= cooloffCountThreshold) {
                  this.logger.debug("Setting cooloff state", {
                    queue,
                    cooloffState,
                    cooloffCountThreshold,
                    consecutiveFailures: cooloffState.consecutiveFailures,
                    cooloffExpiresAt: new Date(
                      Date.now() + (this.options.masterQueueCooloffPeriodMs ?? 10_000)
                    ),
                  });

                  this._queueCooloffStates.set(queue, {
                    _tag: "cooloff",
                    cooloffExpiresAt:
                      Date.now() + (this.options.masterQueueCooloffPeriodMs ?? 10_000),
                  });
                } else {
                  this._queueCooloffStates.set(queue, {
                    _tag: "normal",
                    consecutiveFailures: cooloffState.consecutiveFailures + 1,
                  });
                }
              }

              continue;
            }

            messagesDequeued += messages.length;

            await this.#enqueueMessagesToWorkerQueues(messages);
          }
        }

        span.setAttribute("attempted_envs", attemptedEnvs);
        span.setAttribute("attempted_queues", attemptedQueues);
        span.setAttribute("messages_dequeued", messagesDequeued);
      },
      {
        kind: SpanKind.CONSUMER,
        attributes: {
          [SEMATTRS_MESSAGING_OPERATION]: "receive",
          [SEMATTRS_MESSAGING_SYSTEM]: "runqueue",
        },
      }
    );
  }

  async #processQueueForWorkerQueue(queueKey: string, environmentId: string) {
    const shard = this.keys.masterQueueShardForEnvironment(environmentId, this.shardCount);

    this.logger.debug("processQueueForWorkerQueue", {
      queueKey,
      shard,
      service: this.name,
    });

    // Dispatch CK wildcard queues to the CK-aware dequeue path
    const messages = this.keys.isCkWildcard(queueKey)
      ? await this.#callDequeueMessagesFromCkQueue({
          ckWildcardQueue: queueKey,
          shard,
          maxCount: 10,
        })
      : await this.#callDequeueMessagesFromQueue({
          messageQueue: queueKey,
          shard,
          maxCount: 10,
        });

    if (messages.length === 0) {
      return;
    }

    await this.#enqueueMessagesToWorkerQueues(messages);

    if (messages.length === 10) {
      await this.#processQueueForWorkerQueue(queueKey, environmentId);
    }
  }

  async #enqueueMessagesToWorkerQueues(messages: DequeuedMessage[]) {
    await this.#trace("enqueueMessagesToWorkerQueues", async (span) => {
      span.setAttribute("message_count", messages.length);

      const pipeline = this.redis.pipeline();

      const operations = [];

      for (const message of messages) {
        const workerQueueKey = this.keys.workerQueueKey(
          this.#getWorkerQueueFromMessage(message.message)
        );

        const messageKeyValue = this.keys.messageKey(message.message.orgId, message.messageId);

        operations.push({
          workerQueueKey: workerQueueKey,
          messageId: message.messageId,
        });

        pipeline.rpush(workerQueueKey, messageKeyValue);
      }

      span.setAttribute("operations_count", operations.length);

      this.logger.info("enqueueMessagesToWorkerQueues", {
        service: this.name,
        operations,
      });

      await pipeline.exec();
    });
  }

  /**
   * Calls the appropriate Lua enqueue script variant (plain, TTL, CK, or TTL+CK).
   *
   * Each variant receives the same fast-path keys/args in addition to its own specific keys.
   * The Lua script returns 1 (fast path taken) or 0 (slow path taken).
   *
   * All four variants share this KEYS layout:
   *   KEYS[1..N]  = variant-specific keys (queue, message, concurrency, env, master, TTL, CK index)
   *   KEYS[N+1]   = workerQueueKey        (fast-path: RPUSH target)
   *   KEYS[N+2]   = queueConcurrencyLimitKey
   *   KEYS[N+3]   = envConcurrencyLimitKey
   *   KEYS[N+4]   = envConcurrencyLimitBurstFactorKey
   *
   * And this ARGV layout (appended after variant-specific args):
   *   messageKeyValue              (fast-path: value to RPUSH into worker queue)
   *   defaultEnvConcurrencyLimit   (fallback if no Redis key set)
   *   defaultEnvConcurrencyBurstFactor
   *   currentTime                  (ms timestamp for ZRANGEBYSCORE availability check)
   *   enableFastPath               ('1' or '0')
   *
   * @returns true if the fast path was taken (message pushed directly to worker queue)
   */
  async #callEnqueueMessage(
    message: OutputPayloadV2,
    ttlInfo?: {
      ttlExpiresAt: number;
      ttlQueueKey: string;
      ttlMember: string;
    },
    enableFastPath: boolean = false
  ): Promise<boolean> {
    // --- Slow-path keys (used by all variants) ---
    const queueKey = message.queue;
    const messageKey = this.keys.messageKey(message.orgId, message.runId);
    const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKeyFromQueue(message.queue);
    const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKeyFromQueue(message.queue);
    const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKeyFromQueue(message.queue);
    const envCurrentDequeuedKey = this.keys.envCurrentDequeuedKeyFromQueue(message.queue);
    const envQueueKey = this.keys.envQueueKeyFromQueue(message.queue);
    const masterQueueKey = this.keys.masterQueueKeyForEnvironment(
      message.environmentId,
      this.shardCount
    );

    // --- Fast-path keys (appended to each variant's KEYS) ---
    const workerQueueKey = this.keys.workerQueueKey(message.workerQueue);
    const queueConcurrencyLimitKey = this.keys.queueConcurrencyLimitKeyFromQueue(message.queue);
    const envConcurrencyLimitKey = this.keys.envConcurrencyLimitKeyFromQueue(message.queue);
    const envConcurrencyLimitBurstFactorKey = this.keys.envConcurrencyLimitBurstFactorKeyFromQueue(
      message.queue
    );
    // The value stored in the worker queue list — used to look up the message payload on dequeue
    const messageKeyValue = messageKey;

    const queueName = message.queue;
    const messageId = message.runId;
    const messageData = JSON.stringify(message);
    const messageScore = String(message.timestamp);
    const currentTime = String(Date.now());
    const enableFastPathArg = enableFastPath ? "1" : "0";
    const defaultEnvConcurrencyLimit = String(this.options.defaultEnvConcurrency);
    const defaultEnvConcurrencyBurstFactor = String(
      this.options.defaultEnvConcurrencyBurstFactor ?? 1.0
    );

    this.logger.debug("Calling enqueueMessage", {
      queueKey,
      messageKey,
      queueCurrentConcurrencyKey,
      envCurrentConcurrencyKey,
      queueCurrentDequeuedKey,
      envCurrentDequeuedKey,
      envQueueKey,
      queueName,
      messageId,
      messageData,
      messageScore,
      masterQueueKey,
      enableFastPath,
      ttlInfo,
      service: this.name,
    });

    let result: number;

    // Use CK-aware enqueue for messages with concurrency keys
    if (message.concurrencyKey) {
      const ckIndexKey = this.keys.ckIndexKeyFromQueue(message.queue);
      const ckWildcardName = this.keys.toCkWildcard(message.queue);
      const lengthCounterKey = this.keys.queueLengthCounterKeyFromQueue(message.queue);
      const baseQueueKey = this.keys.baseQueueKeyFromQueue(message.queue);
      const ckKeyPrefix = this.options.redis.keyPrefix ?? "";

      if (ttlInfo) {
        result = await this.redis.enqueueMessageWithTtlCkTracked(
          // keys
          masterQueueKey,
          queueKey,
          messageKey,
          queueCurrentConcurrencyKey,
          envCurrentConcurrencyKey,
          queueCurrentDequeuedKey,
          envCurrentDequeuedKey,
          envQueueKey,
          ttlInfo.ttlQueueKey,
          ckIndexKey,
          workerQueueKey,
          queueConcurrencyLimitKey,
          envConcurrencyLimitKey,
          envConcurrencyLimitBurstFactorKey,
          lengthCounterKey,
          baseQueueKey,
          // args
          queueName,
          messageId,
          messageData,
          messageScore,
          ttlInfo.ttlMember,
          String(ttlInfo.ttlExpiresAt),
          ckWildcardName,
          messageKeyValue,
          defaultEnvConcurrencyLimit,
          defaultEnvConcurrencyBurstFactor,
          currentTime,
          enableFastPathArg,
          ckKeyPrefix,
          String(this.counterTtlSeconds)
        );
      } else {
        result = await this.redis.enqueueMessageCkTracked(
          // keys
          masterQueueKey,
          queueKey,
          messageKey,
          queueCurrentConcurrencyKey,
          envCurrentConcurrencyKey,
          queueCurrentDequeuedKey,
          envCurrentDequeuedKey,
          envQueueKey,
          ckIndexKey,
          workerQueueKey,
          queueConcurrencyLimitKey,
          envConcurrencyLimitKey,
          envConcurrencyLimitBurstFactorKey,
          lengthCounterKey,
          baseQueueKey,
          // args
          queueName,
          messageId,
          messageData,
          messageScore,
          ckWildcardName,
          messageKeyValue,
          defaultEnvConcurrencyLimit,
          defaultEnvConcurrencyBurstFactor,
          currentTime,
          enableFastPathArg,
          ckKeyPrefix,
          String(this.counterTtlSeconds)
        );
      }
    } else if (ttlInfo) {
      // Use the TTL-aware enqueue that atomically adds to both queues
      result = await this.redis.enqueueMessageWithTtl(
        // keys
        masterQueueKey,
        queueKey,
        messageKey,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        queueCurrentDequeuedKey,
        envCurrentDequeuedKey,
        envQueueKey,
        ttlInfo.ttlQueueKey,
        workerQueueKey,
        queueConcurrencyLimitKey,
        envConcurrencyLimitKey,
        envConcurrencyLimitBurstFactorKey,
        // args
        queueName,
        messageId,
        messageData,
        messageScore,
        ttlInfo.ttlMember,
        String(ttlInfo.ttlExpiresAt),
        messageKeyValue,
        defaultEnvConcurrencyLimit,
        defaultEnvConcurrencyBurstFactor,
        currentTime,
        enableFastPathArg
      );
    } else {
      result = await this.redis.enqueueMessage(
        // keys
        masterQueueKey,
        queueKey,
        messageKey,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        queueCurrentDequeuedKey,
        envCurrentDequeuedKey,
        envQueueKey,
        workerQueueKey,
        queueConcurrencyLimitKey,
        envConcurrencyLimitKey,
        envConcurrencyLimitBurstFactorKey,
        // args
        queueName,
        messageId,
        messageData,
        messageScore,
        messageKeyValue,
        defaultEnvConcurrencyLimit,
        defaultEnvConcurrencyBurstFactor,
        currentTime,
        enableFastPathArg
      );
    }

    return result === 1;
  }

  async #callDequeueMessagesFromQueue({
    messageQueue,
    shard,
    maxCount,
  }: {
    messageQueue: string;
    shard: number;
    maxCount: number;
  }): Promise<DequeuedMessage[]> {
    return this.#trace("callDequeueMessagesFromQueue", async (span) => {
      span.setAttributes({
        messageQueue,
        shard,
        maxCount,
      });

      const queueConcurrencyLimitKey = this.keys.queueConcurrencyLimitKeyFromQueue(messageQueue);
      const queueCurrentConcurrencyKey =
        this.keys.queueCurrentConcurrencyKeyFromQueue(messageQueue);
      const envConcurrencyLimitKey = this.keys.envConcurrencyLimitKeyFromQueue(messageQueue);
      const envConcurrencyLimitBurstFactorKey =
        this.keys.envConcurrencyLimitBurstFactorKeyFromQueue(messageQueue);
      const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKeyFromQueue(messageQueue);
      const messageKeyPrefix = this.keys.messageKeyPrefixFromQueue(messageQueue);
      const envQueueKey = this.keys.envQueueKeyFromQueue(messageQueue);
      const masterQueueKey = this.keys.masterQueueKeyForShard(shard);

      // Get TTL queue key if TTL system is enabled
      const ttlShardCount = this.options.ttlSystem?.shardCount ?? this.shardCount;
      const ttlShard = this.keys.masterQueueShardForEnvironment(
        this.keys.envIdFromQueue(messageQueue),
        ttlShardCount
      );
      const ttlQueueKey = this.options.ttlSystem ? this.keys.ttlQueueKeyForShard(ttlShard) : "";

      this.logger.debug("#callDequeueMessagesFromQueue", {
        messageQueue,
        queueConcurrencyLimitKey,
        envConcurrencyLimitKey,
        envConcurrencyLimitBurstFactorKey,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        messageKeyPrefix,
        envQueueKey,
        masterQueueKey,
        ttlQueueKey,
        shard,
        maxCount,
      });

      const result = await this.redis.dequeueMessagesFromQueue(
        //keys
        messageQueue,
        queueConcurrencyLimitKey,
        envConcurrencyLimitKey,
        envConcurrencyLimitBurstFactorKey,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        messageKeyPrefix,
        envQueueKey,
        masterQueueKey,
        ttlQueueKey,
        //args
        messageQueue,
        String(Date.now()),
        String(this.options.defaultEnvConcurrency),
        String(this.options.defaultEnvConcurrencyBurstFactor ?? 1),
        this.options.redis.keyPrefix ?? "",
        String(maxCount)
      );

      if (!result) {
        span.setAttribute("message_count", 0);

        return [];
      }

      this.logger.debug("dequeueMessagesFromQueue raw result", {
        result,
        service: this.name,
      });

      const messages = [];
      for (let i = 0; i < result.length; i += 3) {
        const messageId = result[i];
        const messageScore = result[i + 1];
        const rawMessage = result[i + 2];

        //read message
        const parsedMessage = OutputPayload.safeParse(JSON.parse(rawMessage));
        if (!parsedMessage.success) {
          this.logger.error(`[${this.name}] Failed to parse message`, {
            messageId,
            error: parsedMessage.error,
            service: this.name,
          });

          continue;
        }

        const message = parsedMessage.data;

        messages.push({
          messageId,
          messageScore,
          message,
        });
      }

      this.logger.debug("dequeueMessagesFromQueue parsed result", {
        messages,
        service: this.name,
      });

      const filteredMessages = messages.filter(Boolean) as DequeuedMessage[];

      span.setAttribute("message_count", filteredMessages.length);

      return filteredMessages;
    });
  }

  async #callDequeueMessagesFromCkQueue({
    ckWildcardQueue,
    shard,
    maxCount,
  }: {
    ckWildcardQueue: string;
    shard: number;
    maxCount: number;
  }): Promise<DequeuedMessage[]> {
    return this.#trace("callDequeueMessagesFromCkQueue", async (span) => {
      span.setAttributes({
        ckWildcardQueue,
        shard,
        maxCount,
      });

      const ckIndexKey = this.keys.ckIndexKeyFromQueue(ckWildcardQueue);
      const queueConcurrencyLimitKey = this.keys.queueConcurrencyLimitKeyFromQueue(ckWildcardQueue);
      const envConcurrencyLimitKey = this.keys.envConcurrencyLimitKeyFromQueue(ckWildcardQueue);
      const envConcurrencyLimitBurstFactorKey =
        this.keys.envConcurrencyLimitBurstFactorKeyFromQueue(ckWildcardQueue);
      const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKeyFromQueue(ckWildcardQueue);
      const messageKeyPrefix = this.keys.messageKeyPrefixFromQueue(ckWildcardQueue);
      const envQueueKey = this.keys.envQueueKeyFromQueue(ckWildcardQueue);
      const masterQueueKey = this.keys.masterQueueKeyForShard(shard);

      // Get TTL queue key if TTL system is enabled
      const ttlShardCount = this.options.ttlSystem?.shardCount ?? this.shardCount;
      const ttlShard = this.keys.masterQueueShardForEnvironment(
        this.keys.envIdFromQueue(ckWildcardQueue),
        ttlShardCount
      );
      const ttlQueueKey = this.options.ttlSystem ? this.keys.ttlQueueKeyForShard(ttlShard) : "";

      this.logger.debug("#callDequeueMessagesFromCkQueue", {
        ckWildcardQueue,
        ckIndexKey,
        queueConcurrencyLimitKey,
        envConcurrencyLimitKey,
        envCurrentConcurrencyKey,
        messageKeyPrefix,
        envQueueKey,
        masterQueueKey,
        ttlQueueKey,
        shard,
        maxCount,
      });

      const lengthCounterKey = this.keys.queueLengthCounterKeyFromQueue(ckWildcardQueue);

      const result = await this.redis.dequeueMessagesFromCkQueueTracked(
        //keys
        ckIndexKey,
        queueConcurrencyLimitKey,
        envConcurrencyLimitKey,
        envConcurrencyLimitBurstFactorKey,
        envCurrentConcurrencyKey,
        messageKeyPrefix,
        envQueueKey,
        masterQueueKey,
        ttlQueueKey,
        lengthCounterKey,
        //args
        ckWildcardQueue,
        String(Date.now()),
        String(this.options.defaultEnvConcurrency),
        String(this.options.defaultEnvConcurrencyBurstFactor ?? 1),
        this.options.redis.keyPrefix ?? "",
        String(maxCount)
      );

      if (!result) {
        span.setAttribute("message_count", 0);
        return [];
      }

      this.logger.debug("dequeueMessagesFromCkQueue raw result", {
        result,
        service: this.name,
      });

      const messages = [];
      for (let i = 0; i < result.length; i += 3) {
        const messageId = result[i];
        const messageScore = result[i + 1];
        const rawMessage = result[i + 2];

        const parsedMessage = OutputPayload.safeParse(JSON.parse(rawMessage));
        if (!parsedMessage.success) {
          this.logger.error(`[${this.name}] Failed to parse CK message`, {
            messageId,
            error: parsedMessage.error,
            service: this.name,
          });
          continue;
        }

        messages.push({
          messageId,
          messageScore,
          message: parsedMessage.data,
        });
      }

      const filteredMessages = messages.filter(Boolean) as DequeuedMessage[];
      span.setAttribute("message_count", filteredMessages.length);
      return filteredMessages;
    });
  }

  async #callDequeueMessageFromWorkerQueue({
    workerQueue,
    blockingPop,
    blockingPopTimeoutSeconds,
  }: {
    workerQueue: string;
    blockingPop: boolean;
    blockingPopTimeoutSeconds: number;
  }): Promise<DequeuedMessage | undefined> {
    const workerQueueKey = this.keys.workerQueueKey(workerQueue);

    if (blockingPop) {
      this.logger.debug("#callDequeueMessageFromWorkerQueue blocking pop", {
        workerQueue,
        workerQueueKey,
        blockingPopTimeoutSeconds,
      });

      if (this.abortController.signal.aborted) {
        return;
      }

      const blockingClient = this.#createBlockingDequeueClient();

      async function cleanup() {
        await blockingClient.quit();
      }

      this.abortController.signal.addEventListener("abort", cleanup);

      const result = await this.#trace("popMessageFromWorkerQueue", async (span) => {
        span.setAttributes({
          workerQueue,
          workerQueueKey,
          blockingPopTimeoutSeconds,
          blocking: true,
        });

        return await blockingClient.blpop(workerQueueKey, blockingPopTimeoutSeconds);
      });

      this.abortController.signal.removeEventListener("abort", cleanup);

      cleanup().then(() => {
        this.logger.debug("dequeueMessageFromWorkerQueue cleanup", {
          service: this.name,
        });
      });

      if (!result) {
        return;
      }

      this.logger.debug("dequeueMessageFromWorkerQueue raw result", {
        result,
        service: this.name,
      });

      if (result.length !== 2) {
        this.logger.error("Invalid dequeue message from worker queue result", {
          result,
          service: this.name,
        });
        return;
      }

      // Make sure they are both strings
      if (typeof result[0] !== "string" || typeof result[1] !== "string") {
        this.logger.error("Invalid dequeue message from worker queue result", {
          result,
          service: this.name,
        });
        return;
      }

      const [, messageKey] = result;

      const workerQueueLength = await this.#trace("getWorkerQueueLength", async (span) => {
        span.setAttributes({
          workerQueue,
          workerQueueKey,
        });

        return await this.redis.llen(workerQueueKey);
      });

      const message = await this.#dequeueMessageFromKey(messageKey);

      if (!message) {
        this.logger.error("Failed to dequeue message from worker queue", {
          messageKey,
          workerQueue,
          workerQueueKey,
          workerQueueLength,
          service: this.name,
        });

        return;
      }

      return {
        messageId: message.runId,
        messageScore: String(message.timestamp),
        message,
        workerQueueLength,
      };
    } else {
      this.logger.debug("#callDequeueMessageFromWorkerQueue non-blocking pop", {
        workerQueue,
        workerQueueKey,
      });

      const result = await this.#trace("popMessageFromWorkerQueue", async (span) => {
        span.setAttributes({
          workerQueue,
          workerQueueKey,
          blocking: false,
        });

        return await this.redis.dequeueMessageFromWorkerQueueNonBlocking(workerQueueKey);
      });

      if (!result) {
        return;
      }

      const [messageKey, workerQueueLength] = result;

      const message = await this.#dequeueMessageFromKey(messageKey);

      if (!message) {
        return;
      }

      return {
        messageId: message.runId,
        messageScore: String(message.timestamp),
        message,
        workerQueueLength: Number(workerQueueLength),
      };
    }
  }

  async #callAcknowledgeMessage({
    message,
    removeFromWorkerQueue,
  }: {
    message: OutputPayload;
    removeFromWorkerQueue?: boolean;
  }) {
    const messageId = message.runId;
    const messageKey = this.keys.messageKey(message.orgId, messageId);
    const messageQueue = message.queue;
    const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKeyFromQueue(message.queue);
    const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKeyFromQueue(message.queue);
    const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKeyFromQueue(message.queue);
    const envCurrentDequeuedKey = this.keys.envCurrentDequeuedKeyFromQueue(message.queue);
    const envQueueKey = this.keys.envQueueKeyFromQueue(message.queue);
    const masterQueueKey = this.keys.masterQueueKeyForEnvironment(
      message.environmentId,
      this.shardCount
    );
    const workerQueue = this.#getWorkerQueueFromMessage(message);
    const workerQueueKey = this.keys.workerQueueKey(workerQueue);
    const messageKeyValue = this.keys.messageKey(message.orgId, messageId);

    this.logger.debug("Calling acknowledgeMessage", {
      messageKey,
      messageQueue,
      queueCurrentConcurrencyKey,
      envCurrentConcurrencyKey,
      queueCurrentDequeuedKey,
      envCurrentDequeuedKey,
      envQueueKey,
      messageId,
      masterQueueKey,
      workerQueue,
      workerQueueKey,
      removeFromWorkerQueue,
      messageKeyValue,
      service: this.name,
    });

    if (message.concurrencyKey) {
      const ckIndexKey = this.keys.ckIndexKeyFromQueue(message.queue);
      const ckWildcardName = this.keys.toCkWildcard(message.queue);
      const lengthCounterKey = this.keys.queueLengthCounterKeyFromQueue(message.queue);
      const runningCounterKey = this.keys.queueRunningCounterKeyFromQueue(message.queue);

      return this.redis.acknowledgeMessageCkTracked(
        masterQueueKey,
        messageKey,
        messageQueue,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        queueCurrentDequeuedKey,
        envCurrentDequeuedKey,
        envQueueKey,
        workerQueueKey,
        ckIndexKey,
        lengthCounterKey,
        runningCounterKey,
        messageId,
        messageQueue,
        messageKeyValue,
        removeFromWorkerQueue ? "1" : "0",
        ckWildcardName
      );
    }

    return this.redis.acknowledgeMessage(
      masterQueueKey,
      messageKey,
      messageQueue,
      queueCurrentConcurrencyKey,
      envCurrentConcurrencyKey,
      queueCurrentDequeuedKey,
      envCurrentDequeuedKey,
      envQueueKey,
      workerQueueKey,
      messageId,
      messageQueue,
      messageKeyValue,
      removeFromWorkerQueue ? "1" : "0"
    );
  }

  async #callClearMessageFromConcurrencySets({
    runId,
    orgId,
    queue,
    env,
  }: {
    runId: string;
    orgId: string;
    queue: string;
    env: RunQueueKeyProducerEnvironment;
  }) {
    const messageId = runId;
    const messageKey = this.keys.messageKey(orgId, messageId);
    const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKey(env, queue);
    const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKey(env);
    const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKey(env, queue);
    const envCurrentDequeuedKey = this.keys.envCurrentDequeuedKey(env);

    this.logger.debug("Calling clearMessageFromConcurrencySets", {
      messageKey,
      queue,
      env,
      queueCurrentConcurrencyKey,
      envCurrentConcurrencyKey,
      queueCurrentDequeuedKey,
      envCurrentDequeuedKey,
      messageId,
      service: this.name,
    });

    if (queue.includes(":ck:")) {
      return this.redis.clearMessageFromConcurrencySetsTracked(
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        queueCurrentDequeuedKey,
        envCurrentDequeuedKey,
        this.keys.queueRunningCounterKeyFromQueue(queue),
        this.keys.ckIndexKeyFromQueue(queue),
        messageId,
        this.options.redis.keyPrefix ?? "",
        String(this.counterTtlSeconds)
      );
    }

    return this.redis.clearMessageFromConcurrencySets(
      queueCurrentConcurrencyKey,
      envCurrentConcurrencyKey,
      queueCurrentDequeuedKey,
      envCurrentDequeuedKey,
      messageId
    );
  }

  async #callNackMessage({ message, retryAt }: { message: OutputPayload; retryAt?: number }) {
    const messageId = message.runId;
    const messageKey = this.keys.messageKey(message.orgId, message.runId);
    const messageQueue = message.queue;
    const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKeyFromQueue(message.queue);
    const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKeyFromQueue(message.queue);
    const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKeyFromQueue(message.queue);
    const envCurrentDequeuedKey = this.keys.envCurrentDequeuedKeyFromQueue(message.queue);
    const envQueueKey = this.keys.envQueueKeyFromQueue(message.queue);
    const masterQueueKey = this.keys.masterQueueKeyForEnvironment(
      message.environmentId,
      this.shardCount
    );

    const nextRetryDelay = calculateNextRetryDelay(this.retryOptions, message.attempt);
    const messageScore = retryAt ?? (nextRetryDelay ? Date.now() + nextRetryDelay : Date.now());

    this.logger.debug("Calling nackMessage", {
      messageKey,
      messageQueue,
      masterQueueKey,
      queueCurrentConcurrencyKey,
      envCurrentConcurrencyKey,
      queueCurrentDequeuedKey,
      envCurrentDequeuedKey,
      envQueueKey,
      messageId,
      messageScore,
      attempt: message.attempt,
      service: this.name,
    });

    if (message.concurrencyKey) {
      const ckIndexKey = this.keys.ckIndexKeyFromQueue(message.queue);
      const ckWildcardName = this.keys.toCkWildcard(message.queue);
      const lengthCounterKey = this.keys.queueLengthCounterKeyFromQueue(message.queue);
      const runningCounterKey = this.keys.queueRunningCounterKeyFromQueue(message.queue);

      await this.redis.nackMessageCkTracked(
        //keys
        masterQueueKey,
        messageKey,
        messageQueue,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        queueCurrentDequeuedKey,
        envCurrentDequeuedKey,
        envQueueKey,
        ckIndexKey,
        lengthCounterKey,
        runningCounterKey,
        //args
        messageId,
        messageQueue,
        JSON.stringify(message),
        String(messageScore),
        ckWildcardName,
        this.options.redis.keyPrefix ?? "",
        String(this.counterTtlSeconds)
      );
    } else {
      await this.redis.nackMessage(
        //keys
        masterQueueKey,
        messageKey,
        messageQueue,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        queueCurrentDequeuedKey,
        envCurrentDequeuedKey,
        envQueueKey,
        //args
        messageId,
        messageQueue,
        JSON.stringify(message),
        String(messageScore)
      );
    }
  }

  async #callMoveToDeadLetterQueue({ message }: { message: OutputPayload }) {
    const messageId = message.runId;
    const messageKey = this.keys.messageKey(message.orgId, message.runId);
    const messageQueue = message.queue;
    const queueCurrentConcurrencyKey = this.keys.queueCurrentConcurrencyKeyFromQueue(message.queue);
    const envCurrentConcurrencyKey = this.keys.envCurrentConcurrencyKeyFromQueue(message.queue);
    const queueCurrentDequeuedKey = this.keys.queueCurrentDequeuedKeyFromQueue(message.queue);
    const envCurrentDequeuedKey = this.keys.envCurrentDequeuedKeyFromQueue(message.queue);
    const envQueueKey = this.keys.envQueueKeyFromQueue(message.queue);
    const deadLetterQueueKey = this.keys.deadLetterQueueKeyFromQueue(message.queue);
    const masterQueueKey = this.keys.masterQueueKeyForEnvironment(
      message.environmentId,
      this.shardCount
    );

    if (message.concurrencyKey) {
      const ckIndexKey = this.keys.ckIndexKeyFromQueue(message.queue);
      const ckWildcardName = this.keys.toCkWildcard(message.queue);
      const lengthCounterKey = this.keys.queueLengthCounterKeyFromQueue(message.queue);
      const runningCounterKey = this.keys.queueRunningCounterKeyFromQueue(message.queue);

      await this.redis.moveToDeadLetterQueueCkTracked(
        masterQueueKey,
        messageKey,
        messageQueue,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        queueCurrentDequeuedKey,
        envCurrentDequeuedKey,
        envQueueKey,
        deadLetterQueueKey,
        ckIndexKey,
        lengthCounterKey,
        runningCounterKey,
        messageId,
        messageQueue,
        ckWildcardName
      );
    } else {
      await this.redis.moveToDeadLetterQueue(
        masterQueueKey,
        messageKey,
        messageQueue,
        queueCurrentConcurrencyKey,
        envCurrentConcurrencyKey,
        queueCurrentDequeuedKey,
        envCurrentDequeuedKey,
        envQueueKey,
        deadLetterQueueKey,
        messageId,
        messageQueue
      );
    }
  }

  #callUpdateEnvironmentConcurrencyLimits({
    envConcurrencyLimitKey,
    envConcurrencyLimit,
    envConcurrencyLimitBurstFactorKey,
    envConcurrencyLimitBurstFactor,
  }: {
    envConcurrencyLimitKey: string;
    envConcurrencyLimit: number;
    envConcurrencyLimitBurstFactorKey: string;
    envConcurrencyLimitBurstFactor: string;
  }) {
    return this.redis.updateEnvironmentConcurrencyLimits(
      envConcurrencyLimitKey,
      envConcurrencyLimitBurstFactorKey,
      String(envConcurrencyLimit),
      envConcurrencyLimitBurstFactor
    );
  }

  #getWorkerQueueFromMessage(message: OutputPayload): string {
    return this.workerQueueResolver.getWorkerQueueFromMessage(message);
  }

  #createBlockingDequeueClient() {
    const blockingClient = this.redis.duplicate();

    return blockingClient;
  }

  // Call this every 10 minutes
  private async scanConcurrencySets() {
    if (this.abortController.signal.aborted) {
      this.logger.info("Abort signal received, skipping concurrency scan");

      return;
    }

    this.logger.debug("Scanning concurrency sets for completed runs");

    const stats = {
      streamCallbacks: 0,
      processedKeys: 0,
    };

    const { promise, resolve, reject } = promiseWithResolvers<typeof stats>();

    const { stream, redis } = this.currentConcurrencyScanStream(
      10,
      () => {
        this.logger.debug("Concurrency scan stream closed", { stats });

        resolve(stats);
      },
      (error) => {
        this.logger.error("Concurrency scan stream error", {
          stats,
          error: {
            name: error.name,
            message: error.message,
            stack: error.stack,
          },
        });

        reject(error);
      }
    );

    stream.on("data", async (keys: string[]) => {
      if (!keys || keys.length === 0) {
        return;
      }

      stream.pause();

      if (this.abortController.signal.aborted) {
        stream.destroy();
        return;
      }

      stats.streamCallbacks++;

      const uniqueKeys = Array.from(new Set<string>(keys)).map((key) =>
        key.replace(redis.options.keyPrefix ?? "", "")
      );

      if (uniqueKeys.length === 0) {
        stream.resume();
        return;
      }

      this.logger.debug("Processing concurrency keys from stream", {
        keys: uniqueKeys,
      });

      stats.processedKeys += uniqueKeys.length;

      await Promise.allSettled(uniqueKeys.map((key) => this.processConcurrencySet(key))).finally(
        () => {
          stream.resume();
        }
      );
    });

    return promise;
  }

  private async processConcurrencySet(concurrencyKey: string) {
    const stream = this.redis.sscanStream(concurrencyKey, {
      count: 100,
    });

    const { promise, resolve, reject } = promiseWithResolvers<void>();

    stream.on("end", () => {
      resolve();
    });

    stream.on("error", (error) => {
      this.logger.error("Error in sscanStream for concurrency set", {
        concurrencyKey,
        error,
      });

      reject(error);
    });

    stream.on("data", async (runIds: string[]) => {
      stream.pause();

      if (this.abortController.signal.aborted) {
        stream.destroy();
        return;
      }

      if (!runIds || runIds.length === 0) {
        stream.resume();
        return;
      }

      const deduplicatedRunIds = Array.from(new Set(runIds));

      const [processError] = await tryCatch(
        this.processCurrentConcurrencyRunIds(concurrencyKey, deduplicatedRunIds)
      );

      if (processError) {
        this.logger.error("Error processing concurrency set", {
          concurrencyKey,
          runIds,
          error: processError,
        });
      }

      stream.resume();
    });

    return promise;
  }

  private async processCurrentConcurrencyRunIds(concurrencyKey: string, runIds: string[]) {
    this.logger.debug("Processing concurrency set with runs", {
      concurrencyKey,
      runIds: runIds.slice(0, 5),
      runIdsLength: runIds.length,
    });

    // Call the callback to determine which runs are completed
    const completedRuns = await this.options.concurrencySweeper?.callback(runIds);

    if (!completedRuns) {
      this.logger.debug("No completed runs found in concurrency set", { concurrencyKey });
      return;
    }

    if (completedRuns.length === 0) {
      this.logger.debug("No completed runs found in concurrency set", { concurrencyKey });
      return;
    }

    this.logger.info("Found completed runs to mark for ack", {
      concurrencyKey,
      completedRunIds: completedRuns.map((r) => r.id).slice(0, 5),
      completedRunIdsLength: completedRuns.length,
    });

    // Mark the completed runs for acknowledgment
    await this.markRunsForAck(completedRuns);
  }

  private async markRunsForAck(completedRuns: Array<{ id: string; orgId: string }>) {
    const markedForAckKey = this.keys.markedForAckKey();

    // Prepare arguments: alternating orgId, messageId pairs
    const args: Array<number | string> = [];
    for (const run of completedRuns) {
      this.logger.info("Marking run for acknowledgment", {
        orgId: run.orgId,
        runId: run.id,
      });

      args.push(Date.now());
      args.push(`${run.orgId}:${run.id}`);
    }

    const count = await this.redis.zadd(markedForAckKey, ...args);

    this.logger.info("Marked runs for acknowledgment", {
      markedForAckKey,
      count,
    });
  }

  // Call this every 5 minutes
  private async processMarkedRuns() {
    if (this.abortController.signal.aborted) {
      return;
    }

    try {
      const markedForAckKey = this.keys.markedForAckKey();
      const results = await this.redis.getMarkedRunsForAck(markedForAckKey, "100");

      if (results.length === 0) {
        return;
      }

      const markedRuns: MarkedRun[] = [];

      // Parse results: [orgId1, messageId1, score1, orgId2, messageId2, score2, ...]
      for (let i = 0; i < results.length; i += 3) {
        markedRuns.push({
          orgId: results[i],
          messageId: results[i + 1],
          score: Number(results[i + 2]),
        });
      }

      this.logger.debug(`Processing ${markedRuns.length} marked runs for acknowledgment`, {
        markedRuns: markedRuns, // Log first 3 for debugging
      });

      for (const run of markedRuns) {
        const [processError] = await tryCatch(this.processMarkedRun(run));

        if (processError) {
          this.logger.error("Error processing marked run", {
            error: processError,
            orgId: run.orgId,
            messageId: run.messageId,
          });
        }
      }

      const shouldProcessMoreRuns = (await this.redis.zcard(markedForAckKey)) > 0;

      if (shouldProcessMoreRuns) {
        await setTimeout(1000);
        await this.processMarkedRuns();
      }
    } catch (error) {
      this.logger.error("Error processing marked runs", { error });
    }
  }

  async processMarkedRun(run: MarkedRun) {
    this.logger.info("Acknowledging marked run", {
      orgId: run.orgId,
      messageId: run.messageId,
    });

    await this.acknowledgeMessage(run.orgId, run.messageId, {
      skipDequeueProcessing: true,
      removeFromWorkerQueue: false,
    });
  }

  async #dequeueMessageFromKey(messageKey: string) {
    return this.#trace("dequeueMessageFromKey", async (span) => {
      span.setAttributes({
        messageKey,
      });

      const rawMessage = await this.redis.dequeueMessageFromKeyTracked(
        messageKey,
        this.options.redis.keyPrefix ?? "",
        String(this.counterTtlSeconds)
      );

      if (!rawMessage) {
        span.setAttribute("result", "NO_MESSAGE");

        return;
      }

      const [error, message] = parseRawMessage(rawMessage);

      if (error) {
        this.logger.error(`[${this.name}] Failed to parse message`, {
          messageKey,
          error,
          service: this.name,
          message: message ?? rawMessage,
        });
      }

      if (message) {
        span.setAttribute("result", "SUCCESS");
        span.setAttribute("messageId", message.runId);
      } else {
        span.setAttribute("result", "NO_MESSAGE");
      }

      return message;
    });
  }

  #registerCommands() {
    this.redis.defineCommand("migrateLegacyMasterQueues", {
      numberOfKeys: 1,
      lua: `
local masterQueueKey = KEYS[1]

local keyPrefix = ARGV[1]

for i = 2, #ARGV do
  local queueName = ARGV[i]
  local queueKey = keyPrefix .. queueName
  
  -- Rebalance the parent queues
  local earliestMessage = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')

  if #earliestMessage == 0 then
    redis.call('ZREM', masterQueueKey, queueName)
  else
    redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
  end
end
      `,
    });

    // Enqueue message with optional fast path.
    //
    // Returns 1 (fast path: message pushed directly to worker queue) or 0 (slow path: message
    // added to queue sorted set, needs processQueueForWorkerQueue to move it later).
    //
    // Fast-path conditions (all must be true):
    //   1. enableFastPath == '1' (gated per WorkerInstanceGroup)
    //   2. No available messages in the queue (ZRANGEBYSCORE finds nothing with score <= now)
    //   3. Env concurrency has capacity (SCARD < limit * burstFactor)
    //   4. Queue concurrency has capacity (SCARD < min(queueLimit, envLimit))
    //
    // The fast path performs the same concurrency bookkeeping (SADD to currentConcurrency sets)
    // as dequeueMessagesFromQueue, so ack/nack/release work identically for both paths.
    //
    // When enableFastPath == '0', the script skips the fast-path check entirely and behaves
    // identically to the pre-fast-path version (with the addition of returning 0).
    this.redis.defineCommand("enqueueMessage", {
      numberOfKeys: 12,
      lua: `
local masterQueueKey = KEYS[1]
local queueKey = KEYS[2]
local messageKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
-- Fast-path keys (KEYS 9-12)
local workerQueueKey = KEYS[9]
local queueConcurrencyLimitKey = KEYS[10]
local envConcurrencyLimitKey = KEYS[11]
local envConcurrencyLimitBurstFactorKey = KEYS[12]

local queueName = ARGV[1]
local messageId = ARGV[2]
local messageData = ARGV[3]
local messageScore = ARGV[4]
-- Fast-path args (ARGV 5-9)
local messageKeyValue = ARGV[5]
local defaultEnvConcurrencyLimit = ARGV[6]
local defaultEnvConcurrencyBurstFactor = ARGV[7]
local currentTime = ARGV[8]
local enableFastPath = ARGV[9]

-- Fast path: check if we can skip the queue and go directly to worker queue
if enableFastPath == '1' then
  local available = redis.call('ZRANGEBYSCORE', queueKey, '-inf', currentTime, 'LIMIT', 0, 1)
  if #available == 0 then
    local envCurrent = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
    local envLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
    local envBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
    local envLimitWithBurst = math.floor(envLimit * envBurstFactor)

    if envCurrent < envLimitWithBurst then
      local queueCurrent = tonumber(redis.call('SCARD', queueCurrentConcurrencyKey) or '0')
      local queueLimit = math.min(
        tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
        envLimit
      )

      if queueCurrent < queueLimit then
        redis.call('SET', messageKey, messageData)
        redis.call('SADD', queueCurrentConcurrencyKey, messageId)
        redis.call('SADD', envCurrentConcurrencyKey, messageId)
        redis.call('RPUSH', workerQueueKey, messageKeyValue)
        return 1
      end
    end
  end
end

-- Slow path: normal enqueue
-- Write the message to the message key
redis.call('SET', messageKey, messageData)

-- Add the message to the queue
redis.call('ZADD', queueKey, messageScore, messageId)

-- Add the message to the env queue
redis.call('ZADD', envQueueKey, messageScore, messageId)

-- Rebalance the parent queues
local earliestMessage = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')

if #earliestMessage == 0 then
  redis.call('ZREM', masterQueueKey, queueName)
else
  redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
end

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

return 0
      `,
    });

    // Enqueue with TTL tracking. Same fast-path logic as enqueueMessage.
    // On fast path, the TTL sorted set is intentionally skipped — the expireRun worker job
    // (scheduled independently before enqueue) handles TTL expiry. This mirrors what
    // dequeueMessagesFromQueue does: it removes from the TTL set when dequeuing.
    this.redis.defineCommand("enqueueMessageWithTtl", {
      numberOfKeys: 13,
      lua: `
local masterQueueKey = KEYS[1]
local queueKey = KEYS[2]
local messageKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local ttlQueueKey = KEYS[9]
-- Fast-path keys (KEYS 10-13)
local workerQueueKey = KEYS[10]
local queueConcurrencyLimitKey = KEYS[11]
local envConcurrencyLimitKey = KEYS[12]
local envConcurrencyLimitBurstFactorKey = KEYS[13]

local queueName = ARGV[1]
local messageId = ARGV[2]
local messageData = ARGV[3]
local messageScore = ARGV[4]
local ttlMember = ARGV[5]
local ttlScore = ARGV[6]
-- Fast-path args (ARGV 7-11)
local messageKeyValue = ARGV[7]
local defaultEnvConcurrencyLimit = ARGV[8]
local defaultEnvConcurrencyBurstFactor = ARGV[9]
local currentTime = ARGV[10]
local enableFastPath = ARGV[11]

-- Fast path: check if we can skip the queue and go directly to worker queue
if enableFastPath == '1' then
  local available = redis.call('ZRANGEBYSCORE', queueKey, '-inf', currentTime, 'LIMIT', 0, 1)
  if #available == 0 then
    local envCurrent = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
    local envLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
    local envBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
    local envLimitWithBurst = math.floor(envLimit * envBurstFactor)

    if envCurrent < envLimitWithBurst then
      local queueCurrent = tonumber(redis.call('SCARD', queueCurrentConcurrencyKey) or '0')
      local queueLimit = math.min(
        tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
        envLimit
      )

      if queueCurrent < queueLimit then
        redis.call('SET', messageKey, messageData)
        redis.call('SADD', queueCurrentConcurrencyKey, messageId)
        redis.call('SADD', envCurrentConcurrencyKey, messageId)
        redis.call('RPUSH', workerQueueKey, messageKeyValue)
        -- Skip TTL sorted set: the expireRun worker job handles TTL expiry independently
        return 1
      end
    end
  end
end

-- Slow path: normal enqueue
-- Write the message to the message key
redis.call('SET', messageKey, messageData)

-- Add the message to the queue
redis.call('ZADD', queueKey, messageScore, messageId)

-- Add the message to the env queue
redis.call('ZADD', envQueueKey, messageScore, messageId)

-- Add to TTL sorted set
redis.call('ZADD', ttlQueueKey, ttlScore, ttlMember)

-- Rebalance the parent queues
local earliestMessage = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')

if #earliestMessage == 0 then
  redis.call('ZREM', masterQueueKey, queueName)
else
  redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
end

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

return 0
      `,
    });

    // CK-aware enqueue: adds to CK index + master queue with :ck:* member.
    // Same fast-path logic as enqueueMessage. For CK queues, the fast-path checks the
    // per-CK sub-queue for available messages and the per-CK concurrency limit.
    this.redis.defineCommand("enqueueMessageCk", {
      numberOfKeys: 13,
      lua: `
local masterQueueKey = KEYS[1]
local queueKey = KEYS[2]
local messageKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local ckIndexKey = KEYS[9]
-- Fast-path keys (KEYS 10-13)
local workerQueueKey = KEYS[10]
local queueConcurrencyLimitKey = KEYS[11]
local envConcurrencyLimitKey = KEYS[12]
local envConcurrencyLimitBurstFactorKey = KEYS[13]

local queueName = ARGV[1]
local messageId = ARGV[2]
local messageData = ARGV[3]
local messageScore = ARGV[4]
local ckWildcardName = ARGV[5]
-- Fast-path args (ARGV 6-10)
local messageKeyValue = ARGV[6]
local defaultEnvConcurrencyLimit = ARGV[7]
local defaultEnvConcurrencyBurstFactor = ARGV[8]
local currentTime = ARGV[9]
local enableFastPath = ARGV[10]

-- Fast path: check if we can skip the queue and go directly to worker queue
if enableFastPath == '1' then
  local available = redis.call('ZRANGEBYSCORE', queueKey, '-inf', currentTime, 'LIMIT', 0, 1)
  if #available == 0 then
    local envCurrent = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
    local envLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
    local envBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
    local envLimitWithBurst = math.floor(envLimit * envBurstFactor)

    if envCurrent < envLimitWithBurst then
      -- For CK queues, check per-CK concurrency (same key as queue concurrency)
      local queueCurrent = tonumber(redis.call('SCARD', queueCurrentConcurrencyKey) or '0')
      local queueLimit = math.min(
        tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
        envLimit
      )

      if queueCurrent < queueLimit then
        redis.call('SET', messageKey, messageData)
        redis.call('SADD', queueCurrentConcurrencyKey, messageId)
        redis.call('SADD', envCurrentConcurrencyKey, messageId)
        redis.call('RPUSH', workerQueueKey, messageKeyValue)
        return 1
      end
    end
  end
end

-- Slow path: normal enqueue
-- Write the message to the message key
redis.call('SET', messageKey, messageData)

-- Add the message to the CK-specific queue
redis.call('ZADD', queueKey, messageScore, messageId)

-- Add the message to the env queue
redis.call('ZADD', envQueueKey, messageScore, messageId)

-- Rebalance CK index
local earliest = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')
if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], queueName)
end

-- Rebalance master queue with ck:* member
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx > 0 then
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, queueName)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

return 0
      `,
    });

    // CK-aware enqueue with TTL tracking. Combines CK and TTL behavior with fast path.
    // On fast path: skips both the CK sub-queue and TTL sorted set.
    this.redis.defineCommand("enqueueMessageWithTtlCk", {
      numberOfKeys: 14,
      lua: `
local masterQueueKey = KEYS[1]
local queueKey = KEYS[2]
local messageKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local ttlQueueKey = KEYS[9]
local ckIndexKey = KEYS[10]
-- Fast-path keys (KEYS 11-14)
local workerQueueKey = KEYS[11]
local queueConcurrencyLimitKey = KEYS[12]
local envConcurrencyLimitKey = KEYS[13]
local envConcurrencyLimitBurstFactorKey = KEYS[14]

local queueName = ARGV[1]
local messageId = ARGV[2]
local messageData = ARGV[3]
local messageScore = ARGV[4]
local ttlMember = ARGV[5]
local ttlScore = ARGV[6]
local ckWildcardName = ARGV[7]
-- Fast-path args (ARGV 8-12)
local messageKeyValue = ARGV[8]
local defaultEnvConcurrencyLimit = ARGV[9]
local defaultEnvConcurrencyBurstFactor = ARGV[10]
local currentTime = ARGV[11]
local enableFastPath = ARGV[12]

-- Fast path: check if we can skip the queue and go directly to worker queue
if enableFastPath == '1' then
  local available = redis.call('ZRANGEBYSCORE', queueKey, '-inf', currentTime, 'LIMIT', 0, 1)
  if #available == 0 then
    local envCurrent = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
    local envLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
    local envBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
    local envLimitWithBurst = math.floor(envLimit * envBurstFactor)

    if envCurrent < envLimitWithBurst then
      local queueCurrent = tonumber(redis.call('SCARD', queueCurrentConcurrencyKey) or '0')
      local queueLimit = math.min(
        tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
        envLimit
      )

      if queueCurrent < queueLimit then
        redis.call('SET', messageKey, messageData)
        redis.call('SADD', queueCurrentConcurrencyKey, messageId)
        redis.call('SADD', envCurrentConcurrencyKey, messageId)
        redis.call('RPUSH', workerQueueKey, messageKeyValue)
        -- Skip TTL sorted set: the expireRun worker job handles TTL expiry independently
        return 1
      end
    end
  end
end

-- Slow path: normal enqueue
-- Write the message to the message key
redis.call('SET', messageKey, messageData)

-- Add the message to the CK-specific queue
redis.call('ZADD', queueKey, messageScore, messageId)

-- Add the message to the env queue
redis.call('ZADD', envQueueKey, messageScore, messageId)

-- Add to TTL sorted set
redis.call('ZADD', ttlQueueKey, ttlScore, ttlMember)

-- Rebalance CK index
local earliest = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')
if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], queueName)
end

-- Rebalance master queue with ck:* member
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx > 0 then
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, queueName)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

return 0
      `,
    });

    // Tracked variants of the CK enqueue scripts. Identical to the originals except
    // they maintain a per-base-queue `lengthCounter` so the trigger-time queue-length
    // cap and the dashboard's "Queued" column can see the true aggregate across all
    // CK variants. The counter is lazy-initialized inside the script the first time
    // a CK enqueue touches a base queue by summing ZCARDs across the existing
    // ckIndex variants + the base queue zset. The runningCounter is touched in the
    // *Tracked variants of dequeueMessageFromKey and the ack/nack/dlq/release/clear
    // scripts.
    this.redis.defineCommand("enqueueMessageCkTracked", {
      numberOfKeys: 15,
      lua: `
local masterQueueKey = KEYS[1]
local queueKey = KEYS[2]
local messageKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local ckIndexKey = KEYS[9]
-- Fast-path keys (KEYS 10-13)
local workerQueueKey = KEYS[10]
local queueConcurrencyLimitKey = KEYS[11]
local envConcurrencyLimitKey = KEYS[12]
local envConcurrencyLimitBurstFactorKey = KEYS[13]
-- Counter keys (KEYS 14-15)
local lengthCounterKey = KEYS[14]
local baseQueueKey = KEYS[15]

local queueName = ARGV[1]
local messageId = ARGV[2]
local messageData = ARGV[3]
local messageScore = ARGV[4]
local ckWildcardName = ARGV[5]
-- Fast-path args (ARGV 6-10)
local messageKeyValue = ARGV[6]
local defaultEnvConcurrencyLimit = ARGV[7]
local defaultEnvConcurrencyBurstFactor = ARGV[8]
local currentTime = ARGV[9]
local enableFastPath = ARGV[10]
-- keyPrefix for prepending to variant names stored as values in ckIndex
local keyPrefix = ARGV[11]
-- TTL (seconds) applied to counter lazy-init SETs
local counterTtl = ARGV[12]

-- Fast path: check if we can skip the queue and go directly to worker queue
if enableFastPath == '1' then
  local available = redis.call('ZRANGEBYSCORE', queueKey, '-inf', currentTime, 'LIMIT', 0, 1)
  if #available == 0 then
    local envCurrent = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
    local envLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
    local envBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
    local envLimitWithBurst = math.floor(envLimit * envBurstFactor)

    if envCurrent < envLimitWithBurst then
      local queueCurrent = tonumber(redis.call('SCARD', queueCurrentConcurrencyKey) or '0')
      local queueLimit = math.min(
        tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
        envLimit
      )

      if queueCurrent < queueLimit then
        redis.call('SET', messageKey, messageData)
        redis.call('SADD', queueCurrentConcurrencyKey, messageId)
        redis.call('SADD', envCurrentConcurrencyKey, messageId)
        redis.call('RPUSH', workerQueueKey, messageKeyValue)
        -- Fast-path skips the CK variant zset entirely; lengthCounter is unchanged.
        -- runningCounter is bumped later by dequeueMessageFromKeyTracked when the
        -- worker pulls the message from the worker queue.
        return 1
      end
    end
  end
end

-- Slow path: normal enqueue
redis.call('SET', messageKey, messageData)

-- Lazy-init lengthCounter from existing ckIndex variants (once per base queue per 24h).
-- The 24h TTL means the counter periodically re-anchors to truth, bounding any drift
-- that accumulated during rolling-deploy overlap windows.
-- Run BEFORE the ZADD so we capture pre-state; the subsequent INCR accounts for the new message.
-- The counter tracks ONLY CK-variant messages — the read path adds ZCARD(base) separately,
-- so the base zset is intentionally excluded here.
if redis.call('EXISTS', lengthCounterKey) == 0 then
  local total = 0
  local variants = redis.call('ZRANGE', ckIndexKey, 0, -1)
  for _, v in ipairs(variants) do
    total = total + tonumber(redis.call('ZCARD', keyPrefix .. v) or '0')
  end
  redis.call('SET', lengthCounterKey, total, 'EX', counterTtl)
end

-- INCR is gated on ZADD returning 1 (new entry). A duplicate enqueue (same messageId
-- already in the variant zset) returns 0 and must not bump the counter.
local added = redis.call('ZADD', queueKey, messageScore, messageId)
redis.call('ZADD', envQueueKey, messageScore, messageId)
if added == 1 then
  redis.call('INCR', lengthCounterKey)
end

-- Rebalance CK index
local earliest = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')
if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], queueName)
end

-- Rebalance master queue with ck:* member
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx > 0 then
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, queueName)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

return 0
      `,
    });

    this.redis.defineCommand("enqueueMessageWithTtlCkTracked", {
      numberOfKeys: 16,
      lua: `
local masterQueueKey = KEYS[1]
local queueKey = KEYS[2]
local messageKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local ttlQueueKey = KEYS[9]
local ckIndexKey = KEYS[10]
-- Fast-path keys (KEYS 11-14)
local workerQueueKey = KEYS[11]
local queueConcurrencyLimitKey = KEYS[12]
local envConcurrencyLimitKey = KEYS[13]
local envConcurrencyLimitBurstFactorKey = KEYS[14]
-- Counter keys (KEYS 15-16)
local lengthCounterKey = KEYS[15]
local baseQueueKey = KEYS[16]

local queueName = ARGV[1]
local messageId = ARGV[2]
local messageData = ARGV[3]
local messageScore = ARGV[4]
local ttlMember = ARGV[5]
local ttlScore = ARGV[6]
local ckWildcardName = ARGV[7]
-- Fast-path args (ARGV 8-12)
local messageKeyValue = ARGV[8]
local defaultEnvConcurrencyLimit = ARGV[9]
local defaultEnvConcurrencyBurstFactor = ARGV[10]
local currentTime = ARGV[11]
local enableFastPath = ARGV[12]
-- keyPrefix for prepending to variant names stored as values in ckIndex
local keyPrefix = ARGV[13]
-- TTL (seconds) applied to counter lazy-init SETs
local counterTtl = ARGV[14]

-- Fast path: check if we can skip the queue and go directly to worker queue
if enableFastPath == '1' then
  local available = redis.call('ZRANGEBYSCORE', queueKey, '-inf', currentTime, 'LIMIT', 0, 1)
  if #available == 0 then
    local envCurrent = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
    local envLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
    local envBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
    local envLimitWithBurst = math.floor(envLimit * envBurstFactor)

    if envCurrent < envLimitWithBurst then
      local queueCurrent = tonumber(redis.call('SCARD', queueCurrentConcurrencyKey) or '0')
      local queueLimit = math.min(
        tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'),
        envLimit
      )

      if queueCurrent < queueLimit then
        redis.call('SET', messageKey, messageData)
        redis.call('SADD', queueCurrentConcurrencyKey, messageId)
        redis.call('SADD', envCurrentConcurrencyKey, messageId)
        redis.call('RPUSH', workerQueueKey, messageKeyValue)
        return 1
      end
    end
  end
end

-- Slow path: normal enqueue
redis.call('SET', messageKey, messageData)

-- Lazy-init lengthCounter from existing ckIndex variants (once per base queue per 24h).
-- See enqueueMessageCkTracked for the TTL rationale.
if redis.call('EXISTS', lengthCounterKey) == 0 then
  local total = 0
  local variants = redis.call('ZRANGE', ckIndexKey, 0, -1)
  for _, v in ipairs(variants) do
    total = total + tonumber(redis.call('ZCARD', keyPrefix .. v) or '0')
  end
  redis.call('SET', lengthCounterKey, total, 'EX', counterTtl)
end

-- INCR is gated on ZADD returning 1 (new entry).
local added = redis.call('ZADD', queueKey, messageScore, messageId)
redis.call('ZADD', envQueueKey, messageScore, messageId)
redis.call('ZADD', ttlQueueKey, ttlScore, ttlMember)
if added == 1 then
  redis.call('INCR', lengthCounterKey)
end

-- Rebalance CK index
local earliest = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')
if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], queueName)
end

-- Rebalance master queue with ck:* member
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx > 0 then
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, queueName)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

return 0
      `,
    });

    // Expire TTL runs - atomically removes from TTL set, acknowledges from normal queue, and enqueues to TTL worker
    this.redis.defineCommand("expireTtlRuns", {
      numberOfKeys: 1,
      lua: `
local ttlQueueKey = KEYS[1]
local keyPrefix = ARGV[1]
local currentTime = tonumber(ARGV[2])
local batchSize = tonumber(ARGV[3])
local shardCount = tonumber(ARGV[4])
local workerQueueKey = ARGV[5]
local workerItemsKey = ARGV[6]
local visibilityTimeoutMs = tonumber(ARGV[7])

-- Get expired runs from TTL sorted set (score <= currentTime)
local expiredMembers = redis.call('ZRANGEBYSCORE', ttlQueueKey, '-inf', currentTime, 'LIMIT', 0, batchSize)

if #expiredMembers == 0 then
  return {}
end

local time = redis.call('TIME')
local nowMs = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)

local results = {}

for i, member in ipairs(expiredMembers) do
  -- Parse member format: "queueKey|runId|orgId"
  local pipePos1 = string.find(member, "|", 1, true)
  if pipePos1 then
    local pipePos2 = string.find(member, "|", pipePos1 + 1, true)
    if pipePos2 then
      local rawQueueKey = string.sub(member, 1, pipePos1 - 1)
      local runId = string.sub(member, pipePos1 + 1, pipePos2 - 1)
      local orgId = string.sub(member, pipePos2 + 1)

      -- Prefix the queue key so it matches the actual Redis keys
      local queueKey = keyPrefix .. rawQueueKey

      -- Remove from TTL set
      redis.call('ZREM', ttlQueueKey, member)

      -- Construct keys for acknowledging the run from normal queue
      -- Extract org from rawQueueKey: {org:orgId}:proj:...
      local orgKeyStart = string.find(rawQueueKey, "{org:", 1, true)
      local orgKeyEnd = string.find(rawQueueKey, "}", orgKeyStart, true)
      local orgFromQueue = string.sub(rawQueueKey, orgKeyStart + 5, orgKeyEnd - 1)

      local messageKey = keyPrefix .. "{org:" .. orgFromQueue .. "}:message:" .. runId

      -- Delete message key
      redis.call('DEL', messageKey)

      -- Remove from queue sorted set
      redis.call('ZREM', queueKey, runId)

      -- Remove from env queue (derive from rawQueueKey)
      -- rawQueueKey format: {org:X}:proj:Y:env:Z:queue:Q[:ck:C]
      local envMatch = string.match(rawQueueKey, ":env:([^:]+)")
      if envMatch then
        local envQueueKey = keyPrefix .. "{org:" .. orgFromQueue .. "}:env:" .. envMatch
        redis.call('ZREM', envQueueKey, runId)
      end

      -- Remove from concurrency sets
      local concurrencyKey = queueKey .. ":currentConcurrency"
      local dequeuedKey = queueKey .. ":currentDequeued"
      redis.call('SREM', concurrencyKey, runId)
      redis.call('SREM', dequeuedKey, runId)

      -- Env concurrency (derive from rawQueueKey; must match RunQueueKeyProducer: org + proj + env)
      -- rawQueueKey format: {org:X}:proj:Y:env:Z:queue:Q[:ck:C]
      local projMatch = string.match(rawQueueKey, ":proj:([^:]+):env:")
      local envConcurrencyKey = keyPrefix .. "{org:" .. orgFromQueue .. "}:proj:" .. (projMatch or "") .. ":env:" .. (envMatch or "") .. ":currentConcurrency"
      local envDequeuedKey = keyPrefix .. "{org:" .. orgFromQueue .. "}:proj:" .. (projMatch or "") .. ":env:" .. (envMatch or "") .. ":currentDequeued"
      redis.call('SREM', envConcurrencyKey, runId)
      redis.call('SREM', envDequeuedKey, runId)

      -- Rebalance CK index if this is a CK queue
      local ckMatch = string.match(rawQueueKey, "(.-):ck:")
      if ckMatch then
        local ckIndexKey = keyPrefix .. ckMatch .. ":ckIndex"
        local earliest = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')
        if #earliest == 0 then
          redis.call('ZREM', ckIndexKey, rawQueueKey)
        else
          redis.call('ZADD', ckIndexKey, earliest[2], rawQueueKey)
        end
      end

      -- Enqueue to TTL worker (runId is natural dedup key)
      local serializedItem = cjson.encode({
        job = "expireTtlRun",
        item = { runId = runId, orgId = orgId, queueKey = rawQueueKey },
        visibilityTimeoutMs = visibilityTimeoutMs,
        attempt = 0
      })
      redis.call('ZADD', workerQueueKey, nowMs, runId)
      redis.call('HSET', workerItemsKey, runId, serializedItem)

      -- Add to results
      table.insert(results, member)
    end
  end
end

return results
      `,
    });

    // Tracked variant: same as expireTtlRuns, with floored DECRs of the
    // per-base-queue lengthCounter (for every successful ZREM from a CK variant)
    // and runningCounter (when SREM from currentDequeued actually removed something).
    this.redis.defineCommand("expireTtlRunsTracked", {
      numberOfKeys: 1,
      lua: `
local ttlQueueKey = KEYS[1]
local keyPrefix = ARGV[1]
local currentTime = tonumber(ARGV[2])
local batchSize = tonumber(ARGV[3])
local shardCount = tonumber(ARGV[4])
local workerQueueKey = ARGV[5]
local workerItemsKey = ARGV[6]
local visibilityTimeoutMs = tonumber(ARGV[7])

local function decrFloored(key)
  if tonumber(redis.call('GET', key) or '0') > 0 then
    redis.call('DECR', key)
  end
end

local expiredMembers = redis.call('ZRANGEBYSCORE', ttlQueueKey, '-inf', currentTime, 'LIMIT', 0, batchSize)

if #expiredMembers == 0 then
  return {}
end

local time = redis.call('TIME')
local nowMs = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)

local results = {}

for i, member in ipairs(expiredMembers) do
  local pipePos1 = string.find(member, "|", 1, true)
  if pipePos1 then
    local pipePos2 = string.find(member, "|", pipePos1 + 1, true)
    if pipePos2 then
      local rawQueueKey = string.sub(member, 1, pipePos1 - 1)
      local runId = string.sub(member, pipePos1 + 1, pipePos2 - 1)
      local orgId = string.sub(member, pipePos2 + 1)

      local queueKey = keyPrefix .. rawQueueKey

      redis.call('ZREM', ttlQueueKey, member)

      local orgKeyStart = string.find(rawQueueKey, "{org:", 1, true)
      local orgKeyEnd = string.find(rawQueueKey, "}", orgKeyStart, true)
      local orgFromQueue = string.sub(rawQueueKey, orgKeyStart + 5, orgKeyEnd - 1)

      local messageKey = keyPrefix .. "{org:" .. orgFromQueue .. "}:message:" .. runId

      redis.call('DEL', messageKey)

      -- ZREM from queue; if successful AND this is a CK variant, DECR lengthCounter.
      local removedFromZset = redis.call('ZREM', queueKey, runId)

      local envMatch = string.match(rawQueueKey, ":env:([^:]+)")
      if envMatch then
        local envQueueKey = keyPrefix .. "{org:" .. orgFromQueue .. "}:env:" .. envMatch
        redis.call('ZREM', envQueueKey, runId)
      end

      local concurrencyKey = queueKey .. ":currentConcurrency"
      local dequeuedKey = queueKey .. ":currentDequeued"
      redis.call('SREM', concurrencyKey, runId)
      local removedFromDequeued = redis.call('SREM', dequeuedKey, runId)

      local projMatch = string.match(rawQueueKey, ":proj:([^:]+):env:")
      local envConcurrencyKey = keyPrefix .. "{org:" .. orgFromQueue .. "}:proj:" .. (projMatch or "") .. ":env:" .. (envMatch or "") .. ":currentConcurrency"
      local envDequeuedKey = keyPrefix .. "{org:" .. orgFromQueue .. "}:proj:" .. (projMatch or "") .. ":env:" .. (envMatch or "") .. ":currentDequeued"
      redis.call('SREM', envConcurrencyKey, runId)
      redis.call('SREM', envDequeuedKey, runId)

      -- Rebalance CK index AND update counters if this is a CK queue
      local ckMatch = string.match(rawQueueKey, "(.-):ck:")
      if ckMatch then
        local lengthCounterKey = keyPrefix .. ckMatch .. ":lengthCounter"
        local runningCounterKey = keyPrefix .. ckMatch .. ":runningCounter"
        if removedFromZset == 1 then
          decrFloored(lengthCounterKey)
        end
        if removedFromDequeued == 1 then
          decrFloored(runningCounterKey)
        end

        local ckIndexKey = keyPrefix .. ckMatch .. ":ckIndex"
        local earliest = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')
        if #earliest == 0 then
          redis.call('ZREM', ckIndexKey, rawQueueKey)
        else
          redis.call('ZADD', ckIndexKey, earliest[2], rawQueueKey)
        end
      end

      local serializedItem = cjson.encode({
        job = "expireTtlRun",
        item = { runId = runId, orgId = orgId, queueKey = rawQueueKey },
        visibilityTimeoutMs = visibilityTimeoutMs,
        attempt = 0
      })
      redis.call('ZADD', workerQueueKey, nowMs, runId)
      redis.call('HSET', workerItemsKey, runId, serializedItem)

      table.insert(results, member)
    end
  end
end

return results
      `,
    });

    this.redis.defineCommand("dequeueMessagesFromQueue", {
      numberOfKeys: 10,
      lua: `
local queueKey = KEYS[1]
local queueConcurrencyLimitKey = KEYS[2]
local envConcurrencyLimitKey = KEYS[3]
local envConcurrencyLimitBurstFactorKey = KEYS[4]
local queueCurrentConcurrencyKey = KEYS[5]
local envCurrentConcurrencyKey = KEYS[6]
local messageKeyPrefix = KEYS[7]
local envQueueKey = KEYS[8]
local masterQueueKey = KEYS[9]
local ttlQueueKey = KEYS[10]  -- Optional: TTL sorted set key (empty string if not used)

local queueName = ARGV[1]
local currentTime = tonumber(ARGV[2])
local defaultEnvConcurrencyLimit = ARGV[3]
local defaultEnvConcurrencyBurstFactor = ARGV[4]
local keyPrefix = ARGV[5]
local maxCount = tonumber(ARGV[6] or '1')

-- Check current env concurrency against the limit
local envCurrentConcurrency = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
local envConcurrencyLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
local envConcurrencyLimitBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
local envConcurrencyLimitWithBurstFactor = math.floor(envConcurrencyLimit * envConcurrencyLimitBurstFactor)

if envCurrentConcurrency >= envConcurrencyLimitWithBurstFactor then
    return nil
end

-- Check current queue concurrency against the limit
local queueCurrentConcurrency = tonumber(redis.call('SCARD', queueCurrentConcurrencyKey) or '0')
local queueConcurrencyLimit = math.min(tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'), envConcurrencyLimit)
local totalQueueConcurrencyLimit = queueConcurrencyLimit

-- Check condition only if concurrencyLimit exists
if queueCurrentConcurrency >= totalQueueConcurrencyLimit then
    return nil
end

-- Calculate how many messages we can actually dequeue based on concurrency limits
local envAvailableCapacity = envConcurrencyLimitWithBurstFactor - envCurrentConcurrency
local queueAvailableCapacity = totalQueueConcurrencyLimit - queueCurrentConcurrency
local actualMaxCount = math.min(maxCount, envAvailableCapacity, queueAvailableCapacity)

if actualMaxCount <= 0 then
    return nil
end

-- Attempt to dequeue messages up to actualMaxCount
local messages = redis.call('ZRANGEBYSCORE', queueKey, '-inf', currentTime, 'WITHSCORES', 'LIMIT', 0, actualMaxCount)

if #messages == 0 then
    return nil
end

local results = {}
local dequeuedCount = 0

-- Process messages in pairs (messageId, score)
for i = 1, #messages, 2 do
    local messageId = messages[i]
    local messageScore = tonumber(messages[i + 1])

    -- Get the message payload
    local messageKey = messageKeyPrefix .. messageId
    local messagePayload = redis.call('GET', messageKey)

    if messagePayload then
        -- Parse the message to check for TTL expiration
        local messageData = cjson.decode(messagePayload)
        local ttlExpiresAt = messageData and messageData.ttlExpiresAt

        -- Check if TTL has expired
        if ttlExpiresAt and ttlExpiresAt <= currentTime then
            -- TTL expired - remove from dequeue queues so it won't be retried,
            -- but leave messageKey and ttlQueueKey intact for the TTL consumer
            -- to discover and properly expire the run.
            redis.call('ZREM', queueKey, messageId)
            redis.call('ZREM', envQueueKey, messageId)
        else
            -- Not expired - process normally
            redis.call('ZREM', queueKey, messageId)
            redis.call('ZREM', envQueueKey, messageId)
            redis.call('SADD', queueCurrentConcurrencyKey, messageId)
            redis.call('SADD', envCurrentConcurrencyKey, messageId)

            -- Remove from TTL set if provided (run is being executed, not expired)
            if ttlQueueKey and ttlQueueKey ~= '' and ttlExpiresAt then
                local ttlMember = queueName .. '|' .. messageId .. '|' .. (messageData.orgId or '')
                redis.call('ZREM', ttlQueueKey, ttlMember)
            end

            -- Add to results
            table.insert(results, messageId)
            table.insert(results, messageScore)
            table.insert(results, messagePayload)

            dequeuedCount = dequeuedCount + 1
        end
    else
        -- Stale entry: message key was already deleted (e.g. acknowledged),
        -- but the sorted set member was not cleaned up. Remove it so it
        -- doesn't block newer messages from being dequeued.
        redis.call('ZREM', queueKey, messageId)
        redis.call('ZREM', envQueueKey, messageId)
    end
end

-- Rebalance the parent queues only once after all dequeues
local earliestMessage = redis.call('ZRANGE', queueKey, 0, 0, 'WITHSCORES')

if #earliestMessage == 0 then
  redis.call('ZREM', masterQueueKey, queueName)
else
  redis.call('ZADD', masterQueueKey, earliestMessage[2], queueName)
end

-- Return results as a flat array: [messageId1, messageScore1, messagePayload1, messageId2, messageScore2, messagePayload2, ...]
return results
      `,
    });

    // CK-aware dequeue: iterates CK sub-queues from CK index
    this.redis.defineCommand("dequeueMessagesFromCkQueue", {
      numberOfKeys: 9,
      lua: `
local ckIndexKey = KEYS[1]
local queueConcurrencyLimitKey = KEYS[2]
local envConcurrencyLimitKey = KEYS[3]
local envConcurrencyLimitBurstFactorKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local messageKeyPrefix = KEYS[6]
local envQueueKey = KEYS[7]
local masterQueueKey = KEYS[8]
local ttlQueueKey = KEYS[9]

local ckWildcardName = ARGV[1]
local currentTime = tonumber(ARGV[2])
local defaultEnvConcurrencyLimit = ARGV[3]
local defaultEnvConcurrencyBurstFactor = ARGV[4]
local keyPrefix = ARGV[5]
local maxCount = tonumber(ARGV[6] or '1')

-- Check env concurrency
local envCurrentConcurrency = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
local envConcurrencyLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
local envConcurrencyLimitBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
local envConcurrencyLimitWithBurstFactor = math.floor(envConcurrencyLimit * envConcurrencyLimitBurstFactor)

if envCurrentConcurrency >= envConcurrencyLimitWithBurstFactor then
  return nil
end

-- Get base queue concurrency limit
local queueConcurrencyLimit = math.min(tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'), envConcurrencyLimit)

-- Calculate env available capacity
local envAvailableCapacity = envConcurrencyLimitWithBurstFactor - envCurrentConcurrency
local actualMaxCount = math.min(maxCount, envAvailableCapacity)

if actualMaxCount <= 0 then
  return nil
end

-- Get CK sub-queues from CK index with scores <= currentTime
local ckQueues = redis.call('ZRANGEBYSCORE', ckIndexKey, '-inf', tostring(currentTime), 'LIMIT', 0, actualMaxCount * 3)

if #ckQueues == 0 then
  -- Rebalance master queue in case CK index has future-scored entries
  local anyIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
  if #anyIdx == 0 then
    redis.call('ZREM', masterQueueKey, ckWildcardName)
  else
    redis.call('ZADD', masterQueueKey, anyIdx[2], ckWildcardName)
  end
  return nil
end

local results = {}
local dequeuedCount = 0

for _, ckQueueName in ipairs(ckQueues) do
  if dequeuedCount >= actualMaxCount then
    break
  end

  local fullQueueKey = keyPrefix .. ckQueueName

  -- Check CK-specific concurrency
  local ckConcurrencyKey = fullQueueKey .. ':currentConcurrency'
  local ckCurrentConcurrency = tonumber(redis.call('SCARD', ckConcurrencyKey) or '0')

  if ckCurrentConcurrency < queueConcurrencyLimit then
    -- Try to dequeue one message from this CK sub-queue
    local messages = redis.call('ZRANGEBYSCORE', fullQueueKey, '-inf', tostring(currentTime), 'WITHSCORES', 'LIMIT', 0, 1)

    if #messages >= 2 then
      local messageId = messages[1]
      local messageScore = messages[2]

      local messageKey = messageKeyPrefix .. messageId
      local messagePayload = redis.call('GET', messageKey)

      if messagePayload then
        local messageData = cjson.decode(messagePayload)
        local ttlExpiresAt = messageData and messageData.ttlExpiresAt

        if ttlExpiresAt and ttlExpiresAt <= currentTime then
          -- TTL expired - remove from queues
          redis.call('ZREM', fullQueueKey, messageId)
          redis.call('ZREM', envQueueKey, messageId)
        else
          -- Dequeue normally
          redis.call('ZREM', fullQueueKey, messageId)
          redis.call('ZREM', envQueueKey, messageId)
          redis.call('SADD', ckConcurrencyKey, messageId)
          redis.call('SADD', envCurrentConcurrencyKey, messageId)

          -- Remove from TTL set if applicable
          if ttlQueueKey and ttlQueueKey ~= '' and ttlExpiresAt then
            local ttlMember = ckQueueName .. '|' .. messageId .. '|' .. (messageData.orgId or '')
            redis.call('ZREM', ttlQueueKey, ttlMember)
          end

          table.insert(results, messageId)
          table.insert(results, messageScore)
          table.insert(results, messagePayload)

          dequeuedCount = dequeuedCount + 1
        end
      else
        -- Stale entry
        redis.call('ZREM', fullQueueKey, messageId)
        redis.call('ZREM', envQueueKey, messageId)
      end

      -- Rebalance CK index for this sub-queue
      local earliest = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
      if #earliest == 0 then
        redis.call('ZREM', ckIndexKey, ckQueueName)
      else
        redis.call('ZADD', ckIndexKey, earliest[2], ckQueueName)
      end
    else
      -- No messages available in score range, update CK index
      local any = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
      if #any == 0 then
        redis.call('ZREM', ckIndexKey, ckQueueName)
      else
        redis.call('ZADD', ckIndexKey, any[2], ckQueueName)
      end
    end
  end
end

-- Rebalance master queue
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx == 0 then
  redis.call('ZREM', masterQueueKey, ckWildcardName)
else
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

return results
      `,
    });

    // Tracked variant: same as dequeueMessagesFromCkQueue plus DECR of the
    // per-base-queue lengthCounter for every message removed from a CK variant
    // (normal dequeue, TTL-expired, or stale-orphan path — all of which were
    // counted at enqueue time).
    this.redis.defineCommand("dequeueMessagesFromCkQueueTracked", {
      numberOfKeys: 10,
      lua: `
local ckIndexKey = KEYS[1]
local queueConcurrencyLimitKey = KEYS[2]
local envConcurrencyLimitKey = KEYS[3]
local envConcurrencyLimitBurstFactorKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local messageKeyPrefix = KEYS[6]
local envQueueKey = KEYS[7]
local masterQueueKey = KEYS[8]
local ttlQueueKey = KEYS[9]
local lengthCounterKey = KEYS[10]

local ckWildcardName = ARGV[1]
local currentTime = tonumber(ARGV[2])
local defaultEnvConcurrencyLimit = ARGV[3]
local defaultEnvConcurrencyBurstFactor = ARGV[4]
local keyPrefix = ARGV[5]
local maxCount = tonumber(ARGV[6] or '1')

local function decrLengthCounter()
  if tonumber(redis.call('GET', lengthCounterKey) or '0') > 0 then
    redis.call('DECR', lengthCounterKey)
  end
end

-- Check env concurrency
local envCurrentConcurrency = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
local envConcurrencyLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
local envConcurrencyLimitBurstFactor = tonumber(redis.call('GET', envConcurrencyLimitBurstFactorKey) or defaultEnvConcurrencyBurstFactor)
local envConcurrencyLimitWithBurstFactor = math.floor(envConcurrencyLimit * envConcurrencyLimitBurstFactor)

if envCurrentConcurrency >= envConcurrencyLimitWithBurstFactor then
  return nil
end

local queueConcurrencyLimit = math.min(tonumber(redis.call('GET', queueConcurrencyLimitKey) or '1000000'), envConcurrencyLimit)

local envAvailableCapacity = envConcurrencyLimitWithBurstFactor - envCurrentConcurrency
local actualMaxCount = math.min(maxCount, envAvailableCapacity)

if actualMaxCount <= 0 then
  return nil
end

local ckQueues = redis.call('ZRANGEBYSCORE', ckIndexKey, '-inf', tostring(currentTime), 'LIMIT', 0, actualMaxCount * 3)

if #ckQueues == 0 then
  local anyIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
  if #anyIdx == 0 then
    redis.call('ZREM', masterQueueKey, ckWildcardName)
  else
    redis.call('ZADD', masterQueueKey, anyIdx[2], ckWildcardName)
  end
  return nil
end

local results = {}
local dequeuedCount = 0

for _, ckQueueName in ipairs(ckQueues) do
  if dequeuedCount >= actualMaxCount then
    break
  end

  local fullQueueKey = keyPrefix .. ckQueueName

  local ckConcurrencyKey = fullQueueKey .. ':currentConcurrency'
  local ckCurrentConcurrency = tonumber(redis.call('SCARD', ckConcurrencyKey) or '0')

  if ckCurrentConcurrency < queueConcurrencyLimit then
    local messages = redis.call('ZRANGEBYSCORE', fullQueueKey, '-inf', tostring(currentTime), 'WITHSCORES', 'LIMIT', 0, 1)

    if #messages >= 2 then
      local messageId = messages[1]
      local messageScore = messages[2]

      local messageKey = messageKeyPrefix .. messageId
      local messagePayload = redis.call('GET', messageKey)

      if messagePayload then
        local messageData = cjson.decode(messagePayload)
        local ttlExpiresAt = messageData and messageData.ttlExpiresAt

        if ttlExpiresAt and ttlExpiresAt <= currentTime then
          redis.call('ZREM', fullQueueKey, messageId)
          redis.call('ZREM', envQueueKey, messageId)
          decrLengthCounter()
        else
          redis.call('ZREM', fullQueueKey, messageId)
          redis.call('ZREM', envQueueKey, messageId)
          decrLengthCounter()
          redis.call('SADD', ckConcurrencyKey, messageId)
          redis.call('SADD', envCurrentConcurrencyKey, messageId)

          if ttlQueueKey and ttlQueueKey ~= '' and ttlExpiresAt then
            local ttlMember = ckQueueName .. '|' .. messageId .. '|' .. (messageData.orgId or '')
            redis.call('ZREM', ttlQueueKey, ttlMember)
          end

          table.insert(results, messageId)
          table.insert(results, messageScore)
          table.insert(results, messagePayload)

          dequeuedCount = dequeuedCount + 1
        end
      else
        redis.call('ZREM', fullQueueKey, messageId)
        redis.call('ZREM', envQueueKey, messageId)
        decrLengthCounter()
      end

      local earliest = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
      if #earliest == 0 then
        redis.call('ZREM', ckIndexKey, ckQueueName)
      else
        redis.call('ZADD', ckIndexKey, earliest[2], ckQueueName)
      end
    else
      local any = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
      if #any == 0 then
        redis.call('ZREM', ckIndexKey, ckQueueName)
      else
        redis.call('ZADD', ckIndexKey, any[2], ckQueueName)
      end
    end
  end
end

local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx == 0 then
  redis.call('ZREM', masterQueueKey, ckWildcardName)
else
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

return results
      `,
    });

    this.redis.defineCommand("dequeueMessageFromWorkerQueueNonBlocking", {
      numberOfKeys: 1,
      lua: `
local workerQueueKey = KEYS[1]

-- lpop the first message from the worker queue
local messageId = redis.call('LPOP', workerQueueKey)

-- if there is no messageId, return nil
if not messageId then
    return nil
end

-- get the length of the worker queue
local queueLength = tonumber(redis.call('LLEN', workerQueueKey) or '0')

return {messageId, queueLength} -- Return message details
      `,
    });

    this.redis.defineCommand("dequeueMessageFromKey", {
      numberOfKeys: 1,
      lua: `
-- Keys:
local messageKey = KEYS[1]

-- Args:
local keyPrefix = ARGV[1]

-- Get the raw messages string
local message = redis.call('GET', messageKey)

if not message then
  return nil
end

-- Parse the message
local messageData = cjson.decode(message)

-- Construct the "current dequeued" keys for the queue and the environment
local queueCurrentDequeuedKey = keyPrefix .. messageData.queue .. ':currentDequeued'
local envCurrentDequeuedKey = keyPrefix ..  string.match(messageData.queue, "(.+):queue:") .. ":currentDequeued"

-- Update the concurrency keys
redis.call('SADD', queueCurrentDequeuedKey, messageData.runId)
redis.call('SADD', envCurrentDequeuedKey, messageData.runId)

-- Return the message data
return message
      `,
    });

    // Tracked variant: same as dequeueMessageFromKey, plus runningCounter INCR
    // when the message lives on a CK variant queue. The runningCounter is
    // lazy-initialized from the existing ckIndex variants' currentDequeued sets.
    // Note: ckIndex only tracks variants that currently have queued messages,
    // so the init misses any variant that's "running-only" (all messages
    // dequeued, none queued). Those are reabsorbed naturally as messages ack
    // (floored DECR) and new dequeues land.
    this.redis.defineCommand("dequeueMessageFromKeyTracked", {
      numberOfKeys: 1,
      lua: `
local messageKey = KEYS[1]
local keyPrefix = ARGV[1]
-- TTL (seconds) applied to counter lazy-init SETs
local counterTtl = ARGV[2]

local message = redis.call('GET', messageKey)
if not message then
  return nil
end

local messageData = cjson.decode(message)
local queue = messageData.queue

local queueCurrentDequeuedKey = keyPrefix .. queue .. ':currentDequeued'
local envCurrentDequeuedKey = keyPrefix .. string.match(queue, "(.+):queue:") .. ':currentDequeued'

-- SADD first so we know if this dequeue is new (return 1) or a duplicate (return 0).
-- INCR runningCounter is gated on the new-membership result so re-dequeues don't inflate.
-- The alternative (lazy-init before SADD) was rejected because we need the SADD return
-- value to gate the INCR, and the lazy-init seed under SADD-first already reflects the
-- new membership. We compensate with the total-1 in the seed math below.
local addedDeq = redis.call('SADD', queueCurrentDequeuedKey, messageData.runId)
redis.call('SADD', envCurrentDequeuedKey, messageData.runId)

-- If CK + new addition, bump the runningCounter for the base queue (lazy init from ckIndex).
-- The counter tracks ONLY CK-variant currentDequeued — the read path adds the base
-- SCARD separately, so we exclude the base currentDequeued here. 24h TTL on init
-- bounds any drift from rolling-deploy v1/v2 overlap.
local baseQueue = string.match(queue, "(.-):ck:")
if baseQueue and addedDeq == 1 then
  local runningCounterKey = keyPrefix .. baseQueue .. ':runningCounter'
  if redis.call('EXISTS', runningCounterKey) == 0 then
    local ckIndexKey = keyPrefix .. baseQueue .. ':ckIndex'
    local total = 0
    local ownVariantSeen = false
    local variants = redis.call('ZRANGE', ckIndexKey, 0, -1)
    for _, v in ipairs(variants) do
      if v == queue then
        ownVariantSeen = true
      end
      total = total + tonumber(redis.call('SCARD', keyPrefix .. v .. ':currentDequeued') or '0')
    end
    -- Fast-path messages skip the variant zset and ckIndex entirely (see
    -- enqueueMessageCkTracked fast path). If our variant isn't in ckIndex, the loop
    -- missed its SCARD; add it manually. Either way, the SCARD we sum already
    -- reflects our just-SADD'd member, so subtract 1 before SETting (the INCR
    -- below will add it back).
    if not ownVariantSeen then
      total = total + tonumber(redis.call('SCARD', queueCurrentDequeuedKey) or '0')
    end
    redis.call('SET', runningCounterKey, math.max(0, total - 1), 'EX', counterTtl)
  end
  redis.call('INCR', runningCounterKey)
end

return message
      `,
    });

    this.redis.defineCommand("acknowledgeMessage", {
      numberOfKeys: 9,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueueKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local workerQueueKey = KEYS[9]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]
local messageKeyValue = ARGV[3]
local removeFromWorkerQueue = ARGV[4]

-- Remove the message from the message key
redis.call('DEL', messageKey)

-- Remove the message from the queue
redis.call('ZREM', messageQueueKey, messageId)
redis.call('ZREM', envQueueKey, messageId)

-- Rebalance the parent queues
local earliestMessage = redis.call('ZRANGE', messageQueueKey, 0, 0, 'WITHSCORES')
if #earliestMessage == 0 then
  redis.call('ZREM', masterQueueKey, messageQueueName)
else
  redis.call('ZADD', masterQueueKey, earliestMessage[2], messageQueueName)
end

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

-- Remove the message from the worker queue
if removeFromWorkerQueue == '1' then
  redis.call('LREM', workerQueueKey, 0, messageKeyValue)
end
`,
    });

    this.redis.defineCommand("nackMessage", {
      numberOfKeys: 8,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueueKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]
local messageData = ARGV[3]
local messageScore = tonumber(ARGV[4])

-- Update the message data
redis.call('SET', messageKey, messageData)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

-- Enqueue the message into the queue
redis.call('ZADD', messageQueueKey, messageScore, messageId)
redis.call('ZADD', envQueueKey, messageScore, messageId)

-- Rebalance the parent queues
local earliestMessage = redis.call('ZRANGE', messageQueueKey, 0, 0, 'WITHSCORES')
if #earliestMessage == 0 then
  redis.call('ZREM', masterQueueKey, messageQueueName)
else
  redis.call('ZADD', masterQueueKey, earliestMessage[2], messageQueueName)
end
`,
    });

    this.redis.defineCommand("moveToDeadLetterQueue", {
      numberOfKeys: 9,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueue = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local deadLetterQueueKey = KEYS[9]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]

-- Remove the message from the queue
redis.call('ZREM', messageQueue, messageId)
redis.call('ZREM', envQueueKey, messageId)

-- Rebalance the parent queues
local earliestMessage = redis.call('ZRANGE', messageQueue, 0, 0, 'WITHSCORES')
if #earliestMessage == 0 then
  redis.call('ZREM', masterQueueKey, messageQueueName)
else
  redis.call('ZADD', masterQueueKey, earliestMessage[2], messageQueueName)
end

-- Add the message to the dead letter queue
redis.call('ZADD', deadLetterQueueKey, tonumber(redis.call('TIME')[1]), messageId)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)
`,
    });

    // CK-aware acknowledge: rebalances CK index and master queue with :ck:* member
    this.redis.defineCommand("acknowledgeMessageCk", {
      numberOfKeys: 10,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueueKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local workerQueueKey = KEYS[9]
local ckIndexKey = KEYS[10]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]
local messageKeyValue = ARGV[3]
local removeFromWorkerQueue = ARGV[4]
local ckWildcardName = ARGV[5]

-- Remove the message from the message key
redis.call('DEL', messageKey)

-- Remove the message from the CK-specific queue
redis.call('ZREM', messageQueueKey, messageId)
redis.call('ZREM', envQueueKey, messageId)

-- Rebalance CK index
local earliestInCkQueue = redis.call('ZRANGE', messageQueueKey, 0, 0, 'WITHSCORES')
if #earliestInCkQueue == 0 then
  redis.call('ZREM', ckIndexKey, messageQueueName)
else
  redis.call('ZADD', ckIndexKey, earliestInCkQueue[2], messageQueueName)
end

-- Rebalance master queue with ck:* member
local earliestInCkIndex = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestInCkIndex == 0 then
  redis.call('ZREM', masterQueueKey, ckWildcardName)
else
  redis.call('ZADD', masterQueueKey, earliestInCkIndex[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, messageQueueName)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

-- Remove the message from the worker queue
if removeFromWorkerQueue == '1' then
  redis.call('LREM', workerQueueKey, 0, messageKeyValue)
end
`,
    });

    // CK-aware nack: rebalances CK index and master queue with :ck:* member
    this.redis.defineCommand("nackMessageCk", {
      numberOfKeys: 9,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueueKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local ckIndexKey = KEYS[9]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]
local messageData = ARGV[3]
local messageScore = tonumber(ARGV[4])
local ckWildcardName = ARGV[5]

-- Update the message data
redis.call('SET', messageKey, messageData)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

-- Enqueue the message back into the CK-specific queue
redis.call('ZADD', messageQueueKey, messageScore, messageId)
redis.call('ZADD', envQueueKey, messageScore, messageId)

-- Rebalance CK index
local earliest = redis.call('ZRANGE', messageQueueKey, 0, 0, 'WITHSCORES')
if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], messageQueueName)
end

-- Rebalance master queue with ck:* member
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx == 0 then
  redis.call('ZREM', masterQueueKey, ckWildcardName)
else
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, messageQueueName)
`,
    });

    // CK-aware move to dead letter queue
    this.redis.defineCommand("moveToDeadLetterQueueCk", {
      numberOfKeys: 10,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueue = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local deadLetterQueueKey = KEYS[9]
local ckIndexKey = KEYS[10]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]
local ckWildcardName = ARGV[3]

-- Remove the message from the CK-specific queue
redis.call('ZREM', messageQueue, messageId)
redis.call('ZREM', envQueueKey, messageId)

-- Rebalance CK index
local earliest = redis.call('ZRANGE', messageQueue, 0, 0, 'WITHSCORES')
if #earliest == 0 then
  redis.call('ZREM', ckIndexKey, messageQueueName)
else
  redis.call('ZADD', ckIndexKey, earliest[2], messageQueueName)
end

-- Rebalance master queue with ck:* member
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx == 0 then
  redis.call('ZREM', masterQueueKey, ckWildcardName)
else
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, messageQueueName)

-- Add the message to the dead letter queue
redis.call('ZADD', deadLetterQueueKey, tonumber(redis.call('TIME')[1]), messageId)

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)
`,
    });

    // Tracked variant: same as acknowledgeMessageCk, plus floored DECRs of the
    // per-base-queue lengthCounter (defensive — only fires when the ZREM actually
    // removed something) and runningCounter (when SREM currentDequeued actually
    // removed something).
    this.redis.defineCommand("acknowledgeMessageCkTracked", {
      numberOfKeys: 12,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueueKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local workerQueueKey = KEYS[9]
local ckIndexKey = KEYS[10]
local lengthCounterKey = KEYS[11]
local runningCounterKey = KEYS[12]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]
local messageKeyValue = ARGV[3]
local removeFromWorkerQueue = ARGV[4]
local ckWildcardName = ARGV[5]

local function decrFloored(key)
  if tonumber(redis.call('GET', key) or '0') > 0 then
    redis.call('DECR', key)
  end
end

-- Remove the message from the message key
redis.call('DEL', messageKey)

-- Remove the message from the CK-specific queue. The ZREM is defensive — by
-- ack time the message is normally in currentConcurrency, not the zset — but
-- if it does remove something, the counter was tracking that entry so decr.
local removedFromZset = redis.call('ZREM', messageQueueKey, messageId)
redis.call('ZREM', envQueueKey, messageId)
if removedFromZset == 1 then
  decrFloored(lengthCounterKey)
end

-- Rebalance CK index
local earliestInCkQueue = redis.call('ZRANGE', messageQueueKey, 0, 0, 'WITHSCORES')
if #earliestInCkQueue == 0 then
  redis.call('ZREM', ckIndexKey, messageQueueName)
else
  redis.call('ZADD', ckIndexKey, earliestInCkQueue[2], messageQueueName)
end

-- Rebalance master queue with ck:* member
local earliestInCkIndex = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestInCkIndex == 0 then
  redis.call('ZREM', masterQueueKey, ckWildcardName)
else
  redis.call('ZADD', masterQueueKey, earliestInCkIndex[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, messageQueueName)

-- Update the concurrency keys. DECR runningCounter only when SREM
-- currentDequeued actually removed an entry (the message was in flight).
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)
if removedFromDequeued == 1 then
  decrFloored(runningCounterKey)
end

-- Remove the message from the worker queue
if removeFromWorkerQueue == '1' then
  redis.call('LREM', workerQueueKey, 0, messageKeyValue)
end
`,
    });

    // Tracked variant: same as nackMessageCk. SREM currentDequeued may DECR
    // runningCounter (floored); ZADD back to the variant zset INCRs
    // lengthCounter only when ZADD reported a new entry.
    this.redis.defineCommand("nackMessageCkTracked", {
      numberOfKeys: 11,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueueKey = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local ckIndexKey = KEYS[9]
local lengthCounterKey = KEYS[10]
local runningCounterKey = KEYS[11]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]
local messageData = ARGV[3]
local messageScore = tonumber(ARGV[4])
local ckWildcardName = ARGV[5]
-- keyPrefix for prepending to variant names stored as values in ckIndex (lazy-init only)
local keyPrefix = ARGV[6]
-- TTL (seconds) applied to counter lazy-init SETs
local counterTtl = ARGV[7]

local function decrFloored(key)
  if tonumber(redis.call('GET', key) or '0') > 0 then
    redis.call('DECR', key)
  end
end

-- Update the message data
redis.call('SET', messageKey, messageData)

-- Update the concurrency keys. nack only DECRs runningCounter, never INCRs it,
-- so we skip the eager lazy-init here (unlike releaseConcurrencyTracked, which
-- mirrors the same DECR pattern with init). A post-TTL nack's floored DECR
-- no-ops; the next dequeueMessageFromKeyTracked reseeds from current state.
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)
if removedFromDequeued == 1 then
  decrFloored(runningCounterKey)
end

-- Lazy-init lengthCounter if missing (e.g. expired via 24h TTL). nack re-queues a
-- message, which means lengthCounter must be present before we INCR. Without this,
-- a nack after counter expiry would create the counter at 1 and stay drifted until
-- next reset.
if redis.call('EXISTS', lengthCounterKey) == 0 then
  local total = 0
  local variants = redis.call('ZRANGE', ckIndexKey, 0, -1)
  for _, v in ipairs(variants) do
    total = total + tonumber(redis.call('ZCARD', keyPrefix .. v) or '0')
  end
  redis.call('SET', lengthCounterKey, total, 'EX', counterTtl)
end

-- Enqueue the message back into the CK-specific queue. INCR lengthCounter only if
-- it's a new entry (ZADD returns 1).
local added = redis.call('ZADD', messageQueueKey, messageScore, messageId)
redis.call('ZADD', envQueueKey, messageScore, messageId)
if added == 1 then
  redis.call('INCR', lengthCounterKey)
end

-- Rebalance CK index
local earliest = redis.call('ZRANGE', messageQueueKey, 0, 0, 'WITHSCORES')
if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], messageQueueName)
end

-- Rebalance master queue with ck:* member
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx == 0 then
  redis.call('ZREM', masterQueueKey, ckWildcardName)
else
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, messageQueueName)
`,
    });

    // Tracked variant: same as moveToDeadLetterQueueCk. ZREM may DECR
    // lengthCounter (defensive); SREM currentDequeued may DECR runningCounter.
    this.redis.defineCommand("moveToDeadLetterQueueCkTracked", {
      numberOfKeys: 12,
      lua: `
-- Keys:
local masterQueueKey = KEYS[1]
local messageKey = KEYS[2]
local messageQueue = KEYS[3]
local queueCurrentConcurrencyKey = KEYS[4]
local envCurrentConcurrencyKey = KEYS[5]
local queueCurrentDequeuedKey = KEYS[6]
local envCurrentDequeuedKey = KEYS[7]
local envQueueKey = KEYS[8]
local deadLetterQueueKey = KEYS[9]
local ckIndexKey = KEYS[10]
local lengthCounterKey = KEYS[11]
local runningCounterKey = KEYS[12]

-- Args:
local messageId = ARGV[1]
local messageQueueName = ARGV[2]
local ckWildcardName = ARGV[3]

local function decrFloored(key)
  if tonumber(redis.call('GET', key) or '0') > 0 then
    redis.call('DECR', key)
  end
end

-- Remove the message from the CK-specific queue. ZREM may be a no-op if the
-- message was already moved to currentConcurrency; only decr when it actually
-- removes something.
local removedFromZset = redis.call('ZREM', messageQueue, messageId)
redis.call('ZREM', envQueueKey, messageId)
if removedFromZset == 1 then
  decrFloored(lengthCounterKey)
end

-- Rebalance CK index
local earliest = redis.call('ZRANGE', messageQueue, 0, 0, 'WITHSCORES')
if #earliest == 0 then
  redis.call('ZREM', ckIndexKey, messageQueueName)
else
  redis.call('ZADD', ckIndexKey, earliest[2], messageQueueName)
end

-- Rebalance master queue with ck:* member
local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
if #earliestIdx == 0 then
  redis.call('ZREM', masterQueueKey, ckWildcardName)
else
  redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
end

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, messageQueueName)

-- Add the message to the dead letter queue
redis.call('ZADD', deadLetterQueueKey, tonumber(redis.call('TIME')[1]), messageId)

-- Update the concurrency keys. DECR runningCounter only when SREM
-- currentDequeued actually removed an entry.
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)
if removedFromDequeued == 1 then
  decrFloored(runningCounterKey)
end
`,
    });

    this.redis.defineCommand("releaseConcurrency", {
      numberOfKeys: 4,
      lua: `
-- Keys:
local queueCurrentConcurrencyKey = KEYS[1]
local envCurrentConcurrencyKey = KEYS[2]
local queueCurrentDequeuedKey = KEYS[3]
local envCurrentDequeuedKey = KEYS[4]

-- Args:
local messageId = ARGV[1]

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)
`,
    });

    // Tracked variant: same as releaseConcurrency, with a floored DECR of the
    // base-queue runningCounter when SREM currentDequeued actually removed
    // something. Caller should only invoke this variant for CK queues — non-CK
    // queues should keep calling releaseConcurrency.
    this.redis.defineCommand("releaseConcurrencyTracked", {
      numberOfKeys: 6,
      lua: `
-- Keys:
local queueCurrentConcurrencyKey = KEYS[1]
local envCurrentConcurrencyKey = KEYS[2]
local queueCurrentDequeuedKey = KEYS[3]
local envCurrentDequeuedKey = KEYS[4]
local runningCounterKey = KEYS[5]
local ckIndexKey = KEYS[6]

-- Args:
local messageId = ARGV[1]
local keyPrefix = ARGV[2]
-- TTL (seconds) applied to counter lazy-init SETs
local counterTtl = ARGV[3]

-- Lazy-init runningCounter if missing (e.g. expired via 24h TTL). Runs BEFORE
-- the SREM so the seed captures pre-release state; the subsequent DECR accounts
-- for the message we're about to release. Without this, a release landing after
-- counter expiry would silently no-op the DECR and the next dequeue would seed
-- to post-release truth — bounded drift but inconsistent with nack/enqueue.
if redis.call('EXISTS', runningCounterKey) == 0 then
  local total = 0
  local variants = redis.call('ZRANGE', ckIndexKey, 0, -1)
  for _, v in ipairs(variants) do
    total = total + tonumber(redis.call('SCARD', keyPrefix .. v .. ':currentDequeued') or '0')
  end
  redis.call('SET', runningCounterKey, total, 'EX', counterTtl)
end

redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

if removedFromDequeued == 1 then
  if tonumber(redis.call('GET', runningCounterKey) or '0') > 0 then
    redis.call('DECR', runningCounterKey)
  end
end
`,
    });

    this.redis.defineCommand("updateEnvironmentConcurrencyLimits", {
      numberOfKeys: 2,
      lua: `
-- Keys: envConcurrencyLimitKey, envConcurrencyLimitBurstFactorKey
local envConcurrencyLimitKey = KEYS[1]
local envConcurrencyLimitBurstFactorKey = KEYS[2]

-- Args: envConcurrencyLimit
local envConcurrencyLimit = ARGV[1]
local envConcurrencyLimitBurstFactor = ARGV[2]

redis.call('SET', envConcurrencyLimitKey, envConcurrencyLimit)
redis.call('SET', envConcurrencyLimitBurstFactorKey, envConcurrencyLimitBurstFactor)
      `,
    });

    this.redis.defineCommand("markCompletedRunsForAck", {
      numberOfKeys: 1,
      lua: `
-- Keys:
local markedForAckKey = KEYS[1]

-- Args: alternating orgId, messageId pairs
local currentTime = tonumber(redis.call('TIME')[1]) * 1000

for i = 1, #ARGV, 2 do
    local orgId = ARGV[i]
    local messageId = ARGV[i + 1]
    local markedValue = orgId .. ':' .. messageId
    
    redis.call('ZADD', markedForAckKey, currentTime, markedValue)
end

return #ARGV / 2
      `,
    });

    this.redis.defineCommand("getMarkedRunsForAck", {
      numberOfKeys: 1,
      lua: `
-- Keys:
local markedForAckKey = KEYS[1]

-- Args:
local maxCount = tonumber(ARGV[1] or '10')

-- Get the oldest marked runs
local markedRuns = redis.call('ZRANGE', markedForAckKey, 0, maxCount - 1, 'WITHSCORES')

local results = {}
for i = 1, #markedRuns, 2 do
    local markedValue = markedRuns[i]
    local score = markedRuns[i + 1]
    
    -- Parse orgId:messageId
    local colonIndex = string.find(markedValue, ':')
    if colonIndex then
        local orgId = string.sub(markedValue, 1, colonIndex - 1)
        local messageId = string.sub(markedValue, colonIndex + 1)
        
        table.insert(results, orgId)
        table.insert(results, messageId)
        table.insert(results, score)
    end
end

-- Remove the processed items
if #results > 0 then
    local itemsToRemove = {}
    for i = 1, #markedRuns, 2 do
        table.insert(itemsToRemove, markedRuns[i])
    end
    redis.call('ZREM', markedForAckKey, unpack(itemsToRemove))
end

return results
      `,
    });

    this.redis.defineCommand("clearMessageFromConcurrencySets", {
      numberOfKeys: 4,
      lua: `
-- Keys:
local queueCurrentConcurrencyKey = KEYS[1]
local envCurrentConcurrencyKey = KEYS[2]
local queueCurrentDequeuedKey = KEYS[3]
local envCurrentDequeuedKey = KEYS[4]

-- Args:
local messageId = ARGV[1]

-- Update the concurrency keys
redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)
`,
    });

    // Tracked variant of clearMessageFromConcurrencySets — see releaseConcurrencyTracked
    // for the contract. Only invoke for CK queues.
    this.redis.defineCommand("clearMessageFromConcurrencySetsTracked", {
      numberOfKeys: 6,
      lua: `
-- Keys:
local queueCurrentConcurrencyKey = KEYS[1]
local envCurrentConcurrencyKey = KEYS[2]
local queueCurrentDequeuedKey = KEYS[3]
local envCurrentDequeuedKey = KEYS[4]
local runningCounterKey = KEYS[5]
local ckIndexKey = KEYS[6]

-- Args:
local messageId = ARGV[1]
local keyPrefix = ARGV[2]
-- TTL (seconds) applied to counter lazy-init SETs
local counterTtl = ARGV[3]

-- Lazy-init runningCounter if missing — see releaseConcurrencyTracked for rationale.
if redis.call('EXISTS', runningCounterKey) == 0 then
  local total = 0
  local variants = redis.call('ZRANGE', ckIndexKey, 0, -1)
  for _, v in ipairs(variants) do
    total = total + tonumber(redis.call('SCARD', keyPrefix .. v .. ':currentDequeued') or '0')
  end
  redis.call('SET', runningCounterKey, total, 'EX', counterTtl)
end

redis.call('SREM', queueCurrentConcurrencyKey, messageId)
redis.call('SREM', envCurrentConcurrencyKey, messageId)
local removedFromDequeued = redis.call('SREM', queueCurrentDequeuedKey, messageId)
redis.call('SREM', envCurrentDequeuedKey, messageId)

if removedFromDequeued == 1 then
  if tonumber(redis.call('GET', runningCounterKey) or '0') > 0 then
    redis.call('DECR', runningCounterKey)
  end
end
`,
    });
  }
}

function safeJsonParse(rawMessage: string): unknown {
  try {
    return JSON.parse(rawMessage);
  } catch (e) {
    return undefined;
  }
}

declare module "@internal/redis" {
  interface RedisCommander<Context> {
    enqueueMessage(
      //keys
      masterQueueKey: string,
      queue: string,
      messageKey: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      workerQueueKey: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      //args
      queueName: string,
      messageId: string,
      messageData: string,
      messageScore: string,
      messageKeyValue: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      currentTime: string,
      enableFastPath: string,
      callback?: Callback<number>
    ): Result<number, Context>;

    enqueueMessageWithTtl(
      //keys
      masterQueueKey: string,
      queue: string,
      messageKey: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      ttlQueueKey: string,
      workerQueueKey: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      //args
      queueName: string,
      messageId: string,
      messageData: string,
      messageScore: string,
      ttlMember: string,
      ttlScore: string,
      messageKeyValue: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      currentTime: string,
      enableFastPath: string,
      callback?: Callback<number>
    ): Result<number, Context>;

    expireTtlRuns(
      //keys
      ttlQueueKey: string,
      //args
      keyPrefix: string,
      currentTime: string,
      batchSize: string,
      shardCount: string,
      workerQueueKey: string,
      workerItemsKey: string,
      visibilityTimeoutMs: string,
      callback?: Callback<string[]>
    ): Result<string[], Context>;

    dequeueMessagesFromQueue(
      //keys
      childQueue: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      messageKeyPrefix: string,
      envQueueKey: string,
      masterQueueKey: string,
      ttlQueueKey: string,
      //args
      childQueueName: string,
      currentTime: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      keyPrefix: string,
      maxCount: string,
      callback?: Callback<string[]>
    ): Result<string[], Context>;

    dequeueMessageFromWorkerQueueNonBlocking(
      workerQueueKey: string,
      callback?: Callback<[string, string] | undefined>
    ): Result<[string, string] | undefined, Context>;

    dequeueMessageFromKey(
      // keys
      messageKey: string,
      // args
      keyPrefix: string,
      callback?: Callback<string | null>
    ): Result<string | null, Context>;

    acknowledgeMessage(
      // keys
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      workerQueueKey: string,
      // args
      messageId: string,
      messageQueueName: string,
      messageKeyValue: string,
      removeFromWorkerQueue: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    clearMessageFromConcurrencySets(
      // keys
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      // args
      messageId: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    nackMessage(
      // keys
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      // args
      messageId: string,
      messageQueueName: string,
      messageData: string,
      messageScore: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    moveToDeadLetterQueue(
      // keys
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      deadLetterQueueKey: string,
      // args
      messageId: string,
      messageQueueName: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    releaseConcurrency(
      // keys
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      // args
      messageId: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    updateEnvironmentConcurrencyLimits(
      // keys
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      envConcurrencyLimit: string,
      envConcurrencyLimitBurstFactor: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    migrateLegacyMasterQueues(
      masterQueueKey: string,
      keyPrefix: string,
      ...queueNames: string[]
    ): Result<void, Context>;

    getMarkedRunsForAck(
      markedForAckKey: string,
      maxCount: string,
      callback?: Callback<string[]>
    ): Result<string[], Context>;

    // CK-aware commands
    enqueueMessageCk(
      //keys
      masterQueueKey: string,
      queue: string,
      messageKey: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      ckIndexKey: string,
      workerQueueKey: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      //args
      queueName: string,
      messageId: string,
      messageData: string,
      messageScore: string,
      ckWildcardName: string,
      messageKeyValue: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      currentTime: string,
      enableFastPath: string,
      callback?: Callback<number>
    ): Result<number, Context>;

    enqueueMessageWithTtlCk(
      //keys
      masterQueueKey: string,
      queue: string,
      messageKey: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      ttlQueueKey: string,
      ckIndexKey: string,
      workerQueueKey: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      //args
      queueName: string,
      messageId: string,
      messageData: string,
      messageScore: string,
      ttlMember: string,
      ttlScore: string,
      ckWildcardName: string,
      messageKeyValue: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      currentTime: string,
      enableFastPath: string,
      callback?: Callback<number>
    ): Result<number, Context>;

    dequeueMessagesFromCkQueue(
      //keys
      ckIndexKey: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      envCurrentConcurrencyKey: string,
      messageKeyPrefix: string,
      envQueueKey: string,
      masterQueueKey: string,
      ttlQueueKey: string,
      //args
      ckWildcardName: string,
      currentTime: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      keyPrefix: string,
      maxCount: string,
      callback?: Callback<string[]>
    ): Result<string[], Context>;

    acknowledgeMessageCk(
      // keys
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      workerQueueKey: string,
      ckIndexKey: string,
      // args
      messageId: string,
      messageQueueName: string,
      messageKeyValue: string,
      removeFromWorkerQueue: string,
      ckWildcardName: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    nackMessageCk(
      // keys
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      ckIndexKey: string,
      // args
      messageId: string,
      messageQueueName: string,
      messageData: string,
      messageScore: string,
      ckWildcardName: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    moveToDeadLetterQueueCk(
      // keys
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      deadLetterQueueKey: string,
      ckIndexKey: string,
      // args
      messageId: string,
      messageQueueName: string,
      ckWildcardName: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    // Tracked variants: maintain per-base-queue lengthCounter / runningCounter
    // for CK queues. See defineCommand bodies for details.
    enqueueMessageCkTracked(
      masterQueueKey: string,
      queue: string,
      messageKey: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      ckIndexKey: string,
      workerQueueKey: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      lengthCounterKey: string,
      baseQueueKey: string,
      queueName: string,
      messageId: string,
      messageData: string,
      messageScore: string,
      ckWildcardName: string,
      messageKeyValue: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      currentTime: string,
      enableFastPath: string,
      keyPrefix: string,
      counterTtl: string,
      callback?: Callback<number>
    ): Result<number, Context>;

    enqueueMessageWithTtlCkTracked(
      masterQueueKey: string,
      queue: string,
      messageKey: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      ttlQueueKey: string,
      ckIndexKey: string,
      workerQueueKey: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      lengthCounterKey: string,
      baseQueueKey: string,
      queueName: string,
      messageId: string,
      messageData: string,
      messageScore: string,
      ttlMember: string,
      ttlScore: string,
      ckWildcardName: string,
      messageKeyValue: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      currentTime: string,
      enableFastPath: string,
      keyPrefix: string,
      counterTtl: string,
      callback?: Callback<number>
    ): Result<number, Context>;

    dequeueMessagesFromCkQueueTracked(
      ckIndexKey: string,
      queueConcurrencyLimitKey: string,
      envConcurrencyLimitKey: string,
      envConcurrencyLimitBurstFactorKey: string,
      envCurrentConcurrencyKey: string,
      messageKeyPrefix: string,
      envQueueKey: string,
      masterQueueKey: string,
      ttlQueueKey: string,
      lengthCounterKey: string,
      ckWildcardName: string,
      currentTime: string,
      defaultEnvConcurrencyLimit: string,
      defaultEnvConcurrencyBurstFactor: string,
      keyPrefix: string,
      maxCount: string,
      callback?: Callback<string[]>
    ): Result<string[], Context>;

    dequeueMessageFromKeyTracked(
      messageKey: string,
      keyPrefix: string,
      counterTtl: string,
      callback?: Callback<string | null>
    ): Result<string | null, Context>;

    acknowledgeMessageCkTracked(
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      workerQueueKey: string,
      ckIndexKey: string,
      lengthCounterKey: string,
      runningCounterKey: string,
      messageId: string,
      messageQueueName: string,
      messageKeyValue: string,
      removeFromWorkerQueue: string,
      ckWildcardName: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    nackMessageCkTracked(
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      ckIndexKey: string,
      lengthCounterKey: string,
      runningCounterKey: string,
      messageId: string,
      messageQueueName: string,
      messageData: string,
      messageScore: string,
      ckWildcardName: string,
      keyPrefix: string,
      counterTtl: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    moveToDeadLetterQueueCkTracked(
      masterQueueKey: string,
      messageKey: string,
      messageQueue: string,
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      envQueueKey: string,
      deadLetterQueueKey: string,
      ckIndexKey: string,
      lengthCounterKey: string,
      runningCounterKey: string,
      messageId: string,
      messageQueueName: string,
      ckWildcardName: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    expireTtlRunsTracked(
      ttlQueueKey: string,
      keyPrefix: string,
      currentTime: string,
      batchSize: string,
      shardCount: string,
      workerQueueKey: string,
      workerItemsKey: string,
      visibilityTimeoutMs: string,
      callback?: Callback<string[]>
    ): Result<string[], Context>;

    releaseConcurrencyTracked(
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      runningCounterKey: string,
      ckIndexKey: string,
      messageId: string,
      keyPrefix: string,
      counterTtl: string,
      callback?: Callback<void>
    ): Result<void, Context>;

    clearMessageFromConcurrencySetsTracked(
      queueCurrentConcurrencyKey: string,
      envCurrentConcurrencyKey: string,
      queueCurrentDequeuedKey: string,
      envCurrentDequeuedKey: string,
      runningCounterKey: string,
      ckIndexKey: string,
      messageId: string,
      keyPrefix: string,
      counterTtl: string,
      callback?: Callback<void>
    ): Result<void, Context>;
  }
}

type ParseRawMessageResult = [Error | null, OutputPayload | null];

function parseRawMessage(rawMessage: string): ParseRawMessageResult {
  const deserializedMessage = safeJsonParse(rawMessage);

  const message = OutputPayload.safeParse(deserializedMessage);

  // If the message is not valid, return the error to be logged
  // but also still return the deserialized message as the OutputPayload
  // to prevent forward compatibility issues
  if (!message.success) {
    return [message.error, deserializedMessage as OutputPayload];
  }

  return [null, message.data];
}
