import { type RedisOptions } from "@internal/redis";
import { Meter, Tracer } from "@internal/tracing";
import { Logger, LogLevel } from "@trigger.dev/core/logger";
import {
  MachinePreset,
  MachinePresetName,
  RetryOptions,
  TriggerTraceContext,
} from "@trigger.dev/core/v3";
import { PrismaClient, PrismaReplicaClient, TaskRun, Waitpoint } from "@trigger.dev/database";
import {
  Worker,
  type WorkerConcurrencyOptions,
  type GlobalRateLimiter,
} from "@trigger.dev/redis-worker";
import { FairQueueSelectionStrategyOptions } from "../run-queue/fairQueueSelectionStrategy.js";
import { MinimalAuthenticatedEnvironment } from "../shared/index.js";
import { LockRetryConfig } from "./locking.js";
import { workerCatalog } from "./workerCatalog.js";
import { type BillingPlan } from "./billingCache.js";
import type { DRRConfig } from "../batch-queue/types.js";
import type { PendingVersionRunIdLookup } from "./services/pendingVersionLookup.js";

export type RunEngineOptions = {
  prisma: PrismaClient;
  readOnlyPrisma?: PrismaReplicaClient;
  worker: {
    disabled?: boolean;
    redis: RedisOptions;
    pollIntervalMs?: number;
    immediatePollIntervalMs?: number;
    shutdownTimeoutMs?: number;
  } & WorkerConcurrencyOptions;
  machines: {
    defaultMachine: MachinePresetName;
    machines: Record<string, MachinePreset>;
    baseCostInCents: number;
  };
  billing?: {
    getCurrentPlan: (orgId: string) => Promise<BillingPlan>;
  };
  queue: {
    redis: RedisOptions;
    shardCount?: number;
    masterQueueConsumersDisabled?: boolean;
    processWorkerQueueDebounceMs?: number;
    masterQueueConsumersIntervalMs?: number;
    masterQueueCooloffPeriodMs?: number;
    masterQueueCooloffCountThreshold?: number;
    masterQueueConsumerDequeueCount?: number;
    workerOptions?: WorkerConcurrencyOptions;
    retryOptions?: RetryOptions;
    defaultEnvConcurrency?: number;
    defaultEnvConcurrencyBurstFactor?: number;
    logLevel?: LogLevel;
    queueSelectionStrategyOptions?: Pick<
      FairQueueSelectionStrategyOptions,
      "parentQueueLimit" | "tracer" | "biases" | "reuseSnapshotCount" | "maximumEnvCount"
    >;
    dequeueBlockingTimeoutSeconds?: number;
    concurrencySweeper?: {
      scanSchedule?: string;
      processMarkedSchedule?: string;
      scanJitterInMs?: number;
      processMarkedJitterInMs?: number;
    };
    /** TTL system options 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 the entire TTL system is disabled (default: false) */
      disabled?: boolean;
      /** Whether TTL consumers + worker are disabled on this instance (default: false).
       *  When true, ZADD on enqueue still happens but polling loops and the TTL worker don't run. */
      consumersDisabled?: boolean;
      /** Visibility timeout for TTL worker jobs (ms, default: 120000) */
      visibilityTimeoutMs?: number;
      /** Concurrency limit for the TTL redis-worker (default: 1) */
      workerConcurrency?: number;
      /** Max items to accumulate before flushing a batch (default: 500) */
      batchMaxSize?: number;
      /** Max time (ms) to wait for more items before flushing a batch (default: 5000) */
      batchMaxWaitMs?: number;
    };
  };
  runLock: {
    redis: RedisOptions;
    duration?: number;
    automaticExtensionThreshold?: number;
    retryConfig?: LockRetryConfig;
  };
  cache?: {
    redis: RedisOptions;
  };
  batchQueue?: {
    redis: RedisOptions;
    drr?: Partial<DRRConfig>;
    /** Number of master queue shards (default: 1) */
    shardCount?: number;
    /** Worker queue blocking timeout in seconds (enables two-stage processing) */
    workerQueueBlockingTimeoutSeconds?: number;
    consumerEnabled?: boolean;
    consumerCount?: number;
    consumerIntervalMs?: number;
    /** Default processing concurrency per environment when no specific limit is set */
    defaultConcurrency?: number;
    /** Optional global rate limiter to limit processing across all consumers */
    globalRateLimiter?: GlobalRateLimiter;
    /** Maximum worker queue depth before claiming pauses (protects visibility timeouts) */
    workerQueueMaxDepth?: number;
    /** Retry configuration for failed batch items */
    retry?: {
      /** Maximum number of attempts (including the first). Default: 1 (no retries) */
      maxAttempts: number;
      /** Base delay in milliseconds. Default: 1000 */
      minTimeoutInMs?: number;
      /** Maximum delay in milliseconds. Default: 30000 */
      maxTimeoutInMs?: number;
      /** Exponential backoff factor. Default: 2 */
      factor?: number;
      /** Whether to add jitter to retry delays. Default: true */
      randomize?: boolean;
    };
  };
  debounce?: {
    redis?: RedisOptions;
    /** Maximum duration in milliseconds that a run can be debounced. Default: 1 hour */
    maxDebounceDurationMs?: number;
    /**
     * Bucket size in milliseconds used to quantize the newly computed `delayUntil`.
     * Quantization collapses many concurrent triggers on the same hot debounce key
     * into the same target time, so that the unlocked fast-path skip becomes
     * effective and the redlock on `handleDebounce` is not contended.
     *
     * A run might fire up to `quantizeNewDelayUntilMs` earlier than the strict
     * `now + delay` spec.
     *
     * Set to 0 to disable quantization.
     *
     * Default: 1000 (1s).
     */
    quantizeNewDelayUntilMs?: number;
    /**
     * Whether to read the existing run's `delayUntil` outside of the redlock and
     * short-circuit when the new (quantized) `delayUntil` is not later than the
     * current one. Trailing-mode triggers carrying `updateData` always bypass
     * this fast path and take the lock so payload/metadata/tag updates still
     * land on the run.
     *
     * Default: true.
     */
    fastPathSkipEnabled?: boolean;
    /**
     * Whether to route the unlocked fast-path reads (probe + full-run fetch)
     * through `readOnlyPrisma` (e.g. an Aurora reader) instead of the writer.
     * Safe because debounce is best-effort: a stale `delayUntil` falls
     * through to the locked path (the locked path re-checks under the lock),
     * and a stale `status` at worst returns the existing run, which is the
     * same outcome the caller would see if their trigger had landed a few
     * hundred ms earlier.
     *
     * Default: false.
     */
    useReplicaForFastPathRead?: boolean;
  };
  /** If not set then checkpoints won't ever be used */
  retryWarmStartThresholdMs?: number;
  heartbeatTimeoutsMs?: Partial<HeartbeatTimeouts>;
  repairSnapshotTimeoutMs?: number;
  treatProductionExecutionStallsAsOOM?: boolean;
  suspendedHeartbeatRetriesConfig?: {
    maxCount?: number;
    maxDelayMs?: number;
    initialDelayMs?: number;
    factor?: number;
  };
  queueRunsWaitingForWorkerBatchSize?: number;
  /**
   * Lookup used by {@link PendingVersionSystem} to discover candidate
   * `PENDING_VERSION` run ids without scanning the Postgres status index.
   * Defaults to a noop lookup that returns an empty result, which causes
   * the pending-version resolver to short-circuit. Production callers
   * should inject a ClickHouse-backed implementation.
   */
  pendingVersionRunIdLookup?: PendingVersionRunIdLookup;
  /**
   * When the pending-version lookup returns zero candidates, schedule
   * one more attempt after this delay to cover ClickHouse replication
   * lag. Defaults to 5_000ms.
   */
  pendingVersionLagRetryDelayMs?: number;
  /**
   * Maximum number of lag-aware retries. Each call beyond the first
   * counts against this cap. Defaults to 1 — so a deploy will fire at
   * most two queries spaced by `pendingVersionLagRetryDelayMs`. Set to 0
   * to disable lag-aware retries entirely.
   */
  pendingVersionLagMaxRetries?: number;
  /** Optional maximum TTL for all runs (e.g. "14d"). If set, runs without an explicit TTL
   *  will use this as their TTL, and runs with a TTL larger than this will be clamped. */
  defaultMaxTtl?: string;
  /** When true, `getSnapshotsSince` reads through the read-only replica client instead
   *  of the primary. Defaults to false. Callers passing an explicit `tx` always use
   *  that client regardless of this flag. */
  readReplicaSnapshotsSinceEnabled?: boolean;
  /** Jittered delay bounds for the single replica retry `getSnapshotsSince` performs when
   * the since snapshot is not yet on the replica, before falling back to the primary.
   * Set maxMs to 0 (or any value <= 0) to skip the replica retry and go straight to the primary. */
  readReplicaSnapshotsSinceRetryDelay?: { minMs: number; maxMs: number };
  /**
   * Periodically refreshes the set of worker queues observed by the
   * `runqueue.workerQueue.length` gauge from the WorkerInstanceGroup records, so the
   * gauge reports every active worker queue's length even when this instance is not
   * dequeuing from them (a dequeue is otherwise the only thing that registers a worker
   * queue for observation). When enabled the observer is the source of truth for the
   * observed set, so the per-dequeue registration is skipped. Disabled by default; the
   * server enables it.
   */
  workerQueueObserver?: {
    enabled?: boolean;
    /** How often to refresh the observed worker queue set from the database (ms). Default: 30_000. */
    intervalMs?: number;
    /**
     * Extra suffix variants to also observe for each worker queue, e.g. the scheduled
     * split queue suffix. The suffix value lives with the caller that owns the naming
     * convention rather than in the engine. Default: [].
     */
    additionalQueueSuffixes?: string[];
    /**
     * Worker groups whose `cloudProvider` is in this list are not observed. Groups with
     * no `cloudProvider` are always observed. Matched case-insensitively. Default: [].
     */
    excludedCloudProviders?: string[];
  };
  tracer: Tracer;
  meter?: Meter;
  logger?: Logger;
  logLevel?: LogLevel;
};

export type HeartbeatTimeouts = {
  PENDING_EXECUTING: number;
  PENDING_CANCEL: number;
  EXECUTING: number;
  EXECUTING_WITH_WAITPOINTS: number;
  SUSPENDED: number;
};

export type TriggerParams = {
  number?: number;
  friendlyId: string;
  environment: MinimalAuthenticatedEnvironment;
  idempotencyKey?: string;
  idempotencyKeyExpiresAt?: Date;
  /** The original user-provided idempotency key and scope */
  idempotencyKeyOptions?: { key: string; scope: "run" | "attempt" | "global" };
  taskIdentifier: string;
  payload: string;
  payloadType: string;
  context: any;
  traceContext: TriggerTraceContext;
  traceId: string;
  spanId: string;
  parentSpanId?: string;
  lockedToVersionId?: string;
  taskVersion?: string;
  sdkVersion?: string;
  cliVersion?: string;
  concurrencyKey?: string;
  workerQueue?: string;
  region?: string;
  /** When true, the run queue may push directly to the worker queue if concurrency is available.
   *  Gated per WorkerInstanceGroup (production) or always true (development). */
  enableFastPath?: boolean;
  queue: string;
  lockedQueueId?: string;
  isTest: boolean;
  delayUntil?: Date;
  queuedAt?: Date;
  maxAttempts?: number;
  taskEventStore?: string;
  priorityMs?: number;
  queueTimestamp?: Date;
  ttl?: string;
  tags: string[];
  parentTaskRunId?: string;
  rootTaskRunId?: string;
  replayedFromTaskRunFriendlyId?: string;
  batch?: {
    id: string;
    index: number;
  };
  resumeParentOnCompletion?: boolean;
  depth?: number;
  metadata?: string;
  metadataType?: string;
  seedMetadata?: string;
  seedMetadataType?: string;
  oneTimeUseToken?: string;
  maxDurationInSeconds?: number;
  machine?: MachinePresetName;
  workerId?: string;
  runnerId?: string;
  scheduleId?: string;
  scheduleInstanceId?: string;
  createdAt?: Date;
  bulkActionId?: string;
  planType?: string;
  realtimeStreamsVersion?: string;
  streamBasinName?: string | null;
  debounce?: {
    key: string;
    delay: string;
    mode?: "leading" | "trailing";
    maxDelay?: string;
  };
  annotations?: {
    triggerSource: string;
    triggerAction: string;
    rootTriggerSource: string;
    rootScheduleId?: string;
  };
  /**
   * Called when a run is debounced (existing delayed run found with triggerAndWait).
   * Return spanIdToComplete to enable span closing when the run completes.
   * This allows the webapp to create a trace span for the debounced trigger.
   */
  onDebounced?: (params: {
    existingRun: TaskRun;
    waitpoint: Waitpoint;
    debounceKey: string;
  }) => Promise<string | undefined>;
};

export type EngineWorker = Worker<typeof workerCatalog>;

export type ReportableQueue = {
  name: string;
  concurrencyLimit: number | null;
  type: string;
  paused: boolean;
  friendlyId: string;
};
