class CatalogManager:
    """
    Handles SQLite metadata and FTS operations.

    This manager is responsible for:
    - Creating and managing SQLite tables (metadata and FTS)
    - Adding, deleting, and removing document metadata
    - Building filter clauses for metadata queries
    - FTS5 full-text search indexing

    Note: Vector operations are handled by UsearchIndex, not CatalogManager.

    Args:
        conn: SQLite database connection
        table_name: Name of the metadata table
        fts_table_name: Name of the full-text search table
    """

    def __init__(
        self,
        conn: sqlite3.Connection,
        table_name: str,
        fts_table_name: str,
        lock: threading.RLock | None = None,
    ):
        # Defense-in-depth: validate table names
        _validate_table_name(table_name)
        _validate_table_name(fts_table_name)

        self.conn = conn
        self._table_name = table_name
        self._fts_table_name = fts_table_name
        self._fts_enabled = False
        self._cluster_table_name = f"{table_name}_clusters"
        self._cluster_table_ready = False
        # Serializes Python-level access to the shared sqlite3.Connection. The
        # connection is opened with check_same_thread=False; SQLite itself is
        # safe under WAL, but Python's `with conn:` transaction context is not
        # — two threads entering it simultaneously interleave their writes
        # under one implicit transaction. The lock prevents that.
        self._lock: threading.RLock = lock if lock is not None else threading.RLock()

    def create_tables(self) -> None:
        """Create metadata and FTS tables if they don't exist."""
        self.conn.execute(
            f"""
            CREATE TABLE IF NOT EXISTS {self._table_name} (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                text TEXT NOT NULL,
                metadata TEXT,
                embedding BLOB,
                parent_id INTEGER REFERENCES {self._table_name}(id) ON DELETE SET NULL
            )
            """
        )
        # Create index for parent_id lookups
        self.conn.execute(
            f"""
            CREATE INDEX IF NOT EXISTS idx_{self._table_name}_parent
            ON {self._table_name}(parent_id)
            WHERE parent_id IS NOT NULL
            """
        )
        # Index on text for find_ids_by_texts / remove_texts which previously
        # full-scanned. Costs disk proportional to total text size; payback
        # is large on collections that frequently look up by text content.
        self.conn.execute(
            f"""
            CREATE INDEX IF NOT EXISTS idx_{self._table_name}_text
            ON {self._table_name}(text)
            """
        )
        # Migrate existing tables that lack columns
        self._ensure_embedding_column()
        self._ensure_parent_id_column()
        self._ensure_fts_table()

    def _ensure_embedding_column(self) -> None:
        """Add embedding column if missing (migration for v2.0.0)."""
        try:
            cursor = self.conn.execute(f"PRAGMA table_info({self._table_name})")
            columns = {row[1] for row in cursor.fetchall()}
            if "embedding" not in columns:
                self.conn.execute(
                    f"ALTER TABLE {self._table_name} ADD COLUMN embedding BLOB"
                )
                _logger.info(
                    "Migrated table %s: added embedding column", self._table_name
                )
        except Exception as e:
            _logger.warning("Could not check/add embedding column: %s", e)

    def _ensure_parent_id_column(self) -> None:
        """Add parent_id column if missing (migration for v2.1.0)."""
        try:
            cursor = self.conn.execute(f"PRAGMA table_info({self._table_name})")
            columns = {row[1] for row in cursor.fetchall()}
            if "parent_id" not in columns:
                self.conn.execute(
                    f"ALTER TABLE {self._table_name} ADD COLUMN parent_id INTEGER "
                    f"REFERENCES {self._table_name}(id) ON DELETE SET NULL"
                )
                # Create index for efficient parent lookups
                self.conn.execute(
                    f"""
                    CREATE INDEX IF NOT EXISTS idx_{self._table_name}_parent
                    ON {self._table_name}(parent_id)
                    WHERE parent_id IS NOT NULL
                    """
                )
                _logger.info(
                    "Migrated table %s: added parent_id column", self._table_name
                )
        except Exception as e:
            _logger.warning("Could not check/add parent_id column: %s", e)

    def _ensure_fts_table(self) -> None:
        """Create FTS5 virtual table for full-text search.

        Retries on transient lock errors but permanently disables FTS
        if the module is unavailable.
        """
        import sqlite3

        for attempt in range(3):
            try:
                self.conn.execute(
                    f"""
                    CREATE VIRTUAL TABLE IF NOT EXISTS {self._fts_table_name}
                    USING fts5(text)
                    """
                )
                self._fts_enabled = True
                return
            except sqlite3.OperationalError as e:
                msg = str(e).lower()
                if "database is locked" in msg and attempt < 2:
                    import time
                    time.sleep(0.1 * (attempt + 1))
                    continue
                _logger.warning("FTS5 not available - keyword search disabled: %s", e)
                self._fts_enabled = False
                return

    @property
    def fts_enabled(self) -> bool:
        """Whether FTS5 is available for keyword search."""
        return self._fts_enabled

    def _upsert_fts_rows(self, ids: Sequence[int], texts: Sequence[str]) -> None:
        """Update FTS index for given document IDs.

        Internal helper. Must be called inside an active transaction
        (``with self._lock, self.conn:``) so the FTS shadow table stays in
        sync with the main table on crash.

        Args:
            ids: Document IDs to update
            texts: Corresponding text content
        """
        if not self._fts_enabled or not ids:
            return
        placeholders = ",".join(["?"] * len(ids))
        self.conn.execute(
            f"DELETE FROM {self._fts_table_name} WHERE rowid IN ({placeholders})",
            tuple(ids),
        )
        rows = list(zip(ids, texts))
        self.conn.executemany(
            f"INSERT INTO {self._fts_table_name}(rowid, text) VALUES (?, ?)", rows
        )

    def _delete_fts_rows(self, ids: Sequence[int]) -> None:
        """Remove documents from FTS index.

        Internal helper. Must be called inside an active transaction so
        the FTS shadow table stays in sync with the main table on crash.

        Args:
            ids: Document IDs to remove
        """
        if not self._fts_enabled or not ids:
            return
        placeholders = ",".join(["?"] * len(ids))
        self.conn.execute(
            f"DELETE FROM {self._fts_table_name} WHERE rowid IN ({placeholders})",
            tuple(ids),
        )

    @retry_on_lock(max_retries=5, base_delay=0.1)
    def add_documents(
        self,
        texts: Sequence[str],
        metadatas: Sequence[dict],
        ids: Sequence[int | None] | None = None,
        embeddings: Sequence[Sequence[float]] | None = None,
        parent_ids: Sequence[int | None] | None = None,
    ) -> list[int]:
        """
        Insert or update document metadata.

        Args:
            texts: Document text content
            metadatas: Metadata dicts for each document
            ids: Optional document IDs for upsert behavior
            embeddings: Optional embedding vectors to store
            parent_ids: Optional parent document IDs for hierarchical relationships

        Returns:
            List of document IDs (rowids)
        """
        if not texts:
            return []

        _logger.debug(
            "Adding %d documents to metadata table",
            len(texts),
            extra={"table": self._table_name},
        )

        import numpy as np

        ids_list = list(ids) if ids else [None] * len(texts)
        parent_ids_list = list(parent_ids) if parent_ids else [None] * len(texts)

        # Convert embeddings to bytes if provided
        embedding_blobs: list[bytes | None] = []
        if embeddings is not None:
            # Batch conversion: single np.array call instead of per-item np.asarray
            emb_matrix = np.asarray(embeddings, dtype=np.float32)
            row_bytes = emb_matrix.tobytes()
            stride = emb_matrix.shape[1] * 4  # float32 = 4 bytes
            embedding_blobs = [
                row_bytes[i * stride : (i + 1) * stride]
                for i in range(emb_matrix.shape[0])
            ]
        else:
            embedding_blobs = [None] * len(texts)

        # Pre-serialize metadata (compact separators saves allocation overhead)
        _dumps = json.dumps
        meta_strs = [_dumps(m, separators=(",", ":")) for m in metadatas]

        # Split into auto-ID and explicit-ID groups so each can use the
        # correct INSERT path:
        #   - Explicit IDs: upsert (ON CONFLICT DO UPDATE) so existing rows
        #     are updated in place. last_insert_rowid is unsafe here because
        #     UPSERTs that hit the UPDATE branch do not advance it, breaking
        #     the prior arithmetic.
        #   - Auto IDs (None): plain INSERT, then RETURNING id to recover the
        #     auto-assigned values exactly. Held under self._lock so the
        #     RETURNING result is uncorrupted by concurrent writers.
        explicit_rows = []
        auto_rows = []
        auto_positions = []
        for idx, (uid, txt, meta_str, emb_blob, pid) in enumerate(
            zip(ids_list, texts, meta_strs, embedding_blobs, parent_ids_list)
        ):
            if uid is None:
                auto_rows.append((txt, meta_str, emb_blob, pid))
                auto_positions.append(idx)
            else:
                explicit_rows.append((uid, txt, meta_str, emb_blob, pid))

        real_ids: list[int] = [-1] * len(ids_list)

        with self._lock, self.conn:
            if explicit_rows:
                self.conn.executemany(
                    f"""
                    INSERT INTO {self._table_name}(id, text, metadata, embedding, parent_id)
                    VALUES (?, ?, ?, ?, ?)
                    ON CONFLICT(id) DO UPDATE SET
                        text=excluded.text,
                        metadata=excluded.metadata,
                        embedding=excluded.embedding,
                        parent_id=excluded.parent_id
                    """,
                    explicit_rows,
                )

            if auto_rows:
                # Use a single multi-VALUES INSERT ... RETURNING id so we
                # recover the auto-assigned IDs in the exact insertion order.
                placeholders = ",".join(["(?, ?, ?, ?)"] * len(auto_rows))
                flat_params = [v for r in auto_rows for v in r]
                cursor = self.conn.execute(
                    f"INSERT INTO {self._table_name}"
                    f"(text, metadata, embedding, parent_id) "
                    f"VALUES {placeholders} RETURNING id",
                    flat_params,
                )
                returned = cursor.fetchall()
                if len(returned) != len(auto_rows):
                    raise RuntimeError(
                        f"INSERT RETURNING id returned {len(returned)} rows, "
                        f"expected {len(auto_rows)}"
                    )
                for pos, row in zip(auto_positions, returned):
                    real_ids[pos] = int(row[0])

            # Fill in explicit IDs by their original position
            explicit_iter = iter(explicit_rows)
            for idx, uid in enumerate(ids_list):
                if uid is not None:
                    real_ids[idx] = int(next(explicit_iter)[0])

            # Defense-in-depth: any leftover -1 sentinel here means an
            # INSERT path partially succeeded — never feed that to FTS as
            # a rowid. This catches both retry-loop interaction with the
            # @retry_on_lock decorator and any future code path that
            # forgets to populate real_ids before the FTS upsert.
            if any(rid < 0 for rid in real_ids):
                raise RuntimeError(
                    "Internal error: add_documents produced an unfilled "
                    "rowid sentinel; refusing to update FTS with -1."
                )

            # Update FTS index
            self._upsert_fts_rows(real_ids, texts)

        _logger.debug("Added %d documents, ids=%s", len(real_ids), real_ids[:5])
        return real_ids

    @retry_on_lock(max_retries=5, base_delay=0.1)
    def delete_by_ids(self, ids: Iterable[int]) -> list[int]:
        """
        Delete documents by their IDs.

        Args:
            ids: Document IDs to delete

        Returns:
            List of IDs that were actually deleted
        """
        ids = list(ids)
        if not ids:
            return []

        _logger.debug("Deleting %d documents", len(ids))

        placeholders = ",".join("?" for _ in ids)
        params = tuple(ids)

        with self._lock, self.conn:
            # Check which IDs actually exist
            existing = self.conn.execute(
                f"SELECT id FROM {self._table_name} WHERE id IN ({placeholders})",
                params,
            ).fetchall()
            existing_ids = [r[0] for r in existing]

            if existing_ids:
                placeholders = ",".join("?" for _ in existing_ids)
                self.conn.execute(
                    f"DELETE FROM {self._table_name} WHERE id IN ({placeholders})",
                    tuple(existing_ids),
                )
                self._delete_fts_rows(existing_ids)

        _logger.debug("Deleted %d documents", len(existing_ids))
        return existing_ids

    def get_documents_by_ids(
        self, ids: Sequence[int]
    ) -> dict[int, tuple[str, dict[str, Any]]]:
        """
        Fetch document text and metadata by IDs.

        Args:
            ids: Document IDs to fetch

        Returns:
            Dict mapping id -> (text, metadata)
        """
        if not ids:
            return {}

        placeholders = ",".join(["?"] * len(ids))
        with self._lock:
            rows = self.conn.execute(
                f"SELECT id, text, metadata FROM {self._table_name} WHERE id IN ({placeholders})",
                tuple(ids),
            ).fetchall()

        result = {}
        for row_id, text, meta_json in rows:
            meta = json.loads(meta_json) if meta_json else {}
            result[row_id] = (text, meta)
        return result

    def list_all_ids(self) -> list[int]:
        """Return every doc id in the table, serialized through ``self._lock``.

        Used by the rebuild-index path so the SELECT runs under the same
        re-entrant lock as concurrent writers, eliminating the bare
        ``self.conn.execute(...)`` that previously relied on caller
        discipline alone.
        """
        with self._lock:
            rows = self.conn.execute(
                f"SELECT id FROM {self._table_name}"
            ).fetchall()
        return [row[0] for row in rows]

    def get_embeddings_by_ids(self, ids: Sequence[int]) -> dict[int, Any]:
        """
        Fetch embeddings by document IDs.

        Args:
            ids: Document IDs to fetch

        Returns:
            Dict mapping id -> numpy array (or None if no embedding stored)
        """
        import numpy as np

        if not ids:
            return {}

        placeholders = ",".join(["?"] * len(ids))
        with self._lock:
            rows = self.conn.execute(
                f"SELECT id, embedding FROM {self._table_name} WHERE id IN ({placeholders})",
                tuple(ids),
            ).fetchall()

        result: dict[int, np.ndarray | None] = {}
        for row_id, emb_blob in rows:
            if emb_blob is not None:
                result[row_id] = np.frombuffer(emb_blob, dtype=np.float32)
            else:
                result[row_id] = None
        return result

    def get_documents_and_embeddings_by_ids(
        self, ids: Sequence[int]
    ) -> dict[int, tuple[str, dict[str, Any], Any]]:
        """Fetch documents with their embeddings in a single query.

        Args:
            ids: Document IDs to fetch

        Returns:
            Dict mapping id -> (text, metadata, embedding_array_or_None)
        """
        import numpy as np

        if not ids:
            return {}

        placeholders = ",".join(["?"] * len(ids))
        with self._lock:
            rows = self.conn.execute(
                f"SELECT id, text, metadata, embedding FROM {self._table_name} WHERE id IN ({placeholders})",
                tuple(ids),
            ).fetchall()

        result: dict[int, tuple[str, dict[str, Any], np.ndarray | None]] = {}
        for row_id, text, meta_json, emb_blob in rows:
            meta = json.loads(meta_json) if meta_json else {}
            emb = np.frombuffer(emb_blob, dtype=np.float32) if emb_blob is not None else None
            result[row_id] = (text, meta, emb)
        return result

    def find_ids_by_texts(
        self,
        texts: Sequence[str],
        *,
        limit: int | None = None,
        offset: int | None = None,
    ) -> list[int]:
        """Find document IDs matching exact text content.

        Args:
            texts: Text strings to search for
            limit: Maximum number of IDs to return (None = all)
            offset: Number of IDs to skip (None = 0)
        """
        if not texts:
            return []
        placeholders = ",".join(["?"] * len(texts))
        sql = f"SELECT id FROM {self._table_name} WHERE text IN ({placeholders})"
        params: list[Any] = list(texts)

        if offset is not None and limit is None:
            raise ValueError("offset requires limit")
        if limit is not None:
            sql += " LIMIT ?"
            params.append(limit)
            if offset is not None:
                sql += " OFFSET ?"
                params.append(offset)

        with self._lock:
            rows = self.conn.execute(sql, tuple(params)).fetchall()
        return [r[0] for r in rows]

    def find_ids_by_filter(
        self,
        filter_dict: dict[str, Any],
        filter_builder: Callable[[dict[str, Any], str], tuple[str, list[Any]]],
        *,
        limit: int | None = None,
        offset: int | None = None,
    ) -> list[int]:
        """Find document IDs matching metadata filter.

        Args:
            filter_dict: Metadata key-value pairs to filter by
            filter_builder: Function to build filter clause
            limit: Maximum number of IDs to return (None = all)
            offset: Number of IDs to skip (None = 0)
        """
        if not filter_dict:
            return []

        filter_clause, filter_params = filter_builder(filter_dict, "metadata")
        # Remove leading "AND " from clause
        filter_clause = filter_clause.replace("AND ", "", 1)
        where_clause = f"WHERE {filter_clause}" if filter_clause else ""

        sql = f"SELECT id FROM {self._table_name} {where_clause}"
        params: list[Any] = list(filter_params)

        if offset is not None and limit is None:
            raise ValueError("offset requires limit")
        if limit is not None:
            sql += " LIMIT ?"
            params.append(limit)
            if offset is not None:
                sql += " OFFSET ?"
                params.append(offset)

        with self._lock:
            rows = self.conn.execute(sql, tuple(params)).fetchall()
        return [r[0] for r in rows]

    def keyword_search(
        self,
        query: str,
        k: int,
        filter_dict: dict[str, Any] | None = None,
        filter_builder: Callable | None = None,
    ) -> list[tuple[int, float]]:
        """
        Perform BM25 keyword search using FTS5.

        Args:
            query: Search query (FTS5 syntax supported)
            k: Maximum results
            filter_dict: Optional metadata filter
            filter_builder: Function to build filter clause

        Returns:
            List of (id, bm25_score) tuples, sorted by relevance
        """
        if not self._fts_enabled:
            raise RuntimeError("FTS5 not available - cannot perform keyword search")
        if not query.strip():
            return []

        filter_clause = ""
        filter_params: list[Any] = []
        if filter_dict and filter_builder:
            filter_clause, filter_params = filter_builder(filter_dict, "ti.metadata")

        sql = f"""
            SELECT ti.id, bm25({self._fts_table_name}) as score
            FROM {self._fts_table_name} f
            JOIN {self._table_name} ti ON ti.id = f.rowid
            WHERE {self._fts_table_name} MATCH ?
            {filter_clause}
            ORDER BY score ASC
            LIMIT ?
        """
        params = (query,) + tuple(filter_params) + (k,)
        with self._lock:
            rows = self.conn.execute(sql, params).fetchall()
        return [(int(row[0]), float(row[1])) for row in rows]

    def build_filter_clause(
        self, filter_dict: dict[str, Any] | None, metadata_column: str = "metadata"
    ) -> tuple[str, list[Any]]:
        """
        Build SQL WHERE clause from metadata filter dictionary.

        Args:
            filter_dict: Metadata key-value pairs to filter by
            metadata_column: Name of JSON metadata column

        Returns:
            Tuple of (where_clause, parameters) for SQL query

        Raises:
            ValueError: If filter keys are not strings or values are unsupported types
        """
        if not filter_dict:
            return "", []

        # Validate filter structure before processing
        validate_filter(filter_dict)

        clauses = []
        params: list[Any] = []
        for key, value in filter_dict.items():
            json_path = f"$.{key}"
            if isinstance(value, (int, float)):
                clauses.append(f"json_extract({metadata_column}, ?) = ?")
                params.extend([json_path, value])
            elif isinstance(value, str):
                # Use exact equality for string filters
                clauses.append(f"json_extract({metadata_column}, ?) = ?")
                params.extend([json_path, value])
            elif isinstance(value, list):
                placeholders = ",".join("?" for _ in value)
                clauses.append(
                    f"json_extract({metadata_column}, ?) IN ({placeholders})"
                )
                params.extend([json_path] + value)
            else:
                raise ValueError(f"Unsupported filter value type for {key}")
        where = " AND ".join(clauses)
        return f"AND ({where})" if where else "", params

    def count(self) -> int:
        """Return total number of documents."""
        with self._lock:
            row = self.conn.execute(
                f"SELECT COUNT(*) FROM {self._table_name}"
            ).fetchone()
        return row[0] if row else 0

    def get_all_docs_with_text(
        self,
        filter_dict: dict[str, Any] | None = None,
        filter_builder: Callable[[dict[str, Any], str], tuple[str, list[Any]]]
        | None = None,
        *,
        limit: int | None = None,
        offset: int | None = None,
    ) -> list[tuple[int, str, dict[str, Any]]]:
        """
        Get documents with their text content, with optional pagination.

        Args:
            filter_dict: Optional metadata filter
            filter_builder: Function to build filter clause
            limit: Maximum number of documents to return (None = all)
            offset: Number of documents to skip (None = 0)

        Returns:
            List of (doc_id, text, metadata) tuples
        """
        filter_clause = ""
        filter_params: list[Any] = []
        if filter_dict and filter_builder:
            filter_clause, filter_params = filter_builder(filter_dict, "metadata")

        sql = f"""
            SELECT id, text, metadata FROM {self._table_name}
            WHERE 1=1 {filter_clause}
            ORDER BY id
        """
        params: list[Any] = list(filter_params)

        if offset is not None and limit is None:
            raise ValueError("offset requires limit")
        if limit is not None:
            sql += " LIMIT ?"
            params.append(limit)
            if offset is not None:
                sql += " OFFSET ?"
                params.append(offset)

        with self._lock:
            rows = self.conn.execute(sql, tuple(params)).fetchall()
        result = []
        for row_id, text, meta_json in rows:
            meta = json.loads(meta_json) if meta_json else {}
            result.append((int(row_id), text, meta))
        return result

    def update_metadata_batch(self, updates: list[tuple[int, dict[str, Any]]]) -> int:
        """
        Update metadata for multiple documents in a single transaction.

        Merges new metadata with existing metadata (shallow merge).

        Args:
            updates: List of (doc_id, metadata_updates) tuples

        Returns:
            Number of documents updated
        """
        if not updates:
            return 0

        with self._lock, self.conn:
            updated = 0
            # Batch into chunks of 500 for performance
            for batch in _batched(updates, 500):
                ids = [u[0] for u in batch]

                # Fetch all existing metadata in one query
                placeholders = ",".join(["?"] * len(ids))
                rows = self.conn.execute(
                    f"SELECT id, metadata FROM {self._table_name} WHERE id IN ({placeholders})",
                    ids,
                ).fetchall()

                current_meta_map = {r[0]: (json.loads(r[1]) if r[1] else {}) for r in rows}

                # Prepare updates
                update_data = []
                for doc_id, meta_updates in batch:
                    if doc_id in current_meta_map:
                        meta = current_meta_map[doc_id]
                        meta.update(meta_updates)
                        update_data.append((json.dumps(meta), doc_id))
                        updated += 1

                if update_data:
                    self.conn.executemany(
                        f"UPDATE {self._table_name} SET metadata = ? WHERE id = ?",
                        update_data,
                    )

            return updated

    def check_legacy_sqlite_vec(self, vec_table_name: str) -> bool:
        """
        Check if legacy sqlite-vec tables exist (for migration).

        Args:
            vec_table_name: Expected name of the old vec0 virtual table

        Returns:
            True if legacy sqlite-vec data exists
        """
        try:
            with self._lock:
                row = self.conn.execute(
                    "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
                    (vec_table_name,),
                ).fetchone()
            return row is not None
        except Exception:
            return False

    def get_legacy_vectors(self, vec_table_name: str) -> list[tuple[int, bytes]]:
        """
        Extract vectors from legacy sqlite-vec table for migration.

        Args:
            vec_table_name: Name of the old vec0 virtual table

        Returns:
            List of (rowid, embedding_blob) tuples
        """
        _validate_table_name(vec_table_name)
        try:
            with self._lock:
                rows = self.conn.execute(
                    f"SELECT rowid, embedding FROM {vec_table_name}"
                ).fetchall()
            return [(int(r[0]), r[1]) for r in rows]
        except Exception as e:
            _logger.warning("Failed to read legacy vectors: %s", e)
            return []

    def drop_legacy_vec_table(self, vec_table_name: str) -> None:
        """Drop legacy sqlite-vec table after migration."""
        _validate_table_name(vec_table_name)
        try:
            with self._lock, self.conn:
                self.conn.execute(f"DROP TABLE IF EXISTS {vec_table_name}")
            _logger.info("Dropped legacy sqlite-vec table: %s", vec_table_name)
        except Exception as e:
            _logger.warning("Failed to drop legacy table %s: %s", vec_table_name, e)

    # ------------------------------------------------------------------ #
    # Hierarchical Relationships
    # ------------------------------------------------------------------ #

    def get_children(self, parent_id: int) -> list[tuple[int, str, dict[str, Any]]]:
        """
        Get all direct children of a document.

        Args:
            parent_id: ID of the parent document

        Returns:
            List of (id, text, metadata) tuples for child documents
        """
        with self._lock:
            rows = self.conn.execute(
                f"SELECT id, text, metadata FROM {self._table_name} WHERE parent_id = ?",
                (parent_id,),
            ).fetchall()

        return [(int(r[0]), r[1], json.loads(r[2]) if r[2] else {}) for r in rows]

    def get_parent(self, doc_id: int) -> tuple[int, str, dict[str, Any]] | None:
        """
        Get the parent document of a given document.

        Args:
            doc_id: ID of the child document

        Returns:
            Tuple of (id, text, metadata) for parent, or None if no parent
        """
        # Single self-join instead of two sequential queries
        with self._lock:
            row = self.conn.execute(
                f"""SELECT p.id, p.text, p.metadata
                FROM {self._table_name} c
                JOIN {self._table_name} p ON p.id = c.parent_id
                WHERE c.id = ?""",
                (doc_id,),
            ).fetchone()

        if not row:
            return None

        return (
            int(row[0]),
            row[1],
            json.loads(row[2]) if row[2] else {},
        )

    def get_descendants(
        self, root_id: int, max_depth: int | None = None
    ) -> list[tuple[int, str, dict[str, Any], int]]:
        """
        Get all descendants of a document (recursive).

        Uses a recursive CTE for efficient traversal.

        Args:
            root_id: ID of the root document
            max_depth: Maximum depth to traverse (None uses safety cap
                from constants.MAX_HIERARCHY_DEPTH to prevent infinite recursion)

        Returns:
            List of (id, text, metadata, depth) tuples
        """
        from .. import constants

        # Apply safety cap to prevent infinite recursion from cycles. The
        # depth is bound as a parameter rather than f-string interpolated;
        # int() coercion makes the previous f-string safe today, but the
        # parameter form is one less line away from injection on a future
        # refactor.
        effective_depth = int(max_depth) if max_depth is not None else constants.MAX_HIERARCHY_DEPTH

        sql = f"""
            WITH RECURSIVE descendants(id, text, metadata, depth) AS (
                SELECT id, text, metadata, 1 as depth
                FROM {self._table_name}
                WHERE parent_id = ?

                UNION ALL

                SELECT t.id, t.text, t.metadata, d.depth + 1
                FROM {self._table_name} t
                JOIN descendants d ON t.parent_id = d.id
                WHERE depth < ?
            )
            SELECT id, text, metadata, depth FROM descendants
            ORDER BY depth, id
        """

        with self._lock:
            rows = self.conn.execute(sql, (root_id, effective_depth)).fetchall()

        return [
            (int(r[0]), r[1], json.loads(r[2]) if r[2] else {}, int(r[3])) for r in rows
        ]

    def get_ancestors(
        self, doc_id: int, max_depth: int | None = None
    ) -> list[tuple[int, str, dict[str, Any], int]]:
        """
        Get all ancestors of a document (path to root).

        Args:
            doc_id: ID of the document
            max_depth: Maximum depth to traverse (None uses safety cap
                from constants.MAX_HIERARCHY_DEPTH to prevent infinite recursion)

        Returns:
            List of (id, text, metadata, depth) tuples, from immediate parent to root
        """
        from .. import constants

        # Apply safety cap to prevent infinite recursion from cycles. Bind
        # the depth as a parameter (see get_descendants for rationale).
        effective_depth = int(max_depth) if max_depth is not None else constants.MAX_HIERARCHY_DEPTH

        sql = f"""
            WITH RECURSIVE ancestors(id, text, metadata, parent_id, depth) AS (
                SELECT id, text, metadata, parent_id, 1 as depth
                FROM {self._table_name}
                WHERE id = (SELECT parent_id FROM {self._table_name} WHERE id = ?)

                UNION ALL

                SELECT t.id, t.text, t.metadata, t.parent_id, a.depth + 1
                FROM {self._table_name} t
                JOIN ancestors a ON t.id = a.parent_id
                WHERE a.parent_id IS NOT NULL AND a.depth < ?
            )
            SELECT id, text, metadata, depth FROM ancestors
            ORDER BY depth
        """

        with self._lock:
            rows = self.conn.execute(sql, (doc_id, effective_depth)).fetchall()

        return [
            (int(r[0]), r[1], json.loads(r[2]) if r[2] else {}, int(r[3])) for r in rows
        ]

    def set_parent(self, doc_id: int, parent_id: int | None) -> bool:
        """
        Set or update the parent of a document.

        Args:
            doc_id: ID of the document to update
            parent_id: New parent ID (None to remove parent)

        Returns:
            True if document was updated, False if not found

        Raises:
            ValueError: If setting parent would create a cycle
        """
        # Cycle check + UPDATE inside one critical section so a concurrent
        # writer cannot create a cycle-forming edge between the check and the
        # UPDATE. The lock serializes; `with self.conn:` wraps the UPDATE in
        # an implicit transaction that commits on success.
        with self._lock, self.conn:
            if parent_id is not None:
                if parent_id == doc_id:
                    raise ValueError("A document cannot be its own parent")
                descendants = self.get_descendants(doc_id)
                descendant_ids = {d[0] for d in descendants}
                if parent_id in descendant_ids:
                    raise ValueError(
                        f"Cannot set parent: document {parent_id} is a descendant of {doc_id}"
                    )

            cursor = self.conn.execute(
                f"UPDATE {self._table_name} SET parent_id = ? WHERE id = ?",
                (parent_id, doc_id),
            )
            return cursor.rowcount > 0

    # ------------------------------------------------------------------ #
    # Cluster State Persistence
    # ------------------------------------------------------------------ #

    def _ensure_cluster_table(self) -> None:
        """Create cluster state table if it doesn't exist."""
        if self._cluster_table_ready:
            return
        cluster_table = self._cluster_table_name
        with self._lock, self.conn:
            # Re-check inside the lock so concurrent first-callers don't
            # both run the DDL. The CREATE TABLE IF NOT EXISTS is itself
            # idempotent, but doing the work twice defeats the early-exit.
            if self._cluster_table_ready:
                return
            self.conn.execute(
                f"""
                CREATE TABLE IF NOT EXISTS {cluster_table} (
                    name TEXT PRIMARY KEY,
                    algorithm TEXT NOT NULL,
                    n_clusters INTEGER NOT NULL,
                    centroids BLOB,
                    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
                    metadata TEXT
                )
                """
            )
            self._cluster_table_ready = True

    def save_cluster_state(
        self,
        name: str,
        algorithm: str,
        n_clusters: int,
        centroids: bytes | None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        """
        Save cluster state for later reuse.

        Args:
            name: Unique name for this cluster configuration
            algorithm: Algorithm used (kmeans, minibatch_kmeans, hdbscan)
            n_clusters: Number of clusters
            centroids: Serialized centroid array (numpy bytes)
            metadata: Additional metadata (inertia, silhouette, etc.)
        """
        self._ensure_cluster_table()
        cluster_table = self._cluster_table_name

        meta_json = json.dumps(metadata) if metadata else None

        with self._lock, self.conn:
            self.conn.execute(
                f"""
                INSERT OR REPLACE INTO {cluster_table}
                (name, algorithm, n_clusters, centroids, metadata, created_at)
                VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
                """,
                (name, algorithm, n_clusters, centroids, meta_json),
            )

    def load_cluster_state(
        self, name: str
    ) -> tuple[str, int, bytes | None, dict[str, Any]] | None:
        """
        Load saved cluster state.

        Args:
            name: Name of the cluster configuration

        Returns:
            Tuple of (algorithm, n_clusters, centroids_bytes, metadata) or None
        """
        self._ensure_cluster_table()
        cluster_table = self._cluster_table_name

        # Serialize on the connection-level lock — sqlite3.Connection is not
        # safe for concurrent statement execution from multiple threads.
        with self._lock:
            row = self.conn.execute(
                f"SELECT algorithm, n_clusters, centroids, metadata FROM {cluster_table} WHERE name = ?",
                (name,),
            ).fetchone()

        if not row:
            return None

        algorithm, n_clusters, centroids, meta_json = row
        metadata = json.loads(meta_json) if meta_json else {}
        return (algorithm, n_clusters, centroids, metadata)

    def list_cluster_states(self) -> list[dict[str, Any]]:
        """List all saved cluster configurations."""
        self._ensure_cluster_table()
        cluster_table = self._cluster_table_name

        with self._lock:
            rows = self.conn.execute(
                f"SELECT name, algorithm, n_clusters, created_at, metadata FROM {cluster_table}"
            ).fetchall()

        result = []
        for name, algorithm, n_clusters, created_at, meta_json in rows:
            result.append(
                {
                    "name": name,
                    "algorithm": algorithm,
                    "n_clusters": n_clusters,
                    "created_at": created_at,
                    "metadata": json.loads(meta_json) if meta_json else {},
                }
            )
        return result

    def delete_cluster_state(self, name: str) -> bool:
        """Delete a saved cluster configuration."""
        self._ensure_cluster_table()
        cluster_table = self._cluster_table_name

        with self._lock, self.conn:
            cursor = self.conn.execute(
                f"DELETE FROM {cluster_table} WHERE name = ?", (name,)
            )
        return cursor.rowcount > 0