import { type Job, Queue, Worker } from "bullmq";
import { Redis } from "ioredis";
import { requestContext } from "../api/request-context";
import type { DbConnection, DbRow } from "../db/connection";
import { createTenantDb, createUncheckedSystemDb } from "../db/tenant-db";
import { createDerivativesContext } from "../derivatives/derivatives-context";
import { createSystemUser } from "../engine/system-user";
import {
  type AppContext,
  type DispatchWriteRef,
  type JobContext,
  type JobRunIn,
  type Registry,
  type SessionUser,
  SYSTEM_TENANT_ID,
  type TenantId,
} from "../engine/types";
import { createFileContext } from "../files/file-handle";
import { createFallbackLogger } from "../logging";
import type { Logger } from "../logging/types";
import {
  emitJobQueueDepth,
  getFallbackTracer,
  type Meter,
  type SerializedTraceContext,
  type Tracer,
} from "../observability";
import { createDistributedLock, type DistributedLock } from "../pipeline/distributed-lock";
import { RedisKeys } from "../pipeline/redis-keys";

// Queue-name convention: <prefix>-<lane>. The prefix is fixed in prod
// ("kumiko-jobs") — it must match between enqueuers and consumers, and an
// accidental drift would silently drop jobs. Tests override via
// `queueNamePrefix` for per-run isolation (stale jobs from a prior run
// don't leak into a new test because the queue name includes a timestamp).
const DEFAULT_QUEUE_NAME_PREFIX = "kumiko-jobs";
function queueNameFor(prefix: string, lane: JobRunIn): string {
  return `${prefix}-${lane}`;
}

/**
 * BullMQ job ids are `repeat:<schedulerId>:<millis>`. Colons inside the
 * scheduler id push the segment count to ≥5, which BullMQ's legacy heuristic
 * treated as old repeatables — spawning a new scheduler entry every tick and
 * leaking permanent `repeat:*` hashes (taskforcesh/bullmq#3828, fw#1603 /
 * publicstatus Redis OOM). Strip `.` and `:` from the job QN.
 */
export function schedulerIdForJobName(jobName: string): string {
  return `scheduler-${jobName.replace(/[.:]/g, "-")}`;
}

// Same colon-in-BullMQ-id hazard as schedulerIdForJobName (fw#1603/#1604) —
// a QN like "publicstatus:job:uptime-probe" must not leave ":" in the id.
export function bootJobIdForJobName(jobName: string): string {
  return `boot-${jobName.replace(/[.:]/g, "-")}`;
}

// ponytail: migration shim, remove after fw#1603 deploy is everywhere.
function legacySchedulerIdForJobName(jobName: string): string {
  return `scheduler-${jobName.replace(/\./g, "-")}`;
}

export type JobLogEntry = {
  level: "info" | "warn" | "error";
  message: string;
  timestamp: Temporal.Instant;
};

function createJobLogger(logs: JobLogEntry[], baseContext: Record<string, unknown> = {}): Logger {
  function push(level: "info" | "warn" | "error", msg: string, data?: Record<string, unknown>) {
    const merged = { ...baseContext, ...data };
    const message = Object.keys(merged).length > 0 ? `${msg} ${JSON.stringify(merged)}` : msg;
    logs.push({ level, message, timestamp: Temporal.Now.instant() });
  }
  const logger: Logger = {
    info(msg, data) {
      push("info", msg, data);
    },
    warn(msg, data) {
      push("warn", msg, data);
    },
    error(msg, data) {
      push("error", msg, data);
    },
    // No-op by design, not missing coverage: JobLogEntry.level (and the
    // persisted run-completed/-failed event schema in bundled-features/
    // jobs/events.ts) only has info|warn|error — a job's debug() calls have
    // nowhere durable to land.
    debug() {},
    child(context) {
      return createJobLogger(logs, { ...baseContext, ...context });
    },
  };
  return logger;
}

export type JobMeta = {
  triggeredById?: string | undefined;
  payload?: string | undefined;
  // BullMQ numbers retries from 1 upward; the logger threads this into
  // the run-started event so audit queries can distinguish "fresh run" vs.
  // "nth retry" without joining back to BullMQ-internals.
  attempt?: number | undefined;
  // BullMQ job priority (lower = processed first). Set per-dispatch so the same
  // job definition can be enqueued at different urgencies (e.g. delivery maps
  // critical/normal/low onto it).
  priority?: number | undefined;
};

export type JobRunner = {
  start(): Promise<void>;
  stop(): Promise<void>;
  dispatch(jobName: string, payload?: Record<string, unknown>, meta?: JobMeta): Promise<string>;
  handleEvent(
    eventName: string,
    payload: Record<string, unknown>,
    user?: SessionUser,
  ): Promise<void>;
  // Wires JobContext.write/queryAs to the real dispatcher — called once at
  // boot, after the dispatcher exists (job-runner construction happens
  // before it). Before this runs, JobContext.write/queryAs throw.
  attachDispatcher(ref: DispatchWriteRef): void;
};

export type JobRunnerOptions = {
  registry: Registry;
  context: AppContext;
  redisUrl: string;
  // Which lane this runner CONSUMES — i.e. starts a BullMQ worker for and
  // schedules cron/boot jobs on. Undefined = enqueuer-only: the runner
  // still holds queue-clients for BOTH lanes so dispatch()/handleEvent()
  // can enqueue jobs destined for either lane, but no BullMQ worker is
  // started and no cron schedules fire. API processes that don't
  // runLocalJobs leave this unset; worker processes set "worker"; api-
  // processes with runLocalJobs set "api".
  consumerLane?: JobRunIn | undefined;
  // Override the queue-name prefix. Prod uses the default ("kumiko-jobs").
  // Tests set a unique prefix (e.g. `"test-${Date.now()}"`) for isolation —
  // two parallel test-runners never see each other's jobs.
  queueNamePrefix?: string | undefined;
  // Override how long start() waits for the worker's Redis connection
  // before failing boot. Defaults to BOOT_REDIS_TIMEOUT_MS; tests shrink it
  // to keep an unreachable-Redis assertion fast.
  bootRedisTimeoutMs?: number | undefined;
  getActiveTenantIds?: () => Promise<TenantId[]>;
  onJobStart?: (jobName: string, jobId: string, meta: JobMeta) => void;
  onJobComplete?: (jobName: string, jobId: string, duration: number, logs: JobLogEntry[]) => void;
  onJobFailed?: (jobName: string, jobId: string, error: string, logs: JobLogEntry[]) => void;
};

// Serialized trace context lives under this key in the BullMQ job data.
// Leading underscore matches the existing internal-meta convention
// (_triggeredById, _tenantId, _payload).
const TRACE_CONTEXT_KEY = "_traceContext";

function readTraceContext(data: Record<string, unknown>): SerializedTraceContext | undefined {
  const raw = data[TRACE_CONTEXT_KEY];
  if (!raw || typeof raw !== "object") return undefined;
  const ctx = raw as Partial<SerializedTraceContext>; // @cast-boundary engine-payload
  if (!ctx.traceId || !ctx.spanId) return undefined;
  return { traceId: ctx.traceId, spanId: ctx.spanId };
}

function captureTraceContext(tracer: Tracer): SerializedTraceContext | undefined {
  const span = tracer.getActiveSpan();
  if (!span?.traceId) return undefined;
  return { traceId: span.traceId, spanId: span.spanId };
}

const QUEUE_DEPTH_POLL_INTERVAL_MS = 15_000;

// kumiko_job_queue_depth: only the lane's own consumer polls — the count is
// a global Redis-backed value (BullMQ, not per-process), so one reporter per
// lane is enough. Fires once immediately (metrics exist before the first
// poll tick) then on an interval; the caller stores + clears the handle.
async function startQueueDepthPolling(
  queue: Queue,
  lane: JobRunIn,
  meter: Meter,
): Promise<ReturnType<typeof setInterval>> {
  const pollQueueDepth = async (): Promise<void> => {
    try {
      const counts = await queue.getJobCounts("waiting", "active", "delayed", "failed", "paused");
      emitJobQueueDepth(meter, lane, counts);
    } catch {
      // skip: transient Redis hiccup — next poll retries
    }
  };
  await pollQueueDepth();
  return setInterval(() => void pollQueueDepth(), QUEUE_DEPTH_POLL_INTERVAL_MS);
}

function parseRedisOpts(url: string): { host: string; port: number; db?: number | undefined } {
  const parsed = new URL(url);
  const result: { host: string; port: number; db?: number | undefined } = {
    host: parsed.hostname,
    port: Number(parsed.port) || 6379,
  };
  if (parsed.pathname.length > 1) {
    result.db = Number(parsed.pathname.slice(1));
  }
  return result;
}

// redisOpts carries no connectTimeout/retry cap, so an unreachable Redis
// would otherwise hang start() forever with no health endpoint to notice.
const BOOT_REDIS_TIMEOUT_MS = 10_000;

function timeoutReject(ms: number, message: string): Promise<never> {
  return new Promise((_, reject) => {
    setTimeout(() => reject(new Error(message)), ms);
  });
}

export function createJobRunner(options: JobRunnerOptions): JobRunner {
  const { registry, context, redisUrl, consumerLane } = options;
  const queueNamePrefix = options.queueNamePrefix ?? DEFAULT_QUEUE_NAME_PREFIX;
  const bootRedisTimeoutMs = options.bootRedisTimeoutMs ?? BOOT_REDIS_TIMEOUT_MS;
  const redisOpts = parseRedisOpts(redisUrl);
  // Use the context's tracer when present (observability-provider injected at
  // boot); otherwise noop so dispatch/handleJob stay zero-cost without config.
  const tracer: Tracer = context.tracer ?? getFallbackTracer();
  const errorLogger = createFallbackLogger("job-runner", context.log);
  // Set at the top of stop() — a graceful shutdown closes the redis/BullMQ
  // clients itself, which fires the exact same 'error' listeners below with
  // an expected "Connection is closed." Downgrading to debug once stopping
  // is true keeps those out of error-rate alerts without losing them.
  let stopping = false;

  const allJobs = registry.getAllJobs();

  // Resolve the lane for a job — "worker" is the default because that's the
  // sensible prod lane (heavy async off the request path). Jobs that opted
  // into "api" must have been validated at registry boot already.
  function laneForJob(def: { readonly runIn?: JobRunIn | undefined }): JobRunIn {
    return def.runIn ?? "worker";
  }

  // Sequential coordination: BullMQ OSS has no `group`, so we serialise
  // same-name jobs ourselves with a per-name Redis lock. Only built when at
  // least one job actually requested it — keeps the no-sequential boot path
  // free of the extra Redis client. Scoped under the consumer lane so two
  // runners on different lanes cannot collide on the same lock-key for
  // unrelated jobs.
  const hasSequential = [...allJobs.values()].some((def) => def.concurrency === "sequential");
  let lockRedis: Redis | null = null;
  let sequentialLock: DistributedLock | null = null;
  if (hasSequential) {
    lockRedis = new Redis(redisOpts);
    // Without a listener, a post-close 'error' (e.g. a teardown-race
    // "Connection is closed") is unhandled and crashes the process — in
    // bun:test it gets attributed to whichever test happens to run next
    // (fw#1805).
    lockRedis.on("error", (err) => {
      const log = stopping ? errorLogger.debug : errorLogger.error;
      log("lock redis connection error", { error: err.message });
    });
    const lockScope = consumerLane ?? "enqueue";
    sequentialLock = createDistributedLock(lockRedis, `${RedisKeys.lock}seq:${lockScope}:`);
  }
  // Default lock-TTL for sequential jobs that didn't declare a timeout.
  // 5 minutes matches BullMQ's default stalledInterval — long enough for
  // any reasonable handler, short enough that a crashed worker recovers
  // without manual intervention.
  const SEQUENTIAL_DEFAULT_TTL_SEC = 305;
  // How long to wait before re-trying a busy sequential lock. Short enough
  // to feel responsive, long enough that we don't hammer Redis.
  const SEQUENTIAL_RETRY_DELAY_MS = 200;

  // Two queue-clients — one per lane. Every runner holds both, regardless of
  // its own consumerLane, so dispatch()/handleEvent() always write to the
  // queue matching the target job's runIn. Client-creation is cheap (shared
  // ioredis connection via bullmq), so this doesn't scale with number of
  // processes.
  const queues: Readonly<Record<JobRunIn, Queue>> = {
    api: new Queue(queueNameFor(queueNamePrefix, "api"), { connection: redisOpts }),
    worker: new Queue(queueNameFor(queueNamePrefix, "worker"), { connection: redisOpts }),
  };
  // Same unhandled-'error'-crash hazard as lockRedis above, just via
  // BullMQ's internal ioredis client (fw#1805).
  for (const queue of Object.values(queues)) {
    queue.on("error", (err) => {
      const log = stopping ? errorLogger.debug : errorLogger.error;
      log("queue redis connection error", { error: err.message });
    });
  }
  let worker: Worker | null = null;
  let queueDepthTimer: ReturnType<typeof setInterval> | null = null;
  // Forward reference to the runner's own API, exposed on the job-handler ctx
  // so a handler can dispatch a follow-up job (job→job chaining, e.g.
  // delivery.render → delivery.send). Assigned just before return; reads happen
  // at job-execution time (after start()), so it is always defined by then.
  let selfRunner: JobRunner | undefined;
  // Set by attachDispatcher() once the boot-level dispatcher exists.
  // JobContext.write/queryAs throw until this is set — see JobContext doc.
  let dispatchWriteRef: DispatchWriteRef | undefined;

  // Counts active + waiting jobs with this name for this tenant across
  // BOTH lane queues. Jobs with the same name should only live in one
  // lane (jobDef.runIn is static), but walking both is cheap and avoids
  // a subtle bug if someone ever reassigns a job to a different lane
  // between deploys while old queue contents are still draining.
  async function isOverPerTenantLimit(
    jobName: string,
    tenantId: string,
    max: number,
  ): Promise<boolean> {
    const results = await Promise.all([
      queues.api.getActive(),
      queues.api.getWaiting(),
      queues.worker.getActive(),
      queues.worker.getWaiting(),
    ]);
    let count = 0;
    for (const list of results) {
      for (const j of list) {
        if (j.name !== jobName) continue;
        const t = (j.data as { _tenantId?: string } | undefined)?._tenantId; // @cast-boundary dynamic-key
        if (t === tenantId) {
          count += 1;
          if (count >= max) return true;
        }
      }
    }
    return false;
  }

  async function handleJob(bullJob: Job): Promise<void> {
    const rawName = bullJob.name;

    // Handle perTenant dispatch jobs — fan out to one job per tenant. The
    // fan-out re-enqueues into the lane the actual job is assigned to;
    // the _perTenant wrapper itself always lives in the consumer-lane
    // (it's picked up by this runner's own worker).
    if (rawName.startsWith("_perTenant:")) {
      const actualName = rawName.slice("_perTenant:".length);
      if (!options.getActiveTenantIds) {
        throw new Error(`perTenant job "${actualName}" requires getActiveTenantIds option`);
      }
      const actualDef = allJobs.get(actualName);
      if (!actualDef) {
        throw new Error(`Unknown job: ${actualName}`);
      }
      const tenantIds = await options.getActiveTenantIds();
      const targetQueue = queues[laneForJob(actualDef)];
      for (const tenantId of tenantIds) {
        await targetQueue.add(actualName, { ...bullJob.data, _tenantId: tenantId });
      }
      // skip: fan-out dispatcher job, per-tenant children enqueued
      return;
    }

    const jobName = rawName;
    const jobDef = allJobs.get(jobName);
    if (!jobDef) {
      throw new Error(`Unknown job: ${jobName}`);
    }

    // Sequential gate: try to claim the per-name lock. If another worker
    // (or this worker on a different bullJob) holds it, re-enqueue with a
    // small delay and exit *successfully* — using throw would burn the
    // job's retry budget and pollute failure metrics, but a re-enqueue
    // looks like an ordinary handoff to BullMQ.
    let sequentialToken: string | null = null;
    if (jobDef.concurrency === "sequential" && sequentialLock) {
      const ttlSec = jobDef.timeout
        ? Math.ceil(jobDef.timeout / 1000) + 5
        : SEQUENTIAL_DEFAULT_TTL_SEC;
      sequentialToken = await sequentialLock.acquire(jobName, { ttlSeconds: ttlSec });
      if (!sequentialToken) {
        // Re-enqueue onto the job's own lane-queue. In practice that's the
        // same queue the worker just picked from (since only the consuming
        // lane runs handleJob at all), but route explicitly — no implicit
        // coupling to "whichever queue the caller happened to be on".
        await queues[laneForJob(jobDef)].add(jobName, bullJob.data, {
          delay: SEQUENTIAL_RETRY_DELAY_MS,
        });
        // skip: lock taken, work re-enqueued with delay, current invocation done
        return;
      }
    }

    const jobId = bullJob.id ?? "unknown";
    const startTime = Date.now();
    const logs: JobLogEntry[] = [];

    // Extract meta from job data. `attempt` is BullMQ's own counter
    // (1-based on the first run, incremented on each retry) — threading
    // it through lets the logger tag the run-started event with the
    // retry number, so audit queries distinguish fresh from retry runs
    // without peeking at BullMQ internals.
    const rawData = bullJob.data as DbRow;
    const meta: JobMeta = {
      triggeredById: rawData["_triggeredById"] as string | undefined, // @cast-boundary dynamic-key
      payload: rawData["_payload"] as string | undefined, // @cast-boundary dynamic-key
      attempt: bullJob.attemptsMade + 1,
    };

    // Build handler payload (without internal meta fields)
    const payload: Record<string, unknown> = {};
    for (const [k, v] of Object.entries(rawData)) {
      if (!k.startsWith("_")) payload[k] = v;
    }

    // Determine tenantId and triggeredBy from meta
    const tenantId =
      (rawData["_tenantId"] as string | undefined) ?? // @cast-boundary dynamic-key
      (payload["tenantId"] as string | undefined) ?? // @cast-boundary dynamic-key
      SYSTEM_TENANT_ID;
    const triggeredById = (rawData["_triggeredById"] as string | undefined) ?? null; // @cast-boundary dynamic-key

    // Carry `_triggerName` from rawData when set — handleEvent injects it on
    // multi-trigger dispatch; exposed as jobContext.triggerName so handlers
    // don't dig through the raw payload themselves.
    const triggerName = rawData["_triggerName"] as string | undefined; // @cast-boundary dynamic-key
    // Mirror dispatch-shared.ts buildHandlerContext: ctx.files must resolve
    // through the same _fileProviderResolver for jobs as for write-handlers,
    // otherwise event-triggered jobs silently get an unresolved ctx.files.
    const fileResolver = context._fileProviderResolver;
    const files = fileResolver ? createFileContext(() => fileResolver(tenantId)) : context.files;
    const jobSystemUser = createSystemUser(tenantId);
    // Same buildHandlerContext parity as ctx.files above: ctx.notify/ctx.config
    // must resolve for jobs the same way they do for write-handlers, or an
    // app-author job calling ctx.notify(...)/ctx.config(...) hits a TypeError
    // at runtime the write-handler path never would (framework#1532).
    const notify = context._notifyFactory?.(jobSystemUser, tenantId);
    const configDb = context.db as DbConnection | undefined; // @cast-boundary db-operator
    // Shared by the config accessor and ctx.derivatives below — both need the
    // same tenant-scoped db, and building it twice would let the two calls
    // drift apart. Always "system" mode regardless of the job's own
    // systemScope() status (pre-existing, not something this change alters)
    // — isSystemJob below is what actually keeps ctx.systemDb off a
    // non-system job; it is the only thing standing between this db and an
    // unchecked cross-tenant escape hatch for such a job.
    const tenantScopedDb = configDb ? createTenantDb(configDb, tenantId, "system") : undefined;
    const isSystemJob = registry.isJobSystemScoped(jobName);
    const systemDb =
      isSystemJob && tenantScopedDb ? createUncheckedSystemDb(tenantScopedDb) : undefined;
    const config =
      context._configAccessorFactory && tenantScopedDb
        ? context._configAccessorFactory({
            user: { id: jobSystemUser.id, tenantId },
            db: tenantScopedDb,
            secrets: context.secrets,
          })
        : undefined;
    // Mirror dispatch-shared.ts: ctx.derivatives needs files+db, same
    // tenant-scoped db the config accessor above uses.
    const derivatives =
      files && tenantScopedDb
        ? createDerivativesContext({
            files,
            registry,
            db: tenantScopedDb,
            tenantId,
          })
        : context.derivatives;
    const jobContext: JobContext = {
      ...context,
      // Same union as configDb above — job runners are always constructed
      // with a real DbConnection; JobContext requires it non-optional.
      db: configDb as DbConnection, // @cast-boundary db-operator
      files,
      derivatives,
      ...(notify !== undefined && { notify }),
      ...(config !== undefined && { config }),
      ...(systemDb && { systemDb }),
      // The runner owns the registry it resolved this job from — expose it so
      // workers can reach projections/jobs without the app author duplicating
      // it into `context` (the JobContext contract guarantees `registry`).
      registry,
      // Expose the runner so handlers can chain a follow-up job. Symmetric with
      // how the command-dispatcher hands write-handlers their jobRunner.
      ...(selfRunner !== undefined && { jobRunner: selfRunner }),
      systemUser: jobSystemUser,
      triggeredBy: triggeredById !== null ? { id: triggeredById, tenantId } : null,
      log: createJobLogger(logs),
      ...(triggerName !== undefined && { triggerName }),
      write: (qn: string, payload: unknown) => {
        if (!dispatchWriteRef) {
          throw new Error(
            "JobContext.write called before dispatcher attached — call attachDispatcher() first",
          );
        }
        return dispatchWriteRef.write(jobSystemUser, qn, payload);
      },
      queryAs: (user: SessionUser, qn: string, payload: unknown) => {
        if (!dispatchWriteRef) {
          throw new Error(
            "JobContext.queryAs called before dispatcher attached — call attachDispatcher() first",
          );
        }
        return dispatchWriteRef.queryAs(user, qn, payload);
      },
    };

    await options.onJobStart?.(jobName, jobId, meta);

    // Cross-process trace continuation: if the enqueuing code captured a
    // parent span, start the job.execute span as its child. Works for event
    // and manual triggers; cron jobs start a fresh root span.
    const parentContext = readTraceContext(rawData);

    // Correlation propagation: the scheduling request's correlationId was
    // packed into _correlationId at dispatch time. Re-enter requestContext.run
    // so event writes during this job stamp the same correlation as the
    // request that scheduled it. Cron/boot jobs (no scheduler) start fresh
    // — correlationId = new requestId, no parent causation.
    const inheritedCorrelationId = (rawData["_correlationId"] as string | undefined) ?? undefined; // @cast-boundary dynamic-key
    const jobRequestId = requestContext.generateId();
    const jobCorrelationId = inheritedCorrelationId ?? jobRequestId;

    const runInSpan = async (): Promise<void> => {
      try {
        await requestContext.run({ requestId: jobRequestId, correlationId: jobCorrelationId }, () =>
          jobDef.handler(payload, jobContext),
        );
        const duration = Date.now() - startTime;
        await options.onJobComplete?.(jobName, jobId, duration, logs);
      } catch (err) {
        const errorMsg = err instanceof Error ? err.message : String(err);
        logs.push({ level: "error", message: errorMsg, timestamp: Temporal.Now.instant() });
        await options.onJobFailed?.(jobName, jobId, errorMsg, logs);
        throw err;
      }
    };

    // Unified span creation: withSpan handles start/end + status/exception
    // recording identically for both parent-context and no-parent paths.
    // When parentContext is set, the new parent-aware StartSpanOptions
    // plumbs it through to startSpan — no manual try/finally needed.
    try {
      await tracer.withSpan(
        "job.execute",
        {
          attributes: {
            "job.name": jobName,
            "job.id": jobId,
            "job.attempt": bullJob.attemptsMade + 1,
            "kumiko.tenant_id": tenantId,
            // Lane-routing attributes (Welle 2.6). `run_in` is the job's
            // declared lane (explicit or default-"worker"); `consumer_lane`
            // is which runner actually executed it. They diverge in
            // all-in-one (both lanes live in one process) but must match
            // in split deploys — a mismatch in prod logs signals a
            // misrouted job that slipped past the boot-validator.
            "kumiko.job.run_in": laneForJob(jobDef),
            // Omit attribute entirely when no consumer (enqueuer-only runner) —
            // SpanAttributeValue doesn't accept undefined.
            ...(consumerLane !== undefined ? { "kumiko.job.consumer_lane": consumerLane } : {}),
          },
          ...(parentContext ? { parent: parentContext } : {}),
        },
        runInSpan,
      );
    } finally {
      // Release the sequential lock value-matched (Lua compare-and-delete
      // inside DistributedLock). A TTL-expired lock that's been claimed by
      // a different owner stays put — releasing it would break sequencing
      // for the new owner.
      if (sequentialToken && sequentialLock) {
        await sequentialLock.release(jobName, sequentialToken);
      }
    }
  }

  const runnerApi: JobRunner = {
    async start(): Promise<void> {
      // skip: enqueuer-only runner — no BullMQ worker, no cron schedules,
      // no boot jobs. The API-process (runLocalJobs=false) lands here; it
      // still holds the queue-clients so dispatch()/handleEvent() can
      // target the worker-lane queue, but nothing local consumes.
      if (!consumerLane) {
        return;
      }

      const consumerQueue = queues[consumerLane];
      worker = new Worker(queueNameFor(queueNamePrefix, consumerLane), handleJob, {
        connection: redisOpts,
        concurrency: 5,
      });
      worker.on("error", (err) => {
        const log = stopping ? errorLogger.debug : errorLogger.error;
        log("worker redis connection error", { error: err.message });
      });
      // A caller that calls stop() right after start() otherwise races the
      // still-settling blocking connection: it rejects in-flight commands
      // via ioredis's flushQueue() during close(), which isn't a listenable
      // 'error' event — the only fix is to not return until both of the
      // worker's connections (main + blocking) are ready (fw#1805). This
      // mirrors the wait BullMQ already does internally for
      // upsertJobScheduler()/add() below when the lane has a cron/boot job.
      // Racing a timeout against it keeps an unreachable Redis from hanging
      // start() forever — there's no worker health endpoint to notice.
      await Promise.race([
        worker.waitUntilReady(),
        timeoutReject(
          bootRedisTimeoutMs,
          `job-runner: Redis not reachable within ${bootRedisTimeoutMs}ms (lane=${consumerLane})`,
        ),
      ]);

      // Only schedule cron + boot for jobs that belong to this lane. Jobs
      // assigned to the other lane get their cron/boot wiring from the
      // runner running on that lane. Running both here would double-fire.
      for (const [name, jobDef] of allJobs) {
        if (laneForJob(jobDef) !== consumerLane) continue;
        if ("cron" in jobDef.trigger) {
          const schedulerId = schedulerIdForJobName(name);
          const legacyId = legacySchedulerIdForJobName(name);
          // Drop pre-sanitize scheduler ids so colon-form ghosts stop firing.
          if (legacyId !== schedulerId) {
            try {
              await consumerQueue.removeJobScheduler(legacyId);
            } catch {
              // skip: legacy scheduler absent (fresh install / already purged)
            }
          }
          await consumerQueue.upsertJobScheduler(
            schedulerId,
            { pattern: jobDef.trigger.cron },
            {
              name: jobDef.perTenant ? `_perTenant:${name}` : name,
              data: {},
              opts: {
                removeOnComplete: { count: 100 },
                removeOnFail: { count: 50 },
              },
            },
          );
        }
      }

      for (const [name, jobDef] of allJobs) {
        if (laneForJob(jobDef) !== consumerLane) continue;
        if (jobDef.runOnBoot) {
          const bootName = jobDef.perTenant ? `_perTenant:${name}` : name;
          await consumerQueue.add(bootName, {}, { jobId: bootJobIdForJobName(name) });
        }
      }

      // Skipped when no meter is wired (context.meter is optional, e.g. in
      // tests without an observability provider).
      if (context.meter) {
        queueDepthTimer = await startQueueDepthPolling(consumerQueue, consumerLane, context.meter);
      }
    },

    async stop(): Promise<void> {
      stopping = true;
      if (queueDepthTimer) {
        clearInterval(queueDepthTimer);
        queueDepthTimer = null;
      }
      if (worker) {
        await worker.close();
        worker = null;
      }
      await Promise.all([queues.api.close(), queues.worker.close()]);
      if (lockRedis) {
        // quit() drains in-flight commands; disconnect() would cancel them
        // and risk a half-released lock.
        await lockRedis.quit();
        lockRedis = null;
      }
    },

    async dispatch(
      jobName: string,
      payload?: Record<string, unknown>,
      meta?: JobMeta,
    ): Promise<string> {
      const jobDef = allJobs.get(jobName);
      if (!jobDef) {
        throw new Error(`Unknown job: ${jobName}`);
      }

      // Route to the job's declared lane, not the runner's consumer lane —
      // an api-runner is allowed to enqueue a worker-lane job and vice
      // versa (that's the whole point of both queues being held).
      const targetQueue = queues[laneForJob(jobDef)];

      // perTenant: dispatch the fan-out wrapper instead
      if (jobDef.perTenant) {
        const job = await targetQueue.add(`_perTenant:${jobName}`, payload ?? {});
        return job.id ?? "unknown";
      }

      // maxPerTenant guard: cap concurrent + waiting jobs of the same name
      // for the same tenant. Orthogonal to the concurrency mode below — runs
      // first because if we're over the limit nothing else matters.
      // Requires a `_tenantId` in the payload to know which bucket to count
      // against; without it the guard is inactive (system jobs, ambient
      // dispatch). Fan-out children of perTenant jobs land here on their
      // recursive queue.add and DO carry _tenantId.
      if (jobDef.maxPerTenant !== undefined) {
        const tenantId = (payload as { _tenantId?: string } | undefined)?._tenantId; // @cast-boundary dynamic-key
        if (
          tenantId !== undefined &&
          (await isOverPerTenantLimit(jobName, tenantId, jobDef.maxPerTenant))
        ) {
          return "skipped:max-per-tenant";
        }
      }

      const concurrency = jobDef.concurrency ?? "parallel";
      const bullOpts: Record<string, unknown> = {};

      switch (concurrency) {
        case "skip": {
          const active = await targetQueue.getActive();
          const waiting = await targetQueue.getWaiting();
          const isRunning = [...active, ...waiting].some((j) => j.name === jobName);
          if (isRunning) {
            return "skipped";
          }
          break;
        }
        case "replace": {
          const waiting = await targetQueue.getWaiting();
          for (const j of waiting) {
            if (j.name === jobName && j.id) {
              await j.remove();
            }
          }
          break;
        }
        // case "sequential" is rejected at boot — see createJobRunner. Once
        // the OSS-compatible implementation lands (per-name Redis lock),
        // re-add the dispatch branch here.
        case "debounce": {
          const debounceMs = jobDef.debounceMs ?? 5000;
          bullOpts["debounce"] = { id: jobName, ttl: debounceMs };
          break;
        }
        default:
          break;
      }

      if (jobDef.retries !== undefined) bullOpts["attempts"] = jobDef.retries + 1;
      if (jobDef.backoff) bullOpts["backoff"] = { type: jobDef.backoff };
      if (jobDef.timeout) bullOpts["timeout"] = jobDef.timeout;
      if (meta?.priority !== undefined) bullOpts["priority"] = meta.priority;

      // Pack meta into job data with _ prefix
      const data: Record<string, unknown> = { ...payload };
      if (meta?.triggeredById !== undefined) data["_triggeredById"] = meta.triggeredById;
      if (meta?.payload !== undefined) data["_payload"] = meta.payload;
      // Carry the enqueuing span context into the worker so job.execute shows
      // as a child of the caller.
      const traceCtx = captureTraceContext(tracer);
      if (traceCtx) data[TRACE_CONTEXT_KEY] = traceCtx;
      // Propagate correlation from the scheduling request into the job
      // execution context. The worker re-enters requestContext.run with
      // this value so ctx.appendEvent / executor writes during the job
      // stamp the same correlation as the HTTP request that scheduled it.
      const reqCtx = requestContext.get();
      if (reqCtx?.correlationId) data["_correlationId"] = reqCtx.correlationId;

      const job = await targetQueue.add(jobName, data, bullOpts);
      return job.id ?? "unknown";
    },

    async handleEvent(
      eventName: string,
      payload: Record<string, unknown>,
      user?: SessionUser,
    ): Promise<void> {
      const traceCtx = captureTraceContext(tracer);
      // Same correlation propagation as dispatch(): events triggered from
      // within a request (or an MSP-apply running under requestContext.run)
      // get their correlationId into job data so the job execution keeps
      // the same causal chain.
      const reqCtx = requestContext.get();
      for (const [name, jobDef] of allJobs) {
        if (!("on" in jobDef.trigger)) continue;
        // skip: andere Trigger-Formen (cron, manual) reagieren nicht auf
        // Events. Nur "on"-Trigger werden hier matched.
        const triggerOn = jobDef.trigger.on;
        const matches = Array.isArray(triggerOn)
          ? triggerOn.includes(eventName)
          : triggerOn === eventName;
        if (!matches) continue;
        const data: Record<string, unknown> = { ...payload };
        if (user) {
          data["_tenantId"] = user.tenantId;
          data["_triggeredById"] = user.id;
        }
        // Multi-Trigger: payload bekommt _triggerName damit der Handler
        // weiß, welcher der N Trigger gefeuert hat. Bei Single-Trigger
        // setzen wir es auch — kostet nichts und vereinfacht Handler-Code
        // (kein "ist es Multi?"-Branch nötig).
        data["_triggerName"] = eventName;
        if (traceCtx) data[TRACE_CONTEXT_KEY] = traceCtx;
        if (reqCtx?.correlationId) data["_correlationId"] = reqCtx.correlationId;
        // Same maxPerTenant guard as dispatch — events that fan into many
        // jobs must respect the per-tenant cap or the limit is one-sided.
        if (jobDef.maxPerTenant !== undefined && user?.tenantId !== undefined) {
          if (await isOverPerTenantLimit(name, String(user.tenantId), jobDef.maxPerTenant)) {
            continue;
          }
        }
        // Route to the job's declared lane, not a fixed queue — that's
        // the whole reason both queues are held.
        await queues[laneForJob(jobDef)].add(name, data);
      }
    },
    attachDispatcher(ref: DispatchWriteRef): void {
      dispatchWriteRef = ref;
    },
  };

  selfRunner = runnerApi;
  return runnerApi;
}