import { randomUUID } from "crypto";
import { env as stdEnv } from "std-env";
import { z } from "zod";
import { AdditionalEnvVars, BoolEnv } from "./envUtil.js";

export const Env = z
  .object({
    // This will come from `spec.nodeName` in k8s
    TRIGGER_WORKER_INSTANCE_NAME: z.string().default(randomUUID()),
    TRIGGER_WORKER_HEARTBEAT_INTERVAL_SECONDS: z.coerce.number().default(30),

    // Opt-in, dev-only: stream this process's logs over a local telnet/TCP socket on this port.
    SUPERVISOR_TELNET_LOGS_PORT: z.coerce.number().optional(),

    // Required settings
    TRIGGER_API_URL: z.string().url(),
    TRIGGER_WORKER_TOKEN: z.string(), // accepts file:// path to read from a file
    MANAGED_WORKER_SECRET: z.string(),
    OTEL_EXPORTER_OTLP_ENDPOINT: z.string().url(), // set on the runners

    // Workload API settings (coordinator mode) - the workload API is what the run controller connects to
    TRIGGER_WORKLOAD_API_ENABLED: BoolEnv.default(true),
    TRIGGER_WORKLOAD_API_PROTOCOL: z
      .string()
      .transform((s) => z.enum(["http", "https"]).parse(s.toLowerCase()))
      .default("http"),
    TRIGGER_WORKLOAD_API_DOMAIN: z.string().optional(), // If unset, will use orchestrator-specific default
    TRIGGER_WORKLOAD_API_HOST_INTERNAL: z.string().default("0.0.0.0"),
    TRIGGER_WORKLOAD_API_PORT_INTERNAL: z.coerce.number().default(8020), // This is the port the workload API listens on
    TRIGGER_WORKLOAD_API_PORT_EXTERNAL: z.coerce.number().default(8020), // This is the exposed port passed to the run controller

    // Runner settings
    RUNNER_HEARTBEAT_INTERVAL_SECONDS: z.coerce.number().optional(),
    RUNNER_SNAPSHOT_POLL_INTERVAL_SECONDS: z.coerce.number().optional(),
    RUNNER_ADDITIONAL_ENV_VARS: AdditionalEnvVars, // optional (csv)
    RUNNER_PRETTY_LOGS: BoolEnv.default(false),

    // Dequeue settings (provider mode)
    TRIGGER_DEQUEUE_ENABLED: BoolEnv.default(true),
    // Which worker-queue class this supervisor fleet serves. "default" pulls the
    // region queue (standard/agent runs); "scheduled" pulls the dedicated
    // scheduled-lineage queue. Run a separate fleet per class for isolation.
    TRIGGER_WORKER_QUEUE_CLASS: z.enum(["default", "scheduled"]).default("default"),
    TRIGGER_DEQUEUE_INTERVAL_MS: z.coerce.number().int().default(250),
    TRIGGER_DEQUEUE_IDLE_INTERVAL_MS: z.coerce.number().int().default(1000),
    TRIGGER_DEQUEUE_MAX_RUN_COUNT: z.coerce.number().int().default(1),
    TRIGGER_DEQUEUE_MIN_CONSUMER_COUNT: z.coerce.number().int().default(1),
    TRIGGER_DEQUEUE_MAX_CONSUMER_COUNT: z.coerce.number().int().default(10),
    TRIGGER_DEQUEUE_SCALING_STRATEGY: z.enum(["none", "smooth", "aggressive"]).default("none"),
    TRIGGER_DEQUEUE_SCALING_UP_COOLDOWN_MS: z.coerce.number().int().default(5000), // 5 seconds
    TRIGGER_DEQUEUE_SCALING_DOWN_COOLDOWN_MS: z.coerce.number().int().default(30000), // 30 seconds
    TRIGGER_DEQUEUE_SCALING_TARGET_RATIO: z.coerce.number().default(1.0), // Target ratio of queue items to consumers (1.0 = 1 item per consumer)
    TRIGGER_DEQUEUE_SCALING_EWMA_ALPHA: z.coerce.number().min(0).max(1).default(0.3), // Smooths queue length measurements (0=historical, 1=current)
    TRIGGER_DEQUEUE_SCALING_BATCH_WINDOW_MS: z.coerce.number().int().positive().default(1000), // Batch window for metrics processing (ms)
    TRIGGER_DEQUEUE_SCALING_DAMPING_FACTOR: z.coerce.number().min(0).max(1).default(0.7), // Smooths consumer count changes after EWMA (0=no scaling, 1=immediate)

    // Dequeue backpressure - off by default. When enabled, the supervisor reads a
    // verdict from Redis (written by the cluster-side aggregator) and pauses dequeues
    // while the worker cluster can't schedule pods. Disabled = total no-op: no Redis
    // client is created, no reads happen, and the dequeue loop is unaffected.
    TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED: BoolEnv.default(false),
    // Safety default: even when enabled, backpressure only logs what it would do.
    // Set to false to actually skip dequeues / freeze scale-up.
    TRIGGER_DEQUEUE_BACKPRESSURE_DRY_RUN: BoolEnv.default(true),
    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_KEY: z.string().default("engine:dequeue:backpressure"),
    TRIGGER_DEQUEUE_BACKPRESSURE_REFRESH_MS: z.coerce.number().int().positive().default(1000),
    TRIGGER_DEQUEUE_BACKPRESSURE_RAMP_MS: z.coerce.number().int().min(0).default(30_000), // Resume ramp window after release; 0 = instant resume

    TRIGGER_DEQUEUE_BACKPRESSURE_MAX_VERDICT_AGE_MS: z.coerce
      .number()
      .int()
      .positive()
      .default(15_000), // Stale verdict → fail-open (treat as not engaged)
    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST: z.string().optional(),
    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PORT: z.coerce.number().int().optional(),
    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_USERNAME: z.string().optional(),
    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PASSWORD: z.string().optional(),
    TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_TLS_DISABLED: BoolEnv.default(false),
    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENABLED: BoolEnv.default(false),
    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_DRY_RUN: BoolEnv.default(true),
    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE: z.coerce
      .number()
      .int()
      .positive()
      .default(10_000),
    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE: z.coerce
      .number()
      .int()
      .positive()
      .default(5_000),
    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_REFRESH_MS: z.coerce
      .number()
      .int()
      .positive()
      .default(5_000),
    // Hard timeout on the apiserver /metrics scrape. A hung request would otherwise
    // never settle and freeze the monitor's refresh loop (fail-open silently).
    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_SCRAPE_TIMEOUT_MS: z.coerce
      .number()
      .int()
      .positive()
      .default(10_000),

    // Optional services
    TRIGGER_WARM_START_URL: z.string().optional(),
    TRIGGER_CHECKPOINT_URL: z.string().optional(),
    TRIGGER_METADATA_URL: z.string().optional(),

    // Warm-start delivery verification: after a warm-start hit, probe the
    // platform and cold-start the run if no runner acted on the dispatch
    TRIGGER_WARM_START_VERIFY_ENABLED: BoolEnv.default(false),
    TRIGGER_WARM_START_VERIFY_DELAY_MS: z.coerce
      .number()
      .int()
      .min(1_000)
      .max(60_000)
      .default(10_000),

    // Used by the resource monitor
    RESOURCE_MONITOR_ENABLED: BoolEnv.default(false),
    RESOURCE_MONITOR_OVERRIDE_CPU_TOTAL: z.coerce.number().optional(),
    RESOURCE_MONITOR_OVERRIDE_MEMORY_TOTAL_GB: z.coerce.number().optional(),

    // Docker settings
    DOCKER_API_VERSION: z.string().optional(),
    DOCKER_PLATFORM: z.string().optional(), // e.g. linux/amd64, linux/arm64
    DOCKER_STRIP_IMAGE_DIGEST: BoolEnv.default(true),
    DOCKER_REGISTRY_USERNAME: z.string().optional(),
    DOCKER_REGISTRY_PASSWORD: z.string().optional(),
    DOCKER_REGISTRY_URL: z.string().optional(), // e.g. https://index.docker.io/v1
    DOCKER_ENFORCE_MACHINE_PRESETS: BoolEnv.default(true),
    DOCKER_AUTOREMOVE_EXITED_CONTAINERS: BoolEnv.default(true),
    /**
     * Network mode to use for all runners. Supported standard values are: `bridge`, `host`, `none`, and `container:<name|id>`.
     * Any other value is taken as a custom network's name to which all runners should connect to.
     *
     * Accepts a list of comma-separated values to attach to multiple networks. Additional networks are interpreted as network names and will be attached after container creation.
     *
     * **WARNING**: Specifying multiple networks will slightly increase startup times.
     *
     * @default "host"
     */
    DOCKER_RUNNER_NETWORKS: z.string().default("host"),

    // Compute settings
    COMPUTE_GATEWAY_URL: z.string().url().optional(),
    COMPUTE_GATEWAY_AUTH_TOKEN: z.string().optional(),
    COMPUTE_GATEWAY_TIMEOUT_MS: z.coerce.number().int().default(30_000),
    COMPUTE_SNAPSHOTS_ENABLED: BoolEnv.default(false),
    COMPUTE_TRACE_SPANS_ENABLED: BoolEnv.default(true),
    COMPUTE_TRACE_OTLP_ENDPOINT: z.string().url().optional(), // Override for span export (derived from TRIGGER_API_URL if unset)
    COMPUTE_SNAPSHOT_DELAY_MS: z.coerce.number().int().min(0).max(60_000).default(5_000),
    COMPUTE_SNAPSHOT_DISPATCH_LIMIT: z.coerce.number().int().min(1).max(100).default(10),
    // Instance create retries for transient placement failures (1 = no retries)
    COMPUTE_INSTANCE_CREATE_MAX_ATTEMPTS: z.coerce.number().int().min(1).max(10).default(3),
    COMPUTE_INSTANCE_CREATE_RETRY_BASE_DELAY_MS: z.coerce
      .number()
      .int()
      .min(0)
      .max(10_000)
      .default(250),

    // Kubernetes settings
    KUBERNETES_FORCE_ENABLED: BoolEnv.default(false),
    KUBERNETES_NAMESPACE: z.string().default("default"),
    KUBERNETES_WORKER_NODETYPE_LABEL: z.string().default("v4-worker"),
    KUBERNETES_IMAGE_PULL_SECRETS: z.string().optional(), // csv
    KUBERNETES_EPHEMERAL_STORAGE_SIZE_LIMIT: z.string().default("10Gi"),
    KUBERNETES_EPHEMERAL_STORAGE_SIZE_REQUEST: z.string().default("2Gi"),
    KUBERNETES_STRIP_IMAGE_DIGEST: BoolEnv.default(false),
    KUBERNETES_CPU_REQUEST_MIN_CORES: z.coerce.number().min(0).default(0),
    KUBERNETES_CPU_REQUEST_RATIO: z.coerce.number().min(0).max(1).default(0.75), // Ratio of CPU limit, so 0.75 = 75% of CPU limit
    KUBERNETES_MEMORY_REQUEST_MIN_GB: z.coerce.number().min(0).default(0),
    KUBERNETES_MEMORY_REQUEST_RATIO: z.coerce.number().min(0).max(1).default(1), // Ratio of memory limit, so 1 = 100% of memory limit

    // Per-preset overrides of the global KUBERNETES_CPU_REQUEST_RATIO
    KUBERNETES_CPU_REQUEST_RATIO_MICRO: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_CPU_REQUEST_RATIO_SMALL_1X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_CPU_REQUEST_RATIO_SMALL_2X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_CPU_REQUEST_RATIO_MEDIUM_1X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_CPU_REQUEST_RATIO_MEDIUM_2X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_CPU_REQUEST_RATIO_LARGE_1X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_CPU_REQUEST_RATIO_LARGE_2X: z.coerce.number().min(0).max(1).optional(),

    // Per-preset overrides of the global KUBERNETES_MEMORY_REQUEST_RATIO
    KUBERNETES_MEMORY_REQUEST_RATIO_MICRO: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_MEMORY_REQUEST_RATIO_SMALL_1X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_MEMORY_REQUEST_RATIO_SMALL_2X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_MEMORY_REQUEST_RATIO_MEDIUM_1X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_MEMORY_REQUEST_RATIO_MEDIUM_2X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_MEMORY_REQUEST_RATIO_LARGE_1X: z.coerce.number().min(0).max(1).optional(),
    KUBERNETES_MEMORY_REQUEST_RATIO_LARGE_2X: z.coerce.number().min(0).max(1).optional(),

    KUBERNETES_MEMORY_OVERHEAD_GB: z.coerce.number().min(0).optional(), // Optional memory overhead to add to the limit in GB
    KUBERNETES_SCHEDULER_NAME: z.string().optional(), // Custom scheduler name for pods

    // Pod DNS config — override the cluster default ndots to `KUBERNETES_POD_DNS_NDOTS`.
    // Default k8s ndots is 5: any name with fewer than 5 dots (e.g. `api.example.com`, 2 dots) is first walked
    // through every entry in the cluster search list (`<ns>.svc.cluster.local`, `svc.cluster.local`, `cluster.local`)
    // before being tried as-is, turning one resolution into 4+ CoreDNS queries (×2 with A+AAAA).
    // Overriding the default can be useful to cut CoreDNS query amplification for external domains.
    // Note: before enabling, make sure no code path relies on search-list expansion for names with dots ≥ the value
    // set here — those names will now hit their as-is form first and could resolve externally before falling back.
    KUBERNETES_POD_DNS_NDOTS_OVERRIDE_ENABLED: BoolEnv.default(false),
    KUBERNETES_POD_DNS_NDOTS: z.coerce.number().int().min(1).max(15).default(2),
    // Large machine affinity settings - large-* presets prefer a dedicated pool
    KUBERNETES_LARGE_MACHINE_AFFINITY_ENABLED: BoolEnv.default(false),
    KUBERNETES_LARGE_MACHINE_AFFINITY_POOL_LABEL_KEY: z
      .string()
      .trim()
      .min(1)
      .default("node.cluster.x-k8s.io/machinepool"),
    KUBERNETES_LARGE_MACHINE_AFFINITY_POOL_LABEL_VALUE: z
      .string()
      .trim()
      .min(1)
      .default("large-machines"),
    KUBERNETES_LARGE_MACHINE_AFFINITY_WEIGHT: z.coerce.number().int().min(1).max(100).default(100),

    // Project affinity settings - pods from the same project prefer the same node
    KUBERNETES_PROJECT_AFFINITY_ENABLED: BoolEnv.default(false),
    KUBERNETES_PROJECT_AFFINITY_WEIGHT: z.coerce.number().int().min(1).max(100).default(50),
    KUBERNETES_PROJECT_AFFINITY_TOPOLOGY_KEY: z
      .string()
      .trim()
      .min(1)
      .default("kubernetes.io/hostname"),

    // Schedule affinity settings - runs from schedule trees prefer a dedicated pool
    KUBERNETES_SCHEDULED_RUN_AFFINITY_ENABLED: BoolEnv.default(false),
    KUBERNETES_SCHEDULED_RUN_AFFINITY_POOL_LABEL_KEY: z
      .string()
      .trim()
      .min(1)
      .default("node.cluster.x-k8s.io/machinepool"),
    KUBERNETES_SCHEDULED_RUN_AFFINITY_POOL_LABEL_VALUE: z
      .string()
      .trim()
      .min(1)
      .default("scheduled-runs"),
    KUBERNETES_SCHEDULED_RUN_AFFINITY_WEIGHT: z.coerce.number().int().min(1).max(100).default(80),
    KUBERNETES_SCHEDULED_RUN_ANTI_AFFINITY_WEIGHT: z.coerce
      .number()
      .int()
      .min(1)
      .max(100)
      .default(20),

    // Schedule toleration settings - scheduled runs tolerate taints on the dedicated pool
    // Comma-separated list of tolerations in the format: key=value:effect
    // For Exists operator (no value): key:effect
    KUBERNETES_SCHEDULED_RUN_TOLERATIONS: z
      .string()
      .transform((val, ctx) => {
        const tolerations = val
          .split(",")
          .map((entry) => entry.trim())
          .filter((entry) => entry.length > 0)
          .map((entry) => {
            const colonIdx = entry.lastIndexOf(":");
            if (colonIdx === -1) {
              ctx.addIssue({
                code: z.ZodIssueCode.custom,
                message: `Invalid toleration format (missing effect): "${entry}"`,
              });
              return z.NEVER;
            }

            const effect = entry.slice(colonIdx + 1);
            const validEffects = ["NoSchedule", "NoExecute", "PreferNoSchedule"];
            if (!validEffects.includes(effect)) {
              ctx.addIssue({
                code: z.ZodIssueCode.custom,
                message: `Invalid toleration effect "${effect}" in "${entry}". Must be one of: ${validEffects.join(
                  ", "
                )}`,
              });
              return z.NEVER;
            }

            const keyValue = entry.slice(0, colonIdx);
            const eqIdx = keyValue.indexOf("=");
            const key = eqIdx === -1 ? keyValue : keyValue.slice(0, eqIdx);

            if (!key) {
              ctx.addIssue({
                code: z.ZodIssueCode.custom,
                message: `Invalid toleration format (empty key): "${entry}"`,
              });
              return z.NEVER;
            }

            if (eqIdx === -1) {
              return { key, operator: "Exists" as const, effect };
            }

            return {
              key,
              operator: "Equal" as const,
              value: keyValue.slice(eqIdx + 1),
              effect,
            };
          });

        return tolerations;
      })
      .optional(),

    // Placement tags settings
    PLACEMENT_TAGS_ENABLED: BoolEnv.default(false),
    PLACEMENT_TAGS_PREFIX: z.string().default("node.cluster.x-k8s.io"),

    // Metrics
    METRICS_ENABLED: BoolEnv.default(true),
    METRICS_COLLECT_DEFAULTS: BoolEnv.default(true),
    METRICS_HOST: z.string().default("127.0.0.1"),
    METRICS_PORT: z.coerce.number().int().default(9090),

    // Pod cleaner
    POD_CLEANER_ENABLED: BoolEnv.default(true),
    POD_CLEANER_INTERVAL_MS: z.coerce.number().int().default(10000),
    POD_CLEANER_BATCH_SIZE: z.coerce.number().int().default(500),

    // Failed pod handler
    FAILED_POD_HANDLER_ENABLED: BoolEnv.default(true),
    FAILED_POD_HANDLER_RECONNECT_INTERVAL_MS: z.coerce.number().int().default(1000),

    // Debug
    DEBUG: BoolEnv.default(false),
    SEND_RUN_DEBUG_LOGS: BoolEnv.default(false),

    // Wide-event observability - off by default. Emits one flat-keyed JSON
    // line per natural unit of work (dequeue iteration, HTTP request, socket
    // lifecycle). High-QPS hotpath, so the kill switch must be honoured.
    TRIGGER_WIDE_EVENTS_ENABLED: BoolEnv.default(false),
    // When true, also emit wide events for high-frequency HTTP routes
    // (heartbeat, snapshots-since, logs/debug). Off in prod to keep event
    // volume manageable; on in test environments for full-fidelity debugging.
    TRIGGER_WIDE_EVENTS_NOISY_ROUTES: BoolEnv.default(false),
  })
  .superRefine((data, ctx) => {
    if (
      data.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENABLED &&
      data.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE >=
        data.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE
    ) {
      ctx.addIssue({
        code: z.ZodIssueCode.custom,
        message:
          "TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE must be less than TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE",
        path: ["TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE"],
      });
    }
    if (data.COMPUTE_SNAPSHOTS_ENABLED && !data.TRIGGER_METADATA_URL) {
      ctx.addIssue({
        code: z.ZodIssueCode.custom,
        message: "TRIGGER_METADATA_URL is required when COMPUTE_SNAPSHOTS_ENABLED is true",
        path: ["TRIGGER_METADATA_URL"],
      });
    }
    if (data.COMPUTE_SNAPSHOTS_ENABLED && !data.TRIGGER_WORKLOAD_API_DOMAIN) {
      ctx.addIssue({
        code: z.ZodIssueCode.custom,
        message: "TRIGGER_WORKLOAD_API_DOMAIN is required when COMPUTE_SNAPSHOTS_ENABLED is true",
        path: ["TRIGGER_WORKLOAD_API_DOMAIN"],
      });
    }
    if (
      data.TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED &&
      !data.TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST
    ) {
      ctx.addIssue({
        code: z.ZodIssueCode.custom,
        message:
          "TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST is required when TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED is true",
        path: ["TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST"],
      });
    }
  })
  .transform((data) => ({
    ...data,
    COMPUTE_TRACE_OTLP_ENDPOINT: data.COMPUTE_TRACE_OTLP_ENDPOINT ?? `${data.TRIGGER_API_URL}/otel`,
  }));

export const env = Env.parse(stdEnv);
