import { existsSync } from "node:fs";
import { mkdir, unlink, writeFile } from "node:fs/promises";
import { dirname, join } from "node:path";

import { QdrantClient as QdrantRestClient } from "@qdrant/js-client-rest";
import { v4 as uuid } from "uuid";

import { getQdrantHttpPortEnv, getQdrantUrlEnv } from "../../config/env.js";
import type { AssistantConfig } from "../../config/types.js";
import { getLogger } from "../../util/logger.js";
import { getDataDir } from "../../util/platform.js";

const log = getLogger("qdrant-client");

/**
 * Resolve the Qdrant base URL for this process.
 *
 * Precedence (highest first):
 *   1. `QDRANT_HTTP_PORT` — a locally-spawned Qdrant sidecar on 127.0.0.1:<port>.
 *      Set by the CLI when spawning the daemon with a non-default port
 *      (multi-local instances).
 *   2. `QDRANT_URL` — an external Qdrant instance (K8s sidecar, remote URL).
 *   3. `config.memory.qdrant.url` — static config (defaults to
 *      `http://127.0.0.1:6333`).
 *
 * Every caller that constructs a Qdrant URL should route through this helper
 * so the precedence stays consistent across the daemon, CLI, and any
 * background jobs that re-read the config.
 */
export function resolveQdrantUrl(config: AssistantConfig): string {
  const port = getQdrantHttpPortEnv();
  if (port) {
    return `http://127.0.0.1:${port}`;
  }
  const url = getQdrantUrlEnv();
  if (url) {
    return url;
  }
  return config.memory.qdrant.url;
}

export interface QdrantSparseVector {
  indices: number[];
  values: number[];
}

export interface QdrantClientConfig {
  url: string;
  collection: string;
  vectorSize: number;
  onDisk: boolean;
  quantization: "scalar" | "none";
  embeddingModel?: string;
}

export interface QdrantPointPayload {
  target_type:
    | "segment"
    | "item"
    | "summary"
    | "media"
    | "graph_node"
    | "pkb_file"
    // Plugin-owned semantic index. Never queried by any recall lane; scoped to
    // the owning plugin via the `plugin` field.
    | "plugin_index";
  target_id: string;
  text: string;
  kind?: string;
  subject?: string;
  status?: string;
  importance?: number;
  confidence?: number;
  created_at: number;
  last_seen_at?: number;
  conversation_id?: string;
  message_id?: string;
  modality?: "text" | "image" | "audio" | "video";
  /**
   * Manifest name of the plugin that owns this point. Set on `plugin_index`
   * points so every read/delete can be scoped to the calling plugin and an
   * uninstall can purge the plugin's entire namespace. Absent on host points.
   */
  plugin?: string;
  /**
   * The plugin-facing document id for a `plugin_index` point. The Qdrant
   * `target_id` is namespace-qualified (`<plugin>:<documentId>`) to keep point
   * identity unique across plugins; this field carries the bare id back to the
   * owning plugin on query/get. Absent on host points.
   */
  document_id?: string;
  /**
   * Opaque, plugin-supplied metadata carried alongside a `plugin_index` point
   * (provenance such as a row id, file id, or source tool). Round-tripped
   * verbatim on query/get; never interpreted by the host.
   */
  meta?: Record<string, unknown>;
}

export interface QdrantSearchResult {
  id: string;
  score: number;
  payload: QdrantPointPayload;
}

/**
 * Strip a trailing `:sparse-v<digits>` segment from a sentinel string.
 *
 * Legacy v1 sentinels (pre-decouple) included the sparse encoder version in
 * the embedding-model identity. The suffix is no longer written, but stored
 * sentinels written by older daemons may still carry it — strip it before
 * comparing identities so existing collections aren't unnecessarily rebuilt.
 */
export function stripLegacySparseSuffix(sentinel: string): string {
  return sentinel.replace(/:sparse-v\d+$/, "");
}

/**
 * Marker file written before the v1 collection's destructive dimension/model
 * recreate, cleared by the startup lifecycle hook once the `rebuild_index` job
 * has been enqueued.
 *
 * Closes a data-loss window in {@link VellumQdrantClient.ensureCollection}: the
 * `migrated` signal that tells the lifecycle hook to enqueue `rebuild_index`
 * otherwise lives only in a local variable returned after `deleteCollection` +
 * `createCollection` both succeed. If Qdrant fails between the delete and the
 * return — or the process crashes — that signal is lost: the next startup finds
 * the collection missing, recreates it empty, reports `migrated: false`, and
 * the v1 vectors stay dropped until each row is individually re-touched.
 * Persisting the intent on disk *before* the delete lets it survive crashes and
 * transient failures: every later `ensureCollection` reports `migrated: true`
 * until the hook enqueues the rebuild and clears the sentinel. Mirrors the v2
 * reembed sentinel in `plugins/defaults/memory/v2/qdrant.ts`.
 *
 * Distinct from the in-collection model-identity sentinel point handled by
 * {@link VellumQdrantClient.readSentinel} / `writeSentinel` — that records the
 * embedding model; this records "a rebuild is owed".
 */
const REBUILD_SENTINEL_FILENAME = ".memory-v1-rebuild-required";

function rebuildSentinelPath(): string {
  return join(getDataDir(), REBUILD_SENTINEL_FILENAME);
}

function rebuildSentinelExists(): boolean {
  return existsSync(rebuildSentinelPath());
}

async function writeRebuildSentinel(): Promise<void> {
  const path = rebuildSentinelPath();
  await mkdir(dirname(path), { recursive: true });
  await writeFile(path, "");
}

/**
 * Remove the rebuild sentinel after the startup lifecycle hook has enqueued the
 * `rebuild_index` job. Idempotent — a missing file is not an error.
 */
export async function clearRebuildSentinel(): Promise<void> {
  try {
    await unlink(rebuildSentinelPath());
  } catch (err) {
    if ((err as NodeJS.ErrnoException).code !== "ENOENT") {
      throw err;
    }
  }
}

let _instance: VellumQdrantClient | null = null;

export function getQdrantClient(): VellumQdrantClient {
  if (!_instance) {
    throw new Error(
      "Qdrant client not initialized. Call initQdrantClient() first.",
    );
  }
  return _instance;
}

export function initQdrantClient(
  config: QdrantClientConfig,
): VellumQdrantClient {
  _instance = new VellumQdrantClient(config);
  return _instance;
}

export class VellumQdrantClient {
  private readonly client: QdrantRestClient;
  private readonly collection: string;
  private readonly vectorSize: number;
  private readonly onDisk: boolean;
  private readonly quantization: "scalar" | "none";
  private readonly embeddingModel?: string;
  private collectionReady = false;

  private readonly SENTINEL_ID = "00000000-0000-0000-0000-000000000000";

  constructor(config: QdrantClientConfig) {
    this.client = new QdrantRestClient({
      url: config.url,
      checkCompatibility: false,
    });
    this.collection = config.collection;
    this.vectorSize = config.vectorSize;
    this.onDisk = config.onDisk;
    this.quantization = config.quantization;
    this.embeddingModel = config.embeddingModel;
  }

  async ensureCollection(): Promise<{ migrated: boolean }> {
    if (this.collectionReady) {
      return { migrated: false };
    }

    // A leftover sentinel means a prior boot deleted the collection but never
    // got to enqueue the rebuild (createCollection threw, or the process died
    // before the lifecycle hook ran). Carry that signal forward until the hook
    // clears it, even when the collection now looks freshly created or intact.
    let migrated = rebuildSentinelExists();

    try {
      const exists = await this.client.collectionExists(this.collection);
      if (exists.exists) {
        try {
          const info = await this.client.getCollection(this.collection);
          const vectorsConfig = info.config?.params?.vectors;

          // Detect whether the collection uses unnamed vectors (legacy) vs named vectors
          const isUnnamedVectors =
            vectorsConfig != null &&
            typeof vectorsConfig === "object" &&
            "size" in vectorsConfig;

          const currentSize = isUnnamedVectors
            ? (vectorsConfig as { size?: number })?.size
            : (vectorsConfig as Record<string, { size?: number }> | undefined)
                ?.dense?.size;

          const dimMismatch =
            currentSize != null && currentSize !== this.vectorSize;

          // Check model identity via a sentinel point that stores the embedding model.
          //
          // Legacy sentinels included a ":sparse-v<N>" suffix that conflated the
          // sparse encoder version with the dense model identity. Sparse vectors
          // are upserted in place and never require collection recreation, so a
          // stored value matching the current dense identity after stripping any
          // legacy sparse suffix is treated as compatible. Compatible-but-legacy
          // sentinels are rewritten to the new format below to clean up gradually.
          let modelMismatch = false;
          let needsSentinelRewrite = false;
          if (this.embeddingModel) {
            const sentinel = await this.readSentinel();
            if (sentinel) {
              const normalizedStored = stripLegacySparseSuffix(sentinel);
              const normalizedCurrent = stripLegacySparseSuffix(
                this.embeddingModel,
              );
              if (normalizedStored !== normalizedCurrent) {
                modelMismatch = true;
              } else if (sentinel !== this.embeddingModel) {
                needsSentinelRewrite = true;
              }
            }
          }

          if (isUnnamedVectors) {
            log.warn(
              {
                collection: this.collection,
                currentSize,
                expectedSize: this.vectorSize,
              },
              "Qdrant collection uses unnamed vectors (legacy) — deleting and recreating with named vectors. Embeddings will be re-indexed.",
            );
            await this.deleteCollectionForRebuild();
            migrated = true;
            // Fall through to collection creation below
          } else if (dimMismatch || modelMismatch) {
            // The v1 collection (config.memory.qdrant.collection) is not managed
            // by the startup embedding reconcile — that reconcile owns only the
            // v2 (memory_v2_concept_pages) and v3 (memory_v3_sections)
            // collections. Dimension drift on v1 is therefore repaired here:
            // delete and recreate, and the lifecycle hook re-embeds from the
            // SQLite cache via rebuild_index when ensureCollection reports
            // migrated and v2 is disabled.
            log.warn(
              {
                collection: this.collection,
                currentSize,
                expectedSize: this.vectorSize,
                dimMismatch,
                modelMismatch,
              },
              "Qdrant collection incompatible (dimension/model change) — deleting and recreating. Embeddings will be regenerated on demand.",
            );
            await this.deleteCollectionForRebuild();
            migrated = true;
            // Fall through to collection creation below
          } else {
            await this.reconcileSparseIndexOnDisk(info);
            if (await this.ensurePayloadIndexesSafe()) {
              this.collectionReady = true;
            }
            if (needsSentinelRewrite && this.embeddingModel) {
              await this.writeSentinel(this.embeddingModel);
            }
            // `migrated` is false in steady state; it is only true here when a
            // prior recreate wrote the sentinel but never cleared it, in which
            // case the rebuild is still owed against this (possibly empty)
            // collection.
            return { migrated };
          }
        } catch (err) {
          log.warn(
            { err },
            "Failed to verify collection compatibility, assuming compatible",
          );
          if (await this.ensurePayloadIndexesSafe()) {
            this.collectionReady = true;
          }
          return { migrated };
        }
      }
    } catch {
      // Collection doesn't exist, create it
    }

    log.info(
      { collection: this.collection, vectorSize: this.vectorSize },
      "Creating Qdrant collection with named vectors (dense + sparse)",
    );

    try {
      await this.client.createCollection(this.collection, {
        vectors: {
          dense: {
            size: this.vectorSize,
            distance: "Cosine",
            on_disk: this.onDisk,
          },
        },
        sparse_vectors: {
          // The sparse inverted index lives in RAM unless placed on disk
          // explicitly; it grows with every indexed point, so it follows the
          // same on-disk setting as the dense vectors and payloads.
          sparse: { index: { on_disk: this.onDisk } },
        },
        hnsw_config: {
          on_disk: this.onDisk,
          m: 16,
          ef_construct: 100,
        },
        optimizers_config: {
          default_segment_number: 2,
        },
        quantization_config:
          this.quantization === "scalar"
            ? {
                scalar: {
                  type: "int8",
                  quantile: 0.99,
                  always_ram: true,
                },
              }
            : undefined,
        on_disk_payload: this.onDisk,
      });
    } catch (err) {
      // 409 = collection was created by a concurrent caller — that's fine
      if (
        err instanceof Error &&
        "status" in err &&
        (err as { status: number }).status === 409
      ) {
        if (await this.ensurePayloadIndexesSafe()) {
          this.collectionReady = true;
        }
        return { migrated };
      }
      throw err;
    }

    if (await this.ensurePayloadIndexesSafe()) {
      // Write sentinel point to record the active embedding model
      if (this.embeddingModel) {
        await this.writeSentinel(this.embeddingModel);
      }

      this.collectionReady = true;
      log.info(
        { collection: this.collection },
        "Qdrant collection created with payload indexes",
      );
    }

    return { migrated };
  }

  /**
   * Delete the collection as the first half of a destructive dimension/model
   * recreate. Persists the rebuild sentinel BEFORE the delete so a crash or
   * transient Qdrant failure between the delete and the recreate still triggers
   * `rebuild_index` on the next ensure call. Callers set `migrated = true` and
   * fall through to recreation.
   */
  private async deleteCollectionForRebuild(): Promise<void> {
    await writeRebuildSentinel();
    await this.client.deleteCollection(this.collection);
  }

  /**
   * Align an existing collection's sparse-index placement with the configured
   * `onDisk` flag. Qdrant keeps the sparse inverted index in RAM unless the
   * collection explicitly opts into on-disk, and collections keep whatever
   * they were created with — so long-lived installs hold the whole index in
   * RAM even when `onDisk` is set. `updateCollection` moves it in place (the
   * optimizer rewrites segments in the background) without a reembed.
   *
   * Best-effort: a failed update logs and leaves the collection serving from
   * its current index — search keeps working either way.
   */
  private async reconcileSparseIndexOnDisk(
    info: Awaited<ReturnType<QdrantRestClient["getCollection"]>>,
  ): Promise<void> {
    const sparseParams = (
      info.config?.params as
        | {
            sparse_vectors?: Record<
              string,
              { index?: { on_disk?: boolean | null } | null } | undefined
            >;
          }
        | undefined
    )?.sparse_vectors;
    // No sparse config in the probe — nothing to move. (A collection without
    // the `sparse` named vector cannot accept an index update for it.)
    if (!sparseParams || !("sparse" in sparseParams)) {
      return;
    }

    const current = sparseParams.sparse?.index?.on_disk ?? false;
    if (current === this.onDisk) {
      return;
    }

    try {
      await this.client.updateCollection(this.collection, {
        sparse_vectors: { sparse: { index: { on_disk: this.onDisk } } },
      });
      log.info(
        { collection: this.collection, onDisk: this.onDisk },
        "Moved sparse index placement on existing Qdrant collection",
      );
    } catch (err) {
      log.warn(
        { err, collection: this.collection, onDisk: this.onDisk },
        "Failed to update sparse index placement — continuing with current index",
      );
    }
  }

  async upsert(
    targetType: QdrantPointPayload["target_type"],
    targetId: string,
    vector: number[],
    payload: Omit<QdrantPointPayload, "target_type" | "target_id">,
    sparseVector?: QdrantSparseVector,
  ): Promise<string> {
    await this.ensureCollection();

    // Deterministic point ID: look up existing point by target_type + target_id
    const existing = await this.findByTarget(targetType, targetId);
    const pointId = existing ?? uuid();

    const namedVector: Record<
      string,
      number[] | { indices: number[]; values: number[] }
    > = {
      dense: vector,
    };
    if (sparseVector) {
      namedVector.sparse = {
        indices: sparseVector.indices,
        values: sparseVector.values,
      };
    }

    const point = {
      id: pointId,
      vector: namedVector,
      payload: {
        target_type: targetType,
        target_id: targetId,
        ...payload,
      },
    };

    try {
      await this.client.upsert(this.collection, {
        wait: true,
        points: [point],
      });
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        await this.client.upsert(this.collection, {
          wait: true,
          points: [point],
        });
      } else {
        throw err;
      }
    }

    return pointId;
  }

  async search(
    vector: number[],
    limit: number,
    filter?: Record<string, unknown>,
  ): Promise<QdrantSearchResult[]> {
    await this.ensureCollection();

    const searchParams = {
      vector: { name: "dense", vector },
      limit,
      with_payload: true,
      score_threshold: 0.0,
      filter: filter as Parameters<QdrantRestClient["search"]>[1]["filter"],
    };

    let results;
    try {
      results = await this.client.search(this.collection, searchParams);
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        results = await this.client.search(this.collection, searchParams);
      } else {
        throw err;
      }
    }

    return results.map((result) => ({
      id: typeof result.id === "string" ? result.id : String(result.id),
      score: result.score,
      payload: result.payload as unknown as QdrantPointPayload,
    }));
  }

  async searchWithFilter(
    vector: number[],
    limit: number,
    targetTypes: Array<"segment" | "item" | "summary" | "media">,
    excludeMessageIds?: string[],
  ): Promise<QdrantSearchResult[]> {
    const mustConditions: Array<Record<string, unknown>> = [
      {
        key: "target_type",
        match: { any: targetTypes },
      },
    ];

    if (excludeMessageIds && excludeMessageIds.length > 0) {
      // Only require status=active for items; segments and summaries don't have a status field
      mustConditions.push({
        should: [
          {
            must: [
              { key: "target_type", match: { value: "item" } },
              { key: "status", match: { value: "active" } },
            ],
          },
          {
            key: "target_type",
            match: { any: ["segment", "summary", "media"] },
          },
        ],
      });
    }

    const mustNotConditions: Array<Record<string, unknown>> = [
      { key: "_meta", match: { value: true } },
    ];
    if (excludeMessageIds && excludeMessageIds.length > 0) {
      mustNotConditions.push({
        key: "message_id",
        match: { any: excludeMessageIds },
      });
    }

    const filter: Record<string, unknown> = {
      must: mustConditions,
      must_not: mustNotConditions,
    };

    return this.search(vector, limit, filter);
  }

  /**
   * Hybrid search using both dense and sparse vectors with RRF fusion.
   * Performs two prefetch queries (dense + sparse) and fuses results
   * using Reciprocal Rank Fusion via Qdrant's query API.
   */
  async hybridSearch(params: {
    denseVector: number[];
    sparseVector: QdrantSparseVector;
    filter?: object;
    limit: number;
    prefetchLimit?: number;
  }): Promise<QdrantSearchResult[]> {
    await this.ensureCollection();

    const { denseVector, sparseVector, filter, limit, prefetchLimit } = params;
    const effectivePrefetchLimit = prefetchLimit ?? 40;

    const queryParams = {
      prefetch: [
        {
          query: denseVector as unknown as number[],
          using: "dense",
          limit: effectivePrefetchLimit,
        },
        {
          query: {
            indices: sparseVector.indices,
            values: sparseVector.values,
          },
          using: "sparse",
          limit: effectivePrefetchLimit,
        },
      ],
      query: { fusion: "rrf" as const },
      limit,
      filter: filter as Record<string, unknown> | undefined,
      with_payload: true,
    };

    let results;
    try {
      results = await this.client.query(this.collection, queryParams);
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        results = await this.client.query(this.collection, queryParams);
      } else {
        throw err;
      }
    }

    return (results.points ?? []).map((point) => ({
      id: typeof point.id === "string" ? point.id : String(point.id),
      score: point.score ?? 0,
      payload: point.payload as unknown as QdrantPointPayload,
    }));
  }

  async deleteByTarget(targetType: string, targetId: string): Promise<void> {
    await this.ensureCollection();

    const doDelete = () =>
      this.client.delete(this.collection, {
        wait: true,
        filter: {
          must: [
            { key: "target_type", match: { value: targetType } },
            { key: "target_id", match: { value: targetId } },
          ],
        },
      });

    try {
      await doDelete();
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        await doDelete();
      } else {
        throw err;
      }
    }
  }

  /**
   * Delete all vectors matching a given target_type (bulk cleanup).
   */
  async deleteByTargetType(targetType: string): Promise<void> {
    await this.ensureCollection();

    const doDelete = () =>
      this.client.delete(this.collection, {
        wait: true,
        filter: {
          must: [{ key: "target_type", match: { value: targetType } }],
        },
      });

    try {
      await doDelete();
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        await doDelete();
      } else {
        throw err;
      }
    }
  }

  /**
   * Delete all vectors matching target_type and payload path. Used to remove
   * every chunk belonging to a single PKB file.
   */
  async deleteByTargetTypeAndPath(
    targetType: string,
    path: string,
  ): Promise<void> {
    await this.ensureCollection();

    const doDelete = () =>
      this.client.delete(this.collection, {
        wait: true,
        filter: {
          must: [
            { key: "target_type", match: { value: targetType } },
            { key: "path", match: { value: path } },
          ],
        },
      });

    try {
      await doDelete();
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        await doDelete();
      } else {
        throw err;
      }
    }
  }

  /**
   * Fetch a single point by (target_type, target_id) with its payload,
   * optionally requiring a matching `plugin` scope. Returns null when no such
   * point exists (or it belongs to a different plugin). Used by the plugin
   * index facade to implement `getPluginDocument` without exposing raw scroll.
   */
  async getByTarget(
    targetType: string,
    targetId: string,
    opts?: { plugin?: string },
  ): Promise<{ id: string; payload: QdrantPointPayload } | null> {
    await this.ensureCollection();

    const must: Array<Record<string, unknown>> = [
      { key: "target_type", match: { value: targetType } },
      { key: "target_id", match: { value: targetId } },
    ];
    if (opts?.plugin !== undefined) {
      must.push({ key: "plugin", match: { value: opts.plugin } });
    }
    const filter = {
      must,
      must_not: [{ key: "_meta", match: { value: true } }],
    };

    const doScroll = () =>
      this.client.scroll(this.collection, {
        filter,
        limit: 1,
        with_payload: true,
        with_vector: false,
      });

    let result;
    try {
      result = await doScroll();
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        result = await doScroll();
      } else {
        throw err;
      }
    }

    const point = result.points[0];
    if (!point) {
      return null;
    }
    return {
      id: typeof point.id === "string" ? point.id : String(point.id),
      payload: point.payload as unknown as QdrantPointPayload,
    };
  }

  /**
   * Delete a single plugin-owned point, scoped to (target_type, target_id,
   * plugin). The `plugin` predicate guarantees one plugin can never remove
   * another plugin's point even if it guesses the target id.
   */
  async deleteByTargetAndPlugin(
    targetType: string,
    targetId: string,
    plugin: string,
  ): Promise<void> {
    await this.ensureCollection();

    const doDelete = () =>
      this.client.delete(this.collection, {
        wait: true,
        filter: {
          must: [
            { key: "target_type", match: { value: targetType } },
            { key: "target_id", match: { value: targetId } },
            { key: "plugin", match: { value: plugin } },
          ],
        },
      });

    try {
      await doDelete();
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        await doDelete();
      } else {
        throw err;
      }
    }
  }

  /**
   * Delete every point owned by a plugin (all `plugin` matches, any
   * target_type). Backs `purgePluginEmbeddings` on plugin uninstall so no
   * orphaned vectors survive removal of the plugin directory.
   */
  async deleteByPlugin(plugin: string): Promise<void> {
    await this.ensureCollection();

    const doDelete = () =>
      this.client.delete(this.collection, {
        wait: true,
        filter: {
          must: [{ key: "plugin", match: { value: plugin } }],
        },
      });

    try {
      await doDelete();
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        await doDelete();
      } else {
        throw err;
      }
    }
  }

  async count(): Promise<number> {
    await this.ensureCollection();

    try {
      const result = await this.client.count(this.collection, { exact: false });
      return result.count;
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        const result = await this.client.count(this.collection, {
          exact: false,
        });
        return result.count;
      }
      throw err;
    }
  }

  async deleteCollection(): Promise<boolean> {
    try {
      const exists = await this.client.collectionExists(this.collection);
      if (!exists.exists) {
        return false;
      }
      await this.client.deleteCollection(this.collection);
      this.collectionReady = false;
      return true;
    } catch (err) {
      log.warn(
        { err, collection: this.collection },
        "Failed to delete Qdrant collection",
      );
      return false;
    }
  }

  /**
   * Detect "collection not found" errors from Qdrant so callers can
   * reset collectionReady and retry after an external deletion
   * (e.g. `assistant sessions clear`).
   */
  private isCollectionMissing(err: unknown): boolean {
    if (
      err &&
      typeof err === "object" &&
      "status" in err &&
      (err as { status: number }).status === 404
    ) {
      return true;
    }
    const msg = err instanceof Error ? err.message : String(err);
    return (
      msg.includes("Not found") ||
      msg.includes("doesn't exist") ||
      msg.includes("not found")
    );
  }

  /**
   * Wraps ensurePayloadIndexes so that a 404 (collection deleted between
   * our existence check and index creation) resets collectionReady instead
   * of propagating — the next operation will self-heal via ensureCollection.
   */
  private async ensurePayloadIndexesSafe(): Promise<boolean> {
    try {
      await this.ensurePayloadIndexes();
      return true;
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        return false;
      }
      throw err;
    }
  }

  private async ensurePayloadIndexes(): Promise<void> {
    await Promise.all([
      this.client.createPayloadIndex(this.collection, {
        field_name: "target_type",
        field_schema: "keyword",
      }),
      this.client.createPayloadIndex(this.collection, {
        field_name: "target_id",
        field_schema: "keyword",
      }),
      this.client.createPayloadIndex(this.collection, {
        field_name: "kind",
        field_schema: "keyword",
      }),
      this.client.createPayloadIndex(this.collection, {
        field_name: "status",
        field_schema: "keyword",
      }),
      this.client.createPayloadIndex(this.collection, {
        field_name: "created_at",
        field_schema: "integer",
      }),
      this.client.createPayloadIndex(this.collection, {
        field_name: "conversation_id",
        field_schema: "keyword",
      }),
      this.client.createPayloadIndex(this.collection, {
        field_name: "modality",
        field_schema: "keyword",
      }),
      this.client.createPayloadIndex(this.collection, {
        field_name: "path",
        field_schema: "keyword",
      }),
      this.client.createPayloadIndex(this.collection, {
        field_name: "plugin",
        field_schema: "keyword",
      }),
    ]);
  }

  private async readSentinel(): Promise<string | null> {
    try {
      const points = await this.client.retrieve(this.collection, {
        ids: [this.SENTINEL_ID],
        with_payload: true,
        with_vector: false,
      });
      if (points.length === 0) {
        return null;
      }
      return (
        ((points[0].payload as Record<string, unknown>)
          ?.embedding_model as string) ?? null
      );
    } catch {
      return null;
    }
  }

  private async writeSentinel(model: string): Promise<void> {
    await this.client.upsert(this.collection, {
      wait: true,
      points: [
        {
          id: this.SENTINEL_ID,
          vector: {
            dense: new Array(this.vectorSize).fill(0), // zero vector, never matched in search
          },
          payload: { _meta: true, embedding_model: model },
        },
      ],
    });
  }

  /**
   * Scroll over every point with a given target_type and return the requested
   * payload fields. Handles Qdrant's cursor-based pagination internally.
   *
   * Used by startup reconciliation (e.g. PKB filesystem diffing) to enumerate
   * all indexed points for a target type without loading vectors.
   */
  async scrollByTargetType(
    targetType: string,
    options?: {
      path?: string;
      batchSize?: number;
    },
  ): Promise<Array<{ id: string; payload: Record<string, unknown> }>> {
    await this.ensureCollection();

    const batchSize = options?.batchSize ?? 256;
    const must: Array<Record<string, unknown>> = [
      { key: "target_type", match: { value: targetType } },
    ];
    if (options?.path) {
      must.push({ key: "path", match: { value: options.path } });
    }
    const filter = {
      must,
      must_not: [{ key: "_meta", match: { value: true } }],
    };

    // Guard against a pathological pagination loop.
    const maxIterations = 10_000;
    const doScroll = async () => {
      const out: Array<{ id: string; payload: Record<string, unknown> }> = [];
      let offset: string | number | undefined = undefined;
      for (let i = 0; i < maxIterations; i++) {
        const result = await this.client.scroll(this.collection, {
          filter,
          limit: batchSize,
          with_payload: true,
          with_vector: false,
          ...(offset !== undefined ? { offset } : {}),
        });
        for (const point of result.points) {
          const id = typeof point.id === "string" ? point.id : String(point.id);
          const payload = (point.payload ?? {}) as Record<string, unknown>;
          out.push({ id, payload });
        }
        const next = result.next_page_offset;
        if (next == null) {
          break;
        }
        offset = typeof next === "string" ? next : (next as number);
      }
      return out;
    };

    try {
      return await doScroll();
    } catch (err) {
      if (this.isCollectionMissing(err)) {
        this.collectionReady = false;
        await this.ensureCollection();
        return await doScroll();
      }
      throw err;
    }
  }

  private async findByTarget(
    targetType: string,
    targetId: string,
  ): Promise<string | null> {
    try {
      const results = await this.client.scroll(this.collection, {
        filter: {
          must: [
            { key: "target_type", match: { value: targetType } },
            { key: "target_id", match: { value: targetId } },
          ],
        },
        limit: 1,
        with_payload: false,
        with_vector: false,
      });
      if (results.points.length > 0) {
        const id = results.points[0].id;
        return typeof id === "string" ? id : String(id);
      }
    } catch {
      // Not found
    }
    return null;
  }
}