import { createHash } from "node:crypto";

import { getOllamaBaseUrlEnv } from "../../config/env.js";
import { resolveCallSiteConfig } from "../../config/llm-resolver.js";
import { isMemoryEnabled } from "../../config/memory-v3-gate.js";
import type { AssistantConfig } from "../../config/types.js";
import { PLATFORM_PROVIDER_META } from "../../providers/platform-proxy/constants.js";
import { resolveManagedProxyContext } from "../../providers/platform-proxy/context.js";
import { getProviderKeyAsync } from "../../security/secure-keys.js";
import { getLogger } from "../../util/logger.js";
import {
  EmbeddingBillingBlockError,
  extractHttpStatus,
  isEmbeddingBillingBreakerOpen,
  recordBillingBlock,
  recordBillingSuccess,
} from "./embedding-billing-breaker.js";
import { GeminiEmbeddingBackend } from "./embedding-gemini.js";
import { OllamaEmbeddingBackend } from "./embedding-ollama.js";
import { OpenAIEmbeddingBackend } from "./embedding-openai.js";
import { EmbeddingRuntimeManager } from "./embedding-runtime-manager.js";
import {
  EMBEDDING_DIMENSION_PROBE_TEXT,
  type EmbeddingBackend,
  type EmbeddingInput,
  embeddingInputContentHash,
  type EmbeddingProviderName,
  type EmbeddingRequestOptions,
  type MultimodalEmbeddingInput,
  normalizeEmbeddingInput,
  type SparseEmbedding,
  type TextEmbeddingInput,
} from "./embedding-types.js";
import { SPARSE_VOCAB_SIZE, tokenHash, tokenize } from "./sparse-tokenize.js";

export type { EmbeddingInput, MultimodalEmbeddingInput, TextEmbeddingInput };

const log = getLogger("memory-embeddings");

// Tracks whether the local embedding backend has permanently failed to load
// (e.g., onnxruntime-node missing in a compiled binary). Once set, `auto` mode
// skips `local` as primary, avoiding repeated fallback latency and cost.
let localBackendBroken = false;

/**
 * Set once {@link shutdownEmbeddingBackends} starts, and never cleared: the
 * process is exiting.
 *
 * Emptying the backend cache is not enough on its own. A background turn that
 * was already past its last await can call {@link selectEmbeddingBackend}
 * afterwards, build a fresh backend with its own `disposeRequested` still
 * false, and spawn a worker milliseconds before the process exits, which is
 * precisely the orphan this is all here to prevent.
 */
let embeddingBackendsShutDown = false;

/**
 * Ceiling on {@link shutdownEmbeddingBackends}.
 *
 * The work it bounds is variable: an in-flight-initialization race, then a
 * SIGTERM and SIGKILL wait per backend, then a reclaim sweep that pays the same
 * two waits for every worker it finds. Callers that bound their own shutdown
 * (the daemon's force-exit timer, the memory-worker process) budget against
 * this single number rather than trying to model those parts, and it is
 * enforced here rather than merely documented.
 */
export const EMBEDDING_SHUTDOWN_BUDGET_MS = 8_000;

/**
 * Lazy wrapper around LocalEmbeddingBackend that dynamically imports the
 * module on first use. This avoids eagerly loading @huggingface/transformers
 * (which statically imports onnxruntime-node) at module evaluation time.
 * In compiled binaries where onnxruntime-node isn't bundled, the static
 * import would crash the entire daemon at startup. By deferring the import,
 * the failure is contained and other embedding backends can be used instead.
 */

class LazyLocalEmbeddingBackend implements EmbeddingBackend {
  readonly provider = "local" as const;
  readonly model: string;
  private delegate: EmbeddingBackend | null = null;
  private initPromise: Promise<EmbeddingBackend> | null = null;

  constructor(model: string) {
    this.model = model;
  }

  async embed(
    inputs: EmbeddingInput[],
    options?: EmbeddingRequestOptions,
  ): Promise<number[][]> {
    const backend = await this.getDelegate();
    try {
      return await backend.embed(inputs, options);
    } catch (err) {
      // The onnxruntime-node failure surfaces here during the first embed() call
      // (via LocalEmbeddingBackend.initialize()). Mark broken so auto mode stops
      // selecting local on subsequent requests.
      if (!localBackendBroken && isInitializationError(err)) {
        localBackendBroken = true;
        log.warn(
          { err },
          "Local embedding backend permanently unavailable; auto mode will skip it",
        );
      }
      throw err;
    }
  }

  dispose(): void {
    this.delegate?.dispose?.();
  }

  terminateNow(): void {
    this.delegate?.terminateNow?.();
  }

  async sweepOwnedWorkers(): Promise<void> {
    await this.delegate?.sweepOwnedWorkers?.();
  }

  async shutdown(): Promise<void> {
    // A delegate under construction still ends up owning a worker, so settle
    // the in-flight import before tearing down rather than skipping it.
    if (!this.delegate && this.initPromise) {
      await this.initPromise.catch(() => undefined);
    }
    await this.delegate?.shutdown?.();
  }

  resetForRetry(): void {
    if (!this.delegate) {
      this.initPromise = null;
    }
  }

  private async getDelegate(): Promise<EmbeddingBackend> {
    if (this.delegate) {
      return this.delegate;
    }
    if (!this.initPromise) {
      this.initPromise = (async () => {
        try {
          const { LocalEmbeddingBackend } =
            await import("./embedding-local.js");
          this.delegate = new LocalEmbeddingBackend(this.model);
          return this.delegate;
        } catch (err) {
          localBackendBroken = true;
          log.warn(
            { err },
            "Local embedding backend permanently unavailable; auto mode will skip it",
          );
          throw err;
        }
      })();
    }
    return this.initPromise;
  }
}

/** Detect errors thrown by LocalEmbeddingBackend.initialize() so we can
 *  distinguish permanent init failures from transient embed-time errors. */
function isInitializationError(err: unknown): boolean {
  if (!(err instanceof Error)) {
    return false;
  }
  return err.message.includes("Local embedding backend unavailable");
}

/** Global cache of embedding backend instances, keyed by "provider:model". */
const backendCache = new Map<string, EmbeddingBackend>();

// ── In-memory embedding vector cache ──────────────────────────────
// LRU cache keyed by sha256(provider + model + text) → embedding vector.
// Avoids redundant API calls / local compute for identical content.
// Eviction is based on estimated byte size (32 MB cap) rather than entry count,
// since vector dimensions vary across providers/models.
const VECTOR_CACHE_MAX_BYTES = 33_554_432; // 32 MB
const vectorCache = new Map<string, number[]>();
let vectorCacheBytes = 0;

/** Estimate in-memory byte cost of a single cache entry. */
function estimateEntryBytes(key: string, vector: number[]): number {
  // key: UTF-16 chars (2 bytes each) + vector: 8 bytes per float64
  return key.length * 2 + vector.length * 8;
}

function vectorCacheKey(
  provider: string,
  model: string,
  input: EmbeddingInput,
  extras?: string[],
): string {
  const contentHash = embeddingInputContentHash(input);
  const suffix = extras && extras.length > 0 ? `\0${extras.join("\0")}` : "";
  return createHash("sha256")
    .update(`${provider}\0${model}\0${contentHash}${suffix}`)
    .digest("hex");
}

function getFromVectorCache(
  provider: string,
  model: string,
  input: EmbeddingInput,
  extras?: string[],
): number[] | undefined {
  const key = vectorCacheKey(provider, model, input, extras);
  const v = vectorCache.get(key);
  if (v !== undefined) {
    // LRU refresh: move to end of insertion order
    vectorCache.delete(key);
    vectorCache.set(key, v);
  }
  return v;
}

function putInVectorCache(
  provider: string,
  model: string,
  input: EmbeddingInput,
  vector: number[],
  extras?: string[],
): void {
  const key = vectorCacheKey(provider, model, input, extras);
  // If replacing an existing entry, subtract its old cost first
  const existing = vectorCache.get(key);
  if (existing !== undefined) {
    vectorCacheBytes -= estimateEntryBytes(key, existing);
    vectorCache.delete(key);
  }
  const entryBytes = estimateEntryBytes(key, vector);
  // Evict oldest entries until we have room
  while (
    vectorCacheBytes + entryBytes > VECTOR_CACHE_MAX_BYTES &&
    vectorCache.size > 0
  ) {
    const oldest = vectorCache.keys().next().value;
    if (oldest === undefined) {
      break;
    }
    const oldVec = vectorCache.get(oldest)!;
    vectorCacheBytes -= estimateEntryBytes(oldest, oldVec);
    vectorCache.delete(oldest);
  }
  vectorCache.set(key, vector);
  vectorCacheBytes += entryBytes;
}

/** Clear cached embedding backends and the in-memory vector cache. */
export function clearEmbeddingBackendCache(): void {
  for (const backend of new Set(backendCache.values())) {
    try {
      backend.dispose?.();
    } catch (err) {
      log.warn(
        { err, provider: backend.provider, model: backend.model },
        "Failed to dispose embedding backend during cache clear",
      );
    }
  }
  backendCache.clear();
  vectorCache.clear();
  vectorCacheBytes = 0;
  backendDimCache.clear();
  localBackendBroken = false;
}

/**
 * Tear down every cached backend's OS resources and empty the caches.
 *
 * Called on daemon shutdown so process-owned embedding workers exit with their
 * owner instead of being orphaned. `clearEmbeddingBackendCache()` is the
 * fire-and-forget sibling used on config/credential changes; this one waits for
 * each worker to actually exit (JARVIS-1125).
 */
export async function shutdownEmbeddingBackends(): Promise<void> {
  embeddingBackendsShutDown = true;
  const backends = new Set(backendCache.values());
  backendCache.clear();
  vectorCache.clear();
  vectorCacheBytes = 0;
  backendDimCache.clear();

  const teardown = Promise.all(
    [...backends].map(async (backend) => {
      try {
        await backend.shutdown?.();
        await backend.sweepOwnedWorkers?.();
      } catch (err) {
        log.warn(
          { err, provider: backend.provider, model: backend.model },
          "Failed to shut down embedding backend",
        );
      }
    }),
  );

  const timedOut = Symbol("timeout");
  const outcome = await Promise.race([
    teardown.then(() => undefined),
    Bun.sleep(EMBEDDING_SHUTDOWN_BUDGET_MS).then(() => timedOut),
  ]);
  if (outcome === timedOut) {
    log.warn(
      { budgetMs: EMBEDDING_SHUTDOWN_BUDGET_MS },
      "Embedding backend shutdown exceeded its budget; a worker may outlive this process",
    );
  }
}

/**
 * SIGKILL every cached backend's worker synchronously.
 *
 * For a process that must exit immediately and cannot run the graceful
 * teardown: the memory worker on PID-file eviction, where staying alive to reap
 * would let it keep executing a job its successor has already reclaimed.
 * Without this the child is orphaned only after the successor's single reclaim
 * sweep has passed, leaving two workers alive (JARVIS-1125).
 */
export function terminateEmbeddingWorkersNow(): void {
  for (const backend of new Set(backendCache.values())) {
    try {
      backend.terminateNow?.();
    } catch (err) {
      log.warn(
        { err, provider: backend.provider, model: backend.model },
        "Failed to terminate embedding worker",
      );
    }
  }
}

/** Reset the sticky local-backend failure flag without evicting live backends. */
export function resetLocalEmbeddingFailureState(): void {
  localBackendBroken = false;
  for (const backend of new Set(backendCache.values())) {
    if (backend instanceof LazyLocalEmbeddingBackend) {
      backend.resetForRetry();
    }
  }
}

/**
 * Download the local embedding runtime in the background (non-blocking).
 *
 * Pre-warms the runtime (bun binary + ONNX worker scripts) so the first local
 * embed doesn't pay the download cost inline. A no-op when the runtime is
 * already installed. Failures are swallowed with a warning: local embeddings
 * fall back to cloud backends, so a failed download must never block or crash
 * daemon startup. Callers fire-and-forget; the returned promise settles when
 * the download finishes.
 */
export async function startEmbeddingRuntimeManager(): Promise<void> {
  try {
    const runtimeManager = new EmbeddingRuntimeManager();
    if (runtimeManager.isReady()) {
      return;
    }
    log.info("Downloading embedding runtime in background...");
    await runtimeManager.ensureInstalled();
    // Reset the sticky local-backend failure flag so auto mode retries local
    // embeddings without evicting a worker that may already be live.
    resetLocalEmbeddingFailureState();
    log.info("Embedding runtime download complete");
  } catch (err) {
    log.warn(
      { err },
      "Embedding runtime download failed — local embeddings will use cloud fallback",
    );
  }
}

function cacheKey(provider: string, model: string, extras?: string[]): string {
  if (extras && extras.length > 0) {
    return `${provider}:${model}:${extras.join(":")}`;
  }
  return `${provider}:${model}`;
}

function getCachedOrCreate<T extends EmbeddingBackend>(
  provider: string,
  model: string,
  create: () => T,
  extras?: string[],
): T {
  const key = cacheKey(provider, model, extras);
  const existing = backendCache.get(key);
  if (existing) {
    return existing as T;
  }
  const instance = create();
  backendCache.set(key, instance);
  return instance;
}

/**
 * Look up a previously cached backend instance. Returns undefined when no
 * cached entry exists. Used as a fallback when a provider key lookup
 * returns undefined — a transient credential-store outage should not
 * disable a provider whose backend is already warmed in memory. Explicit
 * key deletion triggers `clearEmbeddingBackendCache()` which empties the
 * cache, so a stale backend is never returned after intentional removal.
 */
function getCached(
  provider: string,
  model: string,
  extras?: string[],
): EmbeddingBackend | undefined {
  return backendCache.get(cacheKey(provider, model, extras));
}

/**
 * The Gemini embedding options that change the output vector for identical
 * input — task type and output dimensionality — rendered as stable cache-key
 * fragments. Empty for a default Gemini config and for every non-Gemini
 * provider. Part of the in-memory vector-cache identity here, and reused by the
 * v3 section dense store so its persistent cache shares the same identity.
 */
export function geminiCacheExtras(config: AssistantConfig): string[] {
  const extras: string[] = [];
  if (config.memory.embeddings.geminiTaskType) {
    extras.push(`task=${config.memory.embeddings.geminiTaskType}`);
  }
  if (config.memory.embeddings.geminiDimensions != null) {
    extras.push(`dim=${config.memory.embeddings.geminiDimensions}`);
  }
  return extras;
}

/** Build (or reuse) the direct-API Gemini backend for the given key. */
function getDirectGeminiBackend(
  config: AssistantConfig,
  geminiKey: string,
): EmbeddingBackend {
  return getCachedOrCreate(
    "gemini",
    config.memory.embeddings.geminiModel,
    () =>
      new GeminiEmbeddingBackend(
        geminiKey,
        config.memory.embeddings.geminiModel,
        {
          taskType: config.memory.embeddings.geminiTaskType,
          dimensions: config.memory.embeddings.geminiDimensions,
        },
      ),
    geminiCacheExtras(config),
  );
}

/**
 * Build (or reuse) the managed-proxy Gemini backend, or return undefined
 * when the managed proxy prerequisites (platform URL + assistant API key)
 * are not satisfied.
 */
async function tryGetManagedGeminiBackend(
  config: AssistantConfig,
): Promise<EmbeddingBackend | undefined> {
  const proxyCtx = await resolveManagedProxyContext();
  const meta = PLATFORM_PROVIDER_META["gemini"];
  if (!proxyCtx.enabled || !meta?.managed || !meta.proxyPath) {
    return undefined;
  }
  const managedBaseUrl = `${proxyCtx.platformBaseUrl}${meta.proxyPath}`;
  return getCachedOrCreate(
    "gemini",
    config.memory.embeddings.geminiModel,
    () =>
      new GeminiEmbeddingBackend(
        proxyCtx.assistantApiKey,
        config.memory.embeddings.geminiModel,
        {
          taskType: config.memory.embeddings.geminiTaskType,
          dimensions: config.memory.embeddings.geminiDimensions ?? 3072,
          managedBaseUrl,
        },
      ),
    [...geminiCacheExtras(config), "managed"],
  );
}

export interface EmbeddingBackendSelection {
  backend: EmbeddingBackend | null;
  reason: string | null;
}

export async function selectEmbeddingBackend(
  config: AssistantConfig,
): Promise<EmbeddingBackendSelection> {
  if (embeddingBackendsShutDown) {
    return { backend: null, reason: "Embedding backends are shutting down" };
  }
  const requested = config.memory.embeddings.provider;
  if (requested === "local") {
    return {
      backend: getCachedOrCreate(
        "local",
        config.memory.embeddings.localModel,
        () =>
          new LazyLocalEmbeddingBackend(config.memory.embeddings.localModel),
      ),
      reason: null,
    };
  }
  if (requested === "ollama") {
    const ollamaKey = (await getProviderKeyAsync("ollama")) ?? undefined;
    return {
      backend: getCachedOrCreate(
        "ollama",
        config.memory.embeddings.ollamaModel,
        () =>
          new OllamaEmbeddingBackend(config.memory.embeddings.ollamaModel, {
            apiKey: ollamaKey,
          }),
      ),
      reason: null,
    };
  }

  // When managed proxy prerequisites are satisfied, insert managed-proxy Gemini
  // at the front of the auto chain so platform assistants use Vellum-managed
  // Gemini embeddings.
  if (requested === "auto" || requested === "gemini") {
    const managed = await tryGetManagedGeminiBackend(config);
    if (managed) {
      return { backend: managed, reason: null };
    }
  }

  // Auto order: local → openai → gemini → ollama
  const order: EmbeddingProviderName[] =
    requested === "auto"
      ? ["local", "openai", "gemini", "ollama"]
      : [requested];

  for (const provider of order) {
    switch (provider) {
      case "local":
        if (localBackendBroken) {
          continue;
        }
        return {
          backend: getCachedOrCreate(
            "local",
            config.memory.embeddings.localModel,
            () =>
              new LazyLocalEmbeddingBackend(
                config.memory.embeddings.localModel,
              ),
          ),
          reason: null,
        };
      case "openai": {
        const openaiKey = await getProviderKeyAsync("openai");
        if (!openaiKey) {
          // Preserve cached backend on transient credential-store failures.
          // Explicit key deletion clears the cache via clearEmbeddingBackendCache().
          const cached = getCached(
            "openai",
            config.memory.embeddings.openaiModel,
          );
          if (cached) {
            return { backend: cached, reason: null };
          }
          continue;
        }
        return {
          backend: getCachedOrCreate(
            "openai",
            config.memory.embeddings.openaiModel,
            () =>
              new OpenAIEmbeddingBackend(
                openaiKey,
                config.memory.embeddings.openaiModel,
              ),
          ),
          reason: null,
        };
      }
      case "gemini": {
        const geminiKey = await getProviderKeyAsync("gemini");
        if (!geminiKey) {
          // Check managed cache variant first so a warm managed backend
          // survives transient proxy-context blips, then non-managed.
          const cached =
            getCached("gemini", config.memory.embeddings.geminiModel, [
              ...geminiCacheExtras(config),
              "managed",
            ]) ??
            getCached(
              "gemini",
              config.memory.embeddings.geminiModel,
              geminiCacheExtras(config),
            );
          if (cached) {
            return { backend: cached, reason: null };
          }
          continue;
        }
        return {
          backend: getDirectGeminiBackend(config, geminiKey),
          reason: null,
        };
      }
      case "ollama": {
        if (!(await isOllamaConfigured(config))) {
          continue;
        }
        const ollamaKey = (await getProviderKeyAsync("ollama")) ?? undefined;
        return {
          backend: getCachedOrCreate(
            "ollama",
            config.memory.embeddings.ollamaModel,
            () =>
              new OllamaEmbeddingBackend(config.memory.embeddings.ollamaModel, {
                apiKey: ollamaKey,
              }),
          ),
          reason: null,
        };
      }
    }
  }

  const reason =
    requested === "auto"
      ? "No embedding backend configured"
      : `Embedding backend "${requested}" is not configured`;
  return { backend: null, reason };
}

export async function getMemoryBackendStatus(config: AssistantConfig): Promise<{
  enabled: boolean;
  degraded: boolean;
  provider: EmbeddingProviderName | null;
  model: string | null;
  reason: string | null;
}> {
  if (!isMemoryEnabled(config)) {
    return {
      enabled: false,
      degraded: false,
      provider: null,
      model: null,
      reason: "memory.disabled",
    };
  }
  const selection = await selectEmbeddingBackend(config);
  if (!selection.backend) {
    return {
      enabled: true,
      degraded: config.memory.embeddings.required,
      provider: null,
      model: null,
      reason: selection.reason,
    };
  }
  return {
    enabled: true,
    degraded: false,
    provider: selection.backend.provider,
    model: selection.backend.model,
    reason: null,
  };
}

/**
 * Memoized output dimension per "provider:model". A backend's vector dimension
 * is fixed for the life of a (provider, model) pair, so a single probe answers
 * every subsequent {@link isEmbeddingDimensionAvailable} call without another
 * backend round-trip. Cleared alongside the backend cache so a credential
 * change or explicit reset re-probes the (possibly different) backend.
 */
const backendDimCache = new Map<string, number>();

/**
 * Whether the currently-reachable embedding backend can produce vectors of the
 * committed Qdrant collection dimension (`config.memory.qdrant.vectorSize`).
 *
 * Read lanes call this BEFORE embedding a query for a dense Qdrant search so a
 * known mismatch (e.g. a 3072-dim collection committed to Gemini, but only a
 * 384-dim local backend reachable while Gemini is down) short-circuits to a
 * clean degraded outcome instead of paying for an embed round-trip that
 * {@link embedWithBackend} would only reject on its dimension assertion. The
 * write path keeps that assertion as the correctness backstop.
 *
 * Returns `false` when memory is degraded for any of these reasons:
 *   - no backend is configured/selectable (`getMemoryBackendStatus().degraded`);
 *   - the selected backend is unreachable (its probe `embed` throws);
 *   - the selected backend's output dimension differs from the committed one.
 *
 * The selected backend's output dimension is not statically known from
 * provider/model alone, so it is measured with a fixed-string probe and
 * memoized per (provider, model) — steady-state calls resolve from
 * {@link backendDimCache} without a backend round-trip.
 */
export async function isEmbeddingDimensionAvailable(
  config: AssistantConfig,
): Promise<boolean> {
  const status = await getMemoryBackendStatus(config);
  if (status.degraded || !status.provider) {
    return false;
  }

  const { backend } = await selectEmbeddingBackend(config);
  if (!backend) {
    return false;
  }

  // Probe the exact backend chain `embedWithBackend` would try (primary +
  // auto-mode fallbacks + the managed→direct Gemini fallback) via the shared
  // assembler, so this preflight can never be more pessimistic than the real
  // embed path. Dense recall stays available if ANY backend in the chain
  // produces the committed dimension. `resolveBackendDimension` memoizes per
  // provider:model, so each backend is probed at most once.
  const expected = config.memory.qdrant.vectorSize;
  for (const candidate of await assembleEmbeddingBackends(config, backend)) {
    if ((await resolveBackendDimension(candidate)) === expected) {
      return true;
    }
  }
  return false;
}

/**
 * Resolve a backend's output dimension, probing once and memoizing the result.
 * Returns `null` when the probe fails (backend unreachable) so the caller can
 * treat the lane as degraded.
 *
 * The single source of truth for backend-dimension measurement: the per-query
 * availability check ({@link isEmbeddingDimensionAvailable}) and the startup
 * reconcile probe both resolve through here, so they share one memoization and
 * cannot disagree.
 */
export async function resolveBackendDimension(
  backend: EmbeddingBackend,
): Promise<number | null> {
  const key = `${backend.provider}:${backend.model}`;
  const cached = backendDimCache.get(key);
  if (cached != null) {
    return cached;
  }

  // Respect the embedding billing breaker: once a 402 opens it,
  // `embedWithBackend` fail-fasts, so probing here would only burn provider
  // requests that cannot succeed (failed probes are not cached, so every read
  // lane calling this would re-probe). Treat an open breaker as "unknown".
  if (isEmbeddingBillingBreakerOpen()) {
    return null;
  }

  try {
    const [vector] = await backend.embed([EMBEDDING_DIMENSION_PROBE_TEXT]);
    const dim = vector?.length;
    if (dim == null) {
      return null;
    }
    backendDimCache.set(key, dim);
    return dim;
  } catch (err) {
    // A 402 means billing is depleted — trip the breaker so sibling probes and
    // embeds stop hammering the provider, mirroring `embedWithBackend`.
    if (extractHttpStatus(err) === 402) {
      recordBillingBlock();
    }
    log.warn(
      { err, provider: backend.provider, model: backend.model },
      "Embedding-dimension availability probe failed; treating as degraded",
    );
    return null;
  }
}

/**
 * Thrown by {@link embedWithBackend} when no embedding backend is configured or
 * available. This is a PROCESS-WIDE condition (backend selection is effectively
 * cached for the run), so a caller embedding many items should treat the first
 * occurrence as fatal to the whole batch rather than a per-item failure — see
 * `backfillAllSections`, which aborts on it instead of churning through deletes.
 */
export class EmbeddingBackendUnavailableError extends Error {
  constructor(message = "No memory embedding backend configured") {
    super(message);
    this.name = "EmbeddingBackendUnavailableError";
  }
}

/**
 * Assemble the ordered backend chain {@link embedWithBackend} will try for a
 * given primary selection: the primary, then (in `auto` mode, non-Gemini
 * primary) the configured fallbacks, then a direct-key Gemini fallback when the
 * primary is managed-proxy Gemini. This is the single source of truth for the
 * fallback chain, shared with the dimension-availability preflight
 * ({@link isEmbeddingDimensionAvailable}) so the preflight cannot diverge from
 * what the real embed path attempts.
 */
async function assembleEmbeddingBackends(
  config: AssistantConfig,
  primary: EmbeddingBackend,
): Promise<EmbeddingBackend[]> {
  // In auto mode, fall through to all configured backends (excluding the
  // primary) so e.g. multimodal inputs can reach Gemini even when the primary
  // is local or openai.
  const fallbacks: EmbeddingBackend[] =
    config.memory.embeddings.provider === "auto" &&
    primary.provider !== "gemini"
      ? await selectFallbackBackends(config, primary.provider)
      : [];

  // A managed-proxy Gemini primary can fail at the proxy (e.g. a stale or
  // revoked platform credential) while a valid direct Gemini key sits in the
  // credential store. The provider chain above never includes Gemini when
  // Gemini IS the primary, so without this the direct key would never be tried.
  if (primary instanceof GeminiEmbeddingBackend && primary.managed) {
    const geminiKey = await getProviderKeyAsync("gemini");
    if (geminiKey) {
      fallbacks.push(getDirectGeminiBackend(config, geminiKey));
    }
  }

  return [primary, ...fallbacks];
}

export async function embedWithBackend(
  config: AssistantConfig,
  inputs: EmbeddingInput[],
  options?: EmbeddingRequestOptions,
): Promise<{
  provider: EmbeddingProviderName;
  model: string;
  vectors: number[][];
}> {
  // Fail-fast when the billing breaker is open — avoids burning a network
  // round-trip on every caller (embed lane jobs, activation recompute, etc.).
  if (isEmbeddingBillingBreakerOpen()) {
    throw new EmbeddingBillingBlockError();
  }

  const selection = await selectEmbeddingBackend(config);
  if (!selection.backend) {
    throw new EmbeddingBackendUnavailableError(
      selection.reason ?? "No memory embedding backend configured",
    );
  }

  const expectedDim = config.memory.qdrant.vectorSize;
  const { provider: primaryProvider, model: primaryModel } = selection.backend;

  // ── Compute provider-specific vector cache extras ───────────────
  const vectorExtras =
    primaryProvider === "gemini" ? geminiCacheExtras(config) : undefined;

  // ── In-memory cache check (primary provider only) ──────────────
  const cached: (number[] | null)[] = inputs.map((input) => {
    const v = getFromVectorCache(
      primaryProvider,
      primaryModel,
      input,
      vectorExtras,
    );
    if (v && v.length === expectedDim) {
      return v;
    }
    return null;
  });
  const uncachedIndices: number[] = [];
  for (let i = 0; i < cached.length; i++) {
    if (!cached[i]) {
      uncachedIndices.push(i);
    }
  }
  if (uncachedIndices.length === 0) {
    return {
      provider: primaryProvider,
      model: primaryModel,
      vectors: cached as number[][],
    };
  }

  // ── Embed uncached inputs ───────────────────────────────────────
  const backends = await assembleEmbeddingBackends(config, selection.backend);

  let lastErr: unknown;
  let anyBackendAttempted = false;
  for (const backend of backends) {
    const isPrimary = backend === selection.backend;
    // For the primary backend, only embed uncached inputs and merge with cached.
    // For fallback backends, embed ALL inputs since the cache was keyed to the primary.
    const inputsToEmbed = isPrimary
      ? uncachedIndices.map((i) => inputs[i])
      : inputs;

    // Skip text-only backends for multimodal inputs
    const hasNonText = inputsToEmbed.some(
      (i) =>
        typeof i !== "string" && normalizeEmbeddingInput(i).type !== "text",
    );
    if (backend.provider !== "gemini" && hasNonText) {
      continue;
    }

    try {
      anyBackendAttempted = true;
      const vectors = await backend.embed(inputsToEmbed, options);
      if (vectors.length !== inputsToEmbed.length) {
        throw new Error(
          `Embedding backend returned ${vectors.length} vectors for ${inputsToEmbed.length} inputs`,
        );
      }
      for (const vec of vectors) {
        if (vec.length !== expectedDim) {
          throw new Error(
            `Embedding backend "${backend.provider}" (model ${backend.model}) returned vectors of dimension ${vec.length}, but Qdrant collection expects ${expectedDim}`,
          );
        }
      }

      // Populate cache with freshly embedded vectors
      const backendExtras =
        backend.provider === "gemini" ? geminiCacheExtras(config) : undefined;
      for (let i = 0; i < inputsToEmbed.length; i++) {
        putInVectorCache(
          backend.provider,
          backend.model,
          inputsToEmbed[i],
          vectors[i],
          backendExtras,
        );
      }

      // A successful backend call proves billing is active — close the
      // breaker if it was in the probe window.
      recordBillingSuccess();

      if (isPrimary) {
        const merged = [...cached] as number[][];
        for (let i = 0; i < uncachedIndices.length; i++) {
          merged[uncachedIndices[i]] = vectors[i];
        }
        return {
          provider: backend.provider,
          model: backend.model,
          vectors: merged,
        };
      }
      return { provider: backend.provider, model: backend.model, vectors };
    } catch (err) {
      lastErr = err;
      // If ANY backend in the chain returns 402, trip the billing breaker
      // immediately — fallbacks will hit the same depleted balance.
      if (extractHttpStatus(err) === 402) {
        recordBillingBlock();
        throw err;
      }
      if (backends.length > 1) {
        log.warn(
          { err, provider: backend.provider },
          "Embedding backend failed, trying next",
        );
      }
    }
  }
  if (!anyBackendAttempted) {
    const hasMultimodal = inputs.some(
      (i) =>
        typeof i !== "string" && normalizeEmbeddingInput(i).type !== "text",
    );
    if (hasMultimodal) {
      throw new Error(
        "No available embedding backend supports multimodal inputs. Gemini API key is required for image/audio/video embeddings.",
      );
    }
  }
  throw lastErr;
}

async function selectFallbackBackends(
  config: AssistantConfig,
  exclude: EmbeddingProviderName,
): Promise<EmbeddingBackend[]> {
  const backends: EmbeddingBackend[] = [];
  const order: EmbeddingProviderName[] = ["openai", "gemini", "ollama"];
  for (const provider of order) {
    if (provider === exclude) {
      continue;
    }
    switch (provider) {
      case "openai": {
        const openaiKey = await getProviderKeyAsync("openai");
        if (openaiKey) {
          backends.push(
            getCachedOrCreate(
              "openai",
              config.memory.embeddings.openaiModel,
              () =>
                new OpenAIEmbeddingBackend(
                  openaiKey,
                  config.memory.embeddings.openaiModel,
                ),
            ),
          );
        } else {
          // Preserve cached backend on transient credential-store failures.
          const cached = getCached(
            "openai",
            config.memory.embeddings.openaiModel,
          );
          if (cached) {
            backends.push(cached);
          }
        }
        break;
      }
      case "gemini": {
        const geminiKey = await getProviderKeyAsync("gemini");
        if (geminiKey) {
          backends.push(getDirectGeminiBackend(config, geminiKey));
        } else {
          // Try managed proxy Gemini as fallback when no direct key exists.
          const managed = await tryGetManagedGeminiBackend(config);
          if (managed) {
            backends.push(managed);
          } else {
            // Check managed cache variant first, then non-managed, so a warm
            // managed backend survives transient proxy-context blips.
            const cached =
              getCached("gemini", config.memory.embeddings.geminiModel, [
                ...geminiCacheExtras(config),
                "managed",
              ]) ??
              getCached(
                "gemini",
                config.memory.embeddings.geminiModel,
                geminiCacheExtras(config),
              );
            if (cached) {
              backends.push(cached);
            }
          }
        }
        break;
      }
      case "ollama": {
        if (await isOllamaConfigured(config)) {
          const ollamaKey = (await getProviderKeyAsync("ollama")) ?? undefined;
          backends.push(
            getCachedOrCreate(
              "ollama",
              config.memory.embeddings.ollamaModel,
              () =>
                new OllamaEmbeddingBackend(
                  config.memory.embeddings.ollamaModel,
                  {
                    apiKey: ollamaKey,
                  },
                ),
            ),
          );
        }
        break;
      }
    }
  }
  return backends;
}

/**
 * Returns true when the active (primary) embedding backend can handle
 * multimodal inputs (images, audio, video). Today only Gemini supports
 * multimodal.
 *
 * Only returns true when Gemini is the primary selected backend — not when
 * it's merely available as a fallback. Writing multimodal embeddings via a
 * fallback provider while queries go through the primary text backend would
 * mix incompatible vector spaces, making retrieval unreliable.
 */
export async function selectedBackendSupportsMultimodal(
  config: AssistantConfig,
): Promise<boolean> {
  const { backend } = await selectEmbeddingBackend(config);
  if (!backend) {
    return false;
  }
  return backend.provider === "gemini";
}

async function isOllamaConfigured(config: AssistantConfig): Promise<boolean> {
  return (
    resolveCallSiteConfig("mainAgent", config.llm).provider === "ollama" ||
    Boolean(await getProviderKeyAsync("ollama")) ||
    Boolean(getOllamaBaseUrlEnv())
  );
}

// ── TF-IDF sparse embedding ───────────────────────────────────────
// Simple tokenizer + TF-IDF sparse encoder. Produces a SparseEmbedding
// with term indices (hashed to a fixed vocabulary) and TF-IDF weights.
// Can be upgraded to a learned sparse encoder (e.g. SPLADE) later.
// Tokenization primitives (`tokenize`, `tokenHash`, `SPARSE_VOCAB_SIZE`) live
// in `./sparse-tokenize.ts` so the BM25 encoder can share them without
// transitively depending on this module.

/**
 * Bump this version whenever the sparse embedding algorithm changes
 * (e.g. hash function fix, tokenizer change). Now inert metadata — the v1
 * Qdrant sentinel was decoupled from this constant, so a bump no longer
 * forces an automatic rebuild. Operators must explicitly run
 * `assistant memory v2 reembed` to rematerialize the v2 sparse index.
 */
export const SPARSE_EMBEDDING_VERSION = 4;

/**
 * Generate a TF-IDF-based sparse embedding for the given text.
 *
 * Term frequency is computed from the input. IDF is approximated using
 * sub-linear TF weighting (1 + log(tf)) since we don't have a corpus-level
 * document frequency table. This still produces useful sparse vectors for
 * lexical matching via Qdrant's sparse vector support.
 */
export function generateSparseEmbedding(text: string): SparseEmbedding {
  const tokens = tokenize(text);
  if (tokens.length === 0) {
    return { indices: [], values: [] };
  }

  // Count term frequencies per hash bucket
  const tf = new Map<number, number>();
  for (const token of tokens) {
    const idx = tokenHash(token, SPARSE_VOCAB_SIZE);
    tf.set(idx, (tf.get(idx) ?? 0) + 1);
  }

  // Convert to sub-linear TF weights: 1 + log(tf)
  const indices: number[] = [];
  const values: number[] = [];
  for (const [idx, count] of tf) {
    indices.push(idx);
    values.push(1 + Math.log(count));
  }

  // L2-normalize the sparse vector so scores are comparable
  let norm = 0;
  for (const v of values) {
    norm += v * v;
  }
  norm = Math.sqrt(norm);
  if (norm > 0) {
    for (let i = 0; i < values.length; i++) {
      values[i] /= norm;
    }
  }

  return { indices, values };
}