"""
Context repository for managing context entries.

This module handles all database operations related to context entries,
including CRUD operations and deduplication logic.
"""

from __future__ import annotations

import hashlib
import json
import logging
import sqlite3
from typing import TYPE_CHECKING
from typing import Any
from typing import cast

from pydantic import ValidationError

from app.backends.base import StorageBackend
from app.repositories.base import BaseRepository

if TYPE_CHECKING:
    import asyncpg

    from app.backends.base import TransactionContext

logger = logging.getLogger(__name__)

# Explicit column list to avoid exposing internal database columns (e.g., text_search_vector)
# This constant is used in all SELECT queries that return context entries to ensure
# only the expected columns are returned, preventing internal PostgreSQL columns from
# leaking into API responses.
CONTEXT_ENTRY_COLUMNS = 'id, thread_id, source, content_type, text_content, metadata, summary, created_at, updated_at'


def compute_content_hash(text: str) -> str:
    """Compute SHA-256 hash of text content for deduplication.

    Used to avoid transferring full text_content over the network when
    checking for duplicates. The hash is stored alongside text_content
    and compared instead of the full text.

    Args:
        text: The text content to hash.

    Returns:
        SHA-256 hex digest string (64 characters).
    """
    return hashlib.sha256(text.encode('utf-8')).hexdigest()


class ContextRepository(BaseRepository):
    """Repository for context entry operations.

    Handles storage, retrieval, search, and deletion of context entries
    with proper deduplication and transaction management.
    """

    def __init__(self, backend: StorageBackend) -> None:
        """Initialize context repository.

        Args:
            backend: Storage backend for executing database operations
        """
        super().__init__(backend)

    async def store_with_deduplication(
        self,
        thread_id: str,
        source: str,
        content_type: str,
        text_content: str,
        metadata: str | None = None,
        summary: str | None = None,
        txn: TransactionContext | None = None,
    ) -> tuple[int, bool]:
        """Store context entry with deduplication logic.

        Checks if the latest entry has identical thread_id, source, and text_content.
        If found, updates metadata (via COALESCE), content_type, and updated_at.
        Otherwise, inserts new entry.

        Args:
            thread_id: Thread identifier
            source: 'user' or 'agent'
            content_type: 'text' or 'multimodal'
            text_content: The actual text content
            metadata: JSON metadata string or None
            summary: LLM-generated summary text or None
            txn: Optional transaction context for atomic multi-repository operations.
                When provided, uses the transaction's connection directly.
                When None, uses execute_write() for standalone operation.

        Returns:
            Tuple of (context_id, was_updated) where was_updated=True means
            an existing entry was updated, False means new entry was inserted.
        """
        backend_type = txn.backend_type if txn else self.backend.backend_type

        # Normalize empty/whitespace summary to None for proper COALESCE behavior.
        # COALESCE(NULL, existing_value) preserves existing; COALESCE("", existing_value) overwrites.
        if summary is not None and not summary.strip():
            summary = None

        # Compute content hash for deduplication optimization.
        # Avoids transferring full text_content over the network for duplicate checks.
        content_hash = compute_content_hash(text_content)

        if backend_type == 'sqlite':

            def _store_sqlite(conn: sqlite3.Connection) -> tuple[int, bool]:
                cursor = conn.cursor()

                # Check if the LATEST entry (by id) for this thread_id and source is a duplicate.
                # Fetches content_hash instead of full text_content to reduce data transfer.
                # Falls back to text comparison when content_hash is NULL (pre-migration rows).
                cursor.execute(
                    f'''
                    SELECT id, content_hash, text_content FROM context_entries
                    WHERE thread_id = {self._placeholder(1)} AND source = {self._placeholder(2)}
                    ORDER BY id DESC
                    LIMIT 1
                    ''',
                    (thread_id, source),
                )

                latest_row = cursor.fetchone()

                is_duplicate = False
                if latest_row:
                    existing_hash = latest_row['content_hash']
                    if existing_hash is not None:
                        # Hash-based comparison (fast path)
                        is_duplicate = existing_hash == content_hash
                    else:
                        # Fallback for pre-migration rows without content_hash
                        is_duplicate = latest_row['text_content'] == text_content

                if is_duplicate and latest_row:
                    # Interleaving check: suppress dedup if opposite-source entries exist
                    # after the candidate. This preserves chronological ordering when identical
                    # text is sent as a new conversational turn rather than a retry.
                    # Intentionally duplicated across 4 blocks (SQLite/PostgreSQL x store/check)
                    # because sync/async closure patterns prevent clean extraction without
                    # losing type safety.
                    existing_id = latest_row['id']
                    opposite_source = 'agent' if source == 'user' else 'user'
                    cursor.execute(
                        f'''
                        SELECT 1 FROM context_entries
                        WHERE thread_id = {self._placeholder(1)} AND source = {self._placeholder(2)}
                        AND id > {self._placeholder(3)}
                        LIMIT 1
                        ''',
                        (thread_id, opposite_source, existing_id),
                    )
                    if cursor.fetchone() is not None:
                        is_duplicate = False

                if is_duplicate and latest_row:
                    # The latest entry has identical text - update metadata, content_type, and timestamp
                    existing_id = latest_row['id']
                    cursor.execute(
                        f'''
                        UPDATE context_entries
                        SET metadata = COALESCE({self._placeholder(1)}, metadata),
                            content_type = {self._placeholder(2)},
                            summary = COALESCE({self._placeholder(3)}, summary),
                            content_hash = {self._placeholder(4)},
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = {self._placeholder(5)}
                        ''',
                        (metadata, content_type, summary, content_hash, existing_id),
                    )
                    rows_affected = cursor.rowcount
                    if rows_affected == 0:
                        logger.warning(
                            'Deduplication UPDATE affected 0 rows for context %d in thread %s',
                            existing_id, thread_id,
                        )
                    logger.debug(f'Updated existing context entry {existing_id} for thread {thread_id}')
                    return existing_id, True

                # No duplicate - insert new entry
                cursor.execute(
                    f'''
                    INSERT INTO context_entries
                    (thread_id, source, content_type, text_content, metadata, summary, content_hash)
                    VALUES ({self._placeholders(7)})
                    ''',
                    (thread_id, source, content_type, text_content, metadata, summary, content_hash),
                )
                new_id: int = cursor.lastrowid if cursor.lastrowid is not None else 0
                logger.debug(f'Inserted new context entry {new_id} for thread {thread_id}')
                return new_id, False

            if txn:
                return _store_sqlite(cast(sqlite3.Connection, txn.connection))
            return await self.backend.execute_write(_store_sqlite)

        # PostgreSQL
        # Note: TYPE_CHECKING ensures asyncpg.Connection type is only used during type checking
        async def _store_postgresql(conn: asyncpg.Connection) -> tuple[int, bool]:
            # Check latest entry - fetches content_hash instead of full text_content.
            # Falls back to text comparison when content_hash is NULL (pre-migration rows).
            latest_row = await conn.fetchrow(
                f'''
                    SELECT id, content_hash, text_content FROM context_entries
                    WHERE thread_id = {self._placeholder(1)} AND source = {self._placeholder(2)}
                    ORDER BY id DESC
                    LIMIT 1
                    ''',
                thread_id,
                source,
            )

            is_duplicate = False
            if latest_row:
                existing_hash = latest_row['content_hash']
                if existing_hash is not None:
                    is_duplicate = existing_hash == content_hash
                else:
                    is_duplicate = latest_row['text_content'] == text_content

            if is_duplicate and latest_row:
                # Interleaving check: suppress dedup if opposite-source entries exist
                # after the candidate. This preserves chronological ordering when identical
                # text is sent as a new conversational turn rather than a retry.
                # Intentionally duplicated across 4 blocks (SQLite/PostgreSQL x store/check)
                # because sync/async closure patterns prevent clean extraction without
                # losing type safety.
                existing_id = latest_row['id']
                opposite_source = 'agent' if source == 'user' else 'user'
                interleaving_row = await conn.fetchrow(
                    f'''
                        SELECT 1 FROM context_entries
                        WHERE thread_id = {self._placeholder(1)} AND source = {self._placeholder(2)}
                        AND id > {self._placeholder(3)}
                        LIMIT 1
                        ''',
                    thread_id,
                    opposite_source,
                    existing_id,
                )
                if interleaving_row is not None:
                    is_duplicate = False

            if is_duplicate and latest_row:
                # Update metadata, content_type, and timestamp
                existing_id = latest_row['id']
                result = await conn.execute(
                    f'''
                        UPDATE context_entries
                        SET metadata = COALESCE({self._placeholder(1)}, metadata),
                            content_type = {self._placeholder(2)},
                            summary = COALESCE({self._placeholder(3)}, summary),
                            content_hash = {self._placeholder(4)},
                            updated_at = CURRENT_TIMESTAMP
                        WHERE id = {self._placeholder(5)}
                        ''',
                    metadata,
                    content_type,
                    summary,
                    content_hash,
                    existing_id,
                )
                rows_affected = int(result.split()[-1]) if result else 0
                if rows_affected == 0:
                    logger.warning(
                        'Deduplication UPDATE affected 0 rows for context %d in thread %s',
                        existing_id, thread_id,
                    )
                logger.debug(f'Updated existing context entry {existing_id} for thread {thread_id}')
                return existing_id, True

            # Insert new entry with RETURNING clause
            new_id_result = await conn.fetchval(
                f'''
                    INSERT INTO context_entries
                    (thread_id, source, content_type, text_content, metadata, summary, content_hash)
                    VALUES ({self._placeholders(7)})
                    RETURNING id
                    ''',
                thread_id,
                source,
                content_type,
                text_content,
                metadata,
                summary,
                content_hash,
            )
            new_id = cast(int, new_id_result)
            logger.debug(f'Inserted new context entry {new_id} for thread {thread_id}')
            return new_id, False

        if txn:
            return await _store_postgresql(cast('asyncpg.Connection', txn.connection))
        return await self.backend.execute_write(_store_postgresql)

    async def check_latest_is_duplicate(
        self,
        thread_id: str,
        source: str,
        text_content: str,
    ) -> int | None:
        """Check if the latest entry matches the given content (read-only pre-check).

        This is a performance optimization for the embedding-first pattern.
        It allows skipping expensive embedding generation when the content
        is identical to the latest entry. The in-transaction deduplication
        in store_with_deduplication remains as the authoritative safety net.

        Includes an interleaving check: if opposite-source entries (agent for
        user source, user for agent source) exist after the candidate duplicate,
        returns None to suppress deduplication. This preserves chronological
        ordering when identical text is sent as a new conversational turn
        rather than a retry.

        Args:
            thread_id: Thread identifier
            source: 'user' or 'agent'
            text_content: Text content to check for duplicates

        Returns:
            The context_id of the matching entry if duplicate found, None if no
            match or if interleaving entries suppress deduplication.
        """
        content_hash = compute_content_hash(text_content)

        if self.backend.backend_type == 'sqlite':

            def _check_sqlite(conn: sqlite3.Connection) -> int | None:
                cursor = conn.cursor()
                cursor.execute(
                    f'''
                    SELECT id, content_hash, text_content FROM context_entries
                    WHERE thread_id = {self._placeholder(1)} AND source = {self._placeholder(2)}
                    ORDER BY id DESC
                    LIMIT 1
                    ''',
                    (thread_id, source),
                )
                row = cursor.fetchone()
                if not row:
                    return None
                # Hash-based comparison; fall back to text for pre-migration rows (NULL hash)
                existing_hash = row['content_hash']
                is_match = (
                    existing_hash == content_hash
                    if existing_hash is not None
                    else row['text_content'] == text_content
                )
                if not is_match:
                    return None
                candidate_id = row['id']
                # Interleaving check: suppress dedup if opposite-source entries exist
                # after the candidate. This preserves chronological ordering when identical
                # text is sent as a new conversational turn rather than a retry.
                # Intentionally duplicated across 4 blocks (SQLite/PostgreSQL x store/check)
                # because sync/async closure patterns prevent clean extraction without
                # losing type safety.
                opposite_source = 'agent' if source == 'user' else 'user'
                cursor.execute(
                    f'''
                    SELECT 1 FROM context_entries
                    WHERE thread_id = {self._placeholder(1)} AND source = {self._placeholder(2)}
                    AND id > {self._placeholder(3)}
                    LIMIT 1
                    ''',
                    (thread_id, opposite_source, candidate_id),
                )
                if cursor.fetchone() is not None:
                    return None
                return cast(int, candidate_id)

            return await self.backend.execute_read(_check_sqlite)

        # PostgreSQL
        async def _check_postgresql(conn: asyncpg.Connection) -> int | None:
            row = await conn.fetchrow(
                f'''
                    SELECT id, content_hash, text_content FROM context_entries
                    WHERE thread_id = {self._placeholder(1)} AND source = {self._placeholder(2)}
                    ORDER BY id DESC
                    LIMIT 1
                    ''',
                thread_id,
                source,
            )
            if not row:
                return None
            # Hash-based comparison; fall back to text for pre-migration rows (NULL hash)
            existing_hash = row['content_hash']
            is_match = (
                existing_hash == content_hash
                if existing_hash is not None
                else row['text_content'] == text_content
            )
            if not is_match:
                return None
            candidate_id = row['id']
            # Interleaving check: suppress dedup if opposite-source entries exist
            # after the candidate. This preserves chronological ordering when identical
            # text is sent as a new conversational turn rather than a retry.
            # Intentionally duplicated across 4 blocks (SQLite/PostgreSQL x store/check)
            # because sync/async closure patterns prevent clean extraction without
            # losing type safety.
            opposite_source = 'agent' if source == 'user' else 'user'
            interleaving_row = await conn.fetchrow(
                f'''
                    SELECT 1 FROM context_entries
                    WHERE thread_id = {self._placeholder(1)} AND source = {self._placeholder(2)}
                    AND id > {self._placeholder(3)}
                    LIMIT 1
                    ''',
                thread_id,
                opposite_source,
                candidate_id,
            )
            if interleaving_row is not None:
                return None
            return cast(int, candidate_id)

        return await self.backend.execute_read(_check_postgresql)

    async def get_summary(self, context_id: int) -> str | None:
        """Get the summary for a context entry.

        Used during deduplication to check if a summary already exists,
        avoiding unnecessary LLM calls for duplicate entries.

        Args:
            context_id: ID of the context entry.

        Returns:
            Summary string if exists, None otherwise.
        """
        if self.backend.backend_type == 'sqlite':

            def _get_summary_sqlite(conn: sqlite3.Connection) -> str | None:
                cursor = conn.cursor()
                cursor.execute(
                    f'SELECT summary FROM context_entries WHERE id = {self._placeholder(1)}',
                    (context_id,),
                )
                row = cursor.fetchone()
                if row:
                    value = cast(str | None, row['summary'])
                    # Normalize empty/whitespace to None for consistent behavior
                    if value is not None and not value.strip():
                        return None
                    return value
                return None

            return await self.backend.execute_read(_get_summary_sqlite)

        # PostgreSQL
        async def _get_summary_postgresql(conn: asyncpg.Connection) -> str | None:
            row = await conn.fetchrow(
                f'SELECT summary FROM context_entries WHERE id = {self._placeholder(1)}',
                context_id,
            )
            if row:
                value = cast(str | None, row['summary'])
                # Normalize empty/whitespace to None for consistent behavior
                if value is not None and not value.strip():
                    return None
                return value
            return None

        return await self.backend.execute_read(_get_summary_postgresql)

    async def search_contexts(
        self,
        thread_id: str | None = None,
        source: str | None = None,
        content_type: str | None = None,
        tags: list[str] | None = None,
        metadata: dict[str, str | int | float | bool] | None = None,
        metadata_filters: list[dict[str, Any]] | None = None,
        start_date: str | None = None,
        end_date: str | None = None,
        limit: int = 50,
        offset: int = 0,
        explain_query: bool = False,
    ) -> tuple[list[Any], dict[str, Any]]:
        """Search for context entries with filtering including metadata and date range.

        Args:
            thread_id: Filter by thread ID
            source: Filter by source ('user' or 'agent')
            content_type: Filter by content type
            tags: Filter by tags (OR logic)
            metadata: Simple metadata filters (key=value)
            metadata_filters: Advanced metadata filters with operators
            start_date: Filter by created_at >= date (ISO 8601 format)
            end_date: Filter by created_at <= date (ISO 8601 format)
            limit: Maximum number of results
            offset: Pagination offset
            explain_query: If True, include query execution plan

        Returns:
            Tuple of (matching rows, query statistics)
            Note: Rows can be sqlite3.Row or asyncpg.Record depending on backend
        """
        import time as time_module

        from app.metadata_types import MetadataFilter
        from app.query_builder import MetadataQueryBuilder

        if self.backend.backend_type == 'sqlite':

            def _search_sqlite(conn: sqlite3.Connection) -> tuple[list[Any], dict[str, Any]]:
                start_time = time_module.time()
                cursor = conn.cursor()

                # Build query with indexed fields first for optimization
                # Use explicit column list to avoid exposing internal columns (e.g., text_search_vector)
                query = f'SELECT {CONTEXT_ENTRY_COLUMNS} FROM context_entries WHERE 1=1'
                params: list[Any] = []

                # Thread filter (indexed)
                if thread_id:
                    query += f' AND thread_id = {self._placeholder(len(params) + 1)}'
                    params.append(thread_id)

                # Source filter (indexed)
                if source:
                    query += f' AND source = {self._placeholder(len(params) + 1)}'
                    params.append(source)

                # Content type filter
                if content_type:
                    query += f' AND content_type = {self._placeholder(len(params) + 1)}'
                    params.append(content_type)

                # Date range filtering - Use datetime() to normalize ISO 8601 input
                # datetime() converts all ISO 8601 formats (T separator, Z suffix, timezone offsets)
                # to SQLite's space-separated format 'YYYY-MM-DD HH:MM:SS' for proper comparison.
                # Without datetime(), TEXT comparison fails because 'T' > ' ' in ASCII ordering.
                if start_date:
                    query += f' AND created_at >= datetime({self._placeholder(len(params) + 1)})'
                    params.append(start_date)

                if end_date:
                    query += f' AND created_at <= datetime({self._placeholder(len(params) + 1)})'
                    params.append(end_date)

                # Add metadata filtering
                metadata_builder = MetadataQueryBuilder(backend_type='sqlite')

                # Simple metadata filters
                if metadata:
                    for key, value in metadata.items():
                        metadata_builder.add_simple_filter(key, value)

                # Advanced metadata filters
                if metadata_filters:
                    validation_errors: list[str] = []
                    for filter_dict in metadata_filters:
                        try:
                            # Convert dict to MetadataFilter
                            filter_spec = MetadataFilter(**filter_dict)
                            metadata_builder.add_advanced_filter(filter_spec)
                        except ValidationError as e:
                            # Collect validation errors to return to user
                            error_msg = f'Invalid metadata filter {filter_dict}: {e}'
                            validation_errors.append(error_msg)
                        except ValueError as e:
                            # Handle value errors (e.g., from field validators)
                            error_msg = f'Invalid metadata filter {filter_dict}: {e}'
                            validation_errors.append(error_msg)
                        except Exception as e:
                            # Unexpected errors - still collect them
                            error_msg = f'Unexpected error in metadata filter {filter_dict}: {e}'
                            validation_errors.append(error_msg)
                            logger.error(f'Unexpected error processing metadata filter: {e}')

                    # If there were validation errors, return them immediately
                    if validation_errors:
                        error_response = {
                            'error': 'Metadata filter validation failed',
                            'validation_errors': validation_errors,
                            'execution_time_ms': 0.0,
                            'filters_applied': 0,
                            'rows_returned': 0,
                        }
                        return [], error_response

                # Add metadata conditions to query
                metadata_clause, metadata_params = metadata_builder.build_where_clause()
                if metadata_clause:
                    query += f' AND {metadata_clause}'
                    params.extend(metadata_params)

                # Tag filter (uses subquery with indexed tag table)
                if tags:
                    normalized_tags = [tag.strip().lower() for tag in tags if tag.strip()]
                    if normalized_tags:
                        tag_placeholders = ','.join([
                            self._placeholder(len(params) + i + 1) for i in range(len(normalized_tags))
                        ])
                        query += f'''
                            AND id IN (
                                SELECT DISTINCT context_entry_id
                                FROM tags
                                WHERE tag IN ({tag_placeholders})
                            )
                        '''
                        params.extend(normalized_tags)

                # Order and pagination - use id as secondary sort for consistency
                limit_placeholder = self._placeholder(len(params) + 1)
                offset_placeholder = self._placeholder(len(params) + 2)
                query += f' ORDER BY created_at DESC, id DESC LIMIT {limit_placeholder} OFFSET {offset_placeholder}'
                params.extend((limit, offset))

                cursor.execute(query, tuple(params))
                rows = cursor.fetchall()

                # Calculate execution time
                execution_time_ms = (time_module.time() - start_time) * 1000

                # Build statistics
                stats: dict[str, Any] = {
                    'execution_time_ms': round(execution_time_ms, 2),
                    'filters_applied': metadata_builder.get_filter_count(),
                    'rows_returned': len(rows),
                    'backend': 'sqlite',
                }

                # Get query plan if requested
                if explain_query:
                    cursor.execute(f'EXPLAIN QUERY PLAN {query}', tuple(params))
                    plan_rows = cursor.fetchall()
                    # Convert sqlite3.Row objects to readable format
                    plan_data: list[str] = []
                    for row in plan_rows:
                        # Convert sqlite3.Row to dict to avoid <Row object> repr
                        row_dict = dict(row)
                        # SQLite EXPLAIN QUERY PLAN columns: id, parent, notused, detail
                        id_val = row_dict.get('id', '?')
                        parent_val = row_dict.get('parent', '?')
                        notused_val = row_dict.get('notused', '?')
                        detail_val = row_dict.get('detail', '?')
                        formatted = f'id:{id_val} parent:{parent_val} notused:{notused_val} detail:{detail_val}'
                        plan_data.append(formatted)
                    stats['query_plan'] = '\n'.join(plan_data)

                # Return list of rows and statistics
                return list(rows), stats

            return await self.backend.execute_read(_search_sqlite)

        # PostgreSQL
        async def _search_postgresql(conn: asyncpg.Connection) -> tuple[list[Any], dict[str, Any]]:
            start_time = time_module.time()

            # Build query with indexed fields first for optimization
            # Use explicit column list to avoid exposing internal columns (e.g., text_search_vector)
            query = f'SELECT {CONTEXT_ENTRY_COLUMNS} FROM context_entries WHERE 1=1'
            params: list[Any] = []

            # Thread filter (indexed)
            if thread_id:
                query += f' AND thread_id = {self._placeholder(len(params) + 1)}'
                params.append(thread_id)

            # Source filter (indexed)
            if source:
                query += f' AND source = {self._placeholder(len(params) + 1)}'
                params.append(source)

            # Content type filter
            if content_type:
                query += f' AND content_type = {self._placeholder(len(params) + 1)}'
                params.append(content_type)

            # Date range filtering - PostgreSQL uses TIMESTAMPTZ comparison
            # asyncpg requires Python datetime objects, not strings, for TIMESTAMPTZ parameters
            if start_date:
                query += f' AND created_at >= {self._placeholder(len(params) + 1)}'
                params.append(self._parse_date_for_postgresql(start_date))

            if end_date:
                query += f' AND created_at <= {self._placeholder(len(params) + 1)}'
                params.append(self._parse_date_for_postgresql(end_date))

            # Add metadata filtering
            # Pass param_offset so metadata builder knows current parameter position
            metadata_builder = MetadataQueryBuilder(backend_type='postgresql', param_offset=len(params))

            # Simple metadata filters
            if metadata:
                for key, value in metadata.items():
                    metadata_builder.add_simple_filter(key, value)

            # Advanced metadata filters
            if metadata_filters:
                validation_errors: list[str] = []
                for filter_dict in metadata_filters:
                    try:
                        # Convert dict to MetadataFilter
                        filter_spec = MetadataFilter(**filter_dict)
                        metadata_builder.add_advanced_filter(filter_spec)
                    except ValidationError as e:
                        # Collect validation errors to return to user
                        error_msg = f'Invalid metadata filter {filter_dict}: {e}'
                        validation_errors.append(error_msg)
                    except ValueError as e:
                        # Handle value errors (e.g., from field validators)
                        error_msg = f'Invalid metadata filter {filter_dict}: {e}'
                        validation_errors.append(error_msg)
                    except Exception as e:
                        # Unexpected errors - still collect them
                        error_msg = f'Unexpected error in metadata filter {filter_dict}: {e}'
                        validation_errors.append(error_msg)
                        logger.error(f'Unexpected error processing metadata filter: {e}')

                # If there were validation errors, return them immediately
                if validation_errors:
                    error_response = {
                        'error': 'Metadata filter validation failed',
                        'validation_errors': validation_errors,
                        'execution_time_ms': 0.0,
                        'filters_applied': 0,
                        'rows_returned': 0,
                    }
                    return [], error_response

            # Add metadata conditions to query
            metadata_clause, metadata_params = metadata_builder.build_where_clause()
            if metadata_clause:
                query += f' AND {metadata_clause}'
                params.extend(metadata_params)

            # Tag filter (uses subquery with indexed tag table)
            if tags:
                normalized_tags = [tag.strip().lower() for tag in tags if tag.strip()]
                if normalized_tags:
                    tag_placeholders = ','.join([self._placeholder(len(params) + i + 1) for i in range(len(normalized_tags))])
                    query += f'''
                        AND id IN (
                            SELECT DISTINCT context_entry_id
                            FROM tags
                            WHERE tag IN ({tag_placeholders})
                        )
                    '''
                    params.extend(normalized_tags)

            # Order and pagination - use id as secondary sort for consistency
            limit_placeholder = self._placeholder(len(params) + 1)
            offset_placeholder = self._placeholder(len(params) + 2)
            query += f' ORDER BY created_at DESC, id DESC LIMIT {limit_placeholder} OFFSET {offset_placeholder}'
            params.extend((limit, offset))

            rows = await conn.fetch(query, *params)

            # Calculate execution time
            execution_time_ms = (time_module.time() - start_time) * 1000

            # Build statistics
            stats: dict[str, Any] = {
                'execution_time_ms': round(execution_time_ms, 2),
                'filters_applied': metadata_builder.get_filter_count(),
                'rows_returned': len(rows),
                'backend': 'postgresql',
            }

            # Get query plan if requested (PostgreSQL EXPLAIN format)
            if explain_query:
                explain_result = await conn.fetch(f'EXPLAIN {query}', *params)
                plan_data: list[str] = [record['QUERY PLAN'] for record in explain_result]
                stats['query_plan'] = '\n'.join(plan_data)

            # Return list of rows and statistics
            return list(rows), stats

        return await self.backend.execute_read(_search_postgresql)

    async def get_by_ids(self, context_ids: list[int]) -> list[Any]:
        """Get context entries by their IDs.

        Args:
            context_ids: List of context entry IDs

        Returns:
            List of context entry rows (sqlite3.Row or asyncpg.Record depending on backend)
        """
        # Defensive check: return empty list if no IDs provided
        # Prevents SQL syntax errors when constructing IN clauses
        if not context_ids:
            return []

        if self.backend.backend_type == 'sqlite':

            def _fetch_sqlite(conn: sqlite3.Connection) -> list[Any]:
                cursor = conn.cursor()
                placeholders = ','.join([self._placeholder(i + 1) for i in range(len(context_ids))])
                # Use explicit column list to avoid exposing internal columns (e.g., text_search_vector)
                query = f'''
                    SELECT {CONTEXT_ENTRY_COLUMNS} FROM context_entries
                    WHERE id IN ({placeholders})
                    ORDER BY created_at DESC
                '''
                cursor.execute(query, tuple(context_ids))
                return list(cursor.fetchall())

            return await self.backend.execute_read(_fetch_sqlite)

        # PostgreSQL
        async def _fetch_postgresql(conn: asyncpg.Connection) -> list[Any]:
            placeholders = ','.join([self._placeholder(i + 1) for i in range(len(context_ids))])
            # Use explicit column list to avoid exposing internal columns (e.g., text_search_vector)
            query = f'''
                SELECT {CONTEXT_ENTRY_COLUMNS} FROM context_entries
                WHERE id IN ({placeholders})
                ORDER BY created_at DESC
            '''
            rows = await conn.fetch(query, *context_ids)
            return list(rows)

        return await self.backend.execute_read(_fetch_postgresql)

    async def delete_by_ids(
        self,
        context_ids: list[int],
        txn: TransactionContext | None = None,
    ) -> int:
        """Delete context entries by their IDs.

        Args:
            context_ids: List of context entry IDs to delete
            txn: Optional transaction context for atomic multi-repository operations.
                When provided, uses the transaction's connection directly.
                When None, uses execute_write() for standalone operation.

        Returns:
            Number of deleted entries
        """
        # Defensive check: return 0 if no IDs provided
        # Prevents SQL syntax errors when constructing IN clauses
        if not context_ids:
            return 0

        backend_type = txn.backend_type if txn else self.backend.backend_type

        if backend_type == 'sqlite':

            def _delete_by_ids_sqlite(conn: sqlite3.Connection) -> int:
                cursor = conn.cursor()
                placeholders = ','.join([self._placeholder(i + 1) for i in range(len(context_ids))])
                cursor.execute(
                    f'DELETE FROM context_entries WHERE id IN ({placeholders})',
                    tuple(context_ids),
                )
                return cursor.rowcount

            if txn:
                return _delete_by_ids_sqlite(cast(sqlite3.Connection, txn.connection))
            return await self.backend.execute_write(_delete_by_ids_sqlite)

        # PostgreSQL
        async def _delete_by_ids_postgresql(conn: asyncpg.Connection) -> int:
            placeholders = ','.join([self._placeholder(i + 1) for i in range(len(context_ids))])
            result = await conn.execute(
                f'DELETE FROM context_entries WHERE id IN ({placeholders})',
                *context_ids,
            )
            # asyncpg returns "DELETE N" where N is the count
            return int(result.split()[-1]) if result else 0

        if txn:
            return await _delete_by_ids_postgresql(cast('asyncpg.Connection', txn.connection))
        return await self.backend.execute_write(_delete_by_ids_postgresql)

    async def delete_by_thread(self, thread_id: str) -> int:
        """Delete all context entries in a thread.

        Args:
            thread_id: Thread ID to delete entries from

        Returns:
            Number of deleted entries
        """
        if self.backend.backend_type == 'sqlite':

            def _delete_by_thread_sqlite(conn: sqlite3.Connection) -> int:
                cursor = conn.cursor()
                cursor.execute(
                    f'DELETE FROM context_entries WHERE thread_id = {self._placeholder(1)}',
                    (thread_id,),
                )
                return cursor.rowcount

            return await self.backend.execute_write(_delete_by_thread_sqlite)

        # PostgreSQL
        async def _delete_by_thread_postgresql(conn: asyncpg.Connection) -> int:
            result = await conn.execute(
                f'DELETE FROM context_entries WHERE thread_id = {self._placeholder(1)}',
                thread_id,
            )
            # asyncpg returns "DELETE N" where N is the count
            return int(result.split()[-1]) if result else 0

        return await self.backend.execute_write(_delete_by_thread_postgresql)

    async def update_context_entry(
        self,
        context_id: int,
        text_content: str | None = None,
        metadata: str | None = None,
        summary: str | None = None,
        clear_summary: bool = False,
        txn: TransactionContext | None = None,
    ) -> tuple[bool, list[str]]:
        """Update text content and/or metadata of a context entry.

        Args:
            context_id: ID of the context entry to update
            text_content: New text content (if provided)
            metadata: New metadata JSON string (if provided)
            summary: New LLM-generated summary text (if provided)
            clear_summary: If True, explicitly set summary to NULL in the database.
                Takes precedence over summary parameter.
            txn: Optional transaction context for atomic multi-repository operations.
                When provided, uses the transaction's connection directly.
                When None, uses execute_write() for standalone operation.

        Returns:
            Tuple of (success, list_of_updated_fields)
        """
        backend_type = txn.backend_type if txn else self.backend.backend_type

        if backend_type == 'sqlite':

            def _update_entry_sqlite(conn: sqlite3.Connection) -> tuple[bool, list[str]]:
                cursor = conn.cursor()
                updated_fields: list[str] = []

                # First, check if the entry exists
                cursor.execute(
                    f'SELECT id FROM context_entries WHERE id = {self._placeholder(1)}',
                    (context_id,),
                )
                if not cursor.fetchone():
                    return False, []

                # Build update query dynamically based on provided fields
                update_parts: list[str] = []
                params: list[Any] = []

                if text_content is not None:
                    update_parts.extend([
                        f'text_content = {self._placeholder(len(params) + 1)}',
                        f'content_hash = {self._placeholder(len(params) + 2)}',
                    ])
                    params.extend([text_content, compute_content_hash(text_content)])
                    updated_fields.append('text_content')

                if metadata is not None:
                    update_parts.append(f'metadata = {self._placeholder(len(params) + 1)}')
                    params.append(metadata)
                    updated_fields.append('metadata')

                if clear_summary:
                    update_parts.append('summary = NULL')
                    updated_fields.append('summary')
                elif summary is not None:
                    update_parts.append(f'summary = {self._placeholder(len(params) + 1)}')
                    params.append(summary)
                    updated_fields.append('summary')

                # If no fields to update, return early
                if not update_parts:
                    return False, []

                # Always update the updated_at timestamp
                update_parts.append('updated_at = CURRENT_TIMESTAMP')

                # Execute update
                query = f"UPDATE context_entries SET {', '.join(update_parts)} WHERE id = {self._placeholder(len(params) + 1)}"
                params.append(context_id)
                cursor.execute(query, tuple(params))

                # Check if any rows were affected
                if cursor.rowcount > 0:
                    logger.debug(f'Updated context entry {context_id}, fields: {updated_fields}')
                    return True, updated_fields

                return False, []

            if txn:
                return _update_entry_sqlite(cast(sqlite3.Connection, txn.connection))
            return await self.backend.execute_write(_update_entry_sqlite)

        # PostgreSQL
        async def _update_entry_postgresql(conn: asyncpg.Connection) -> tuple[bool, list[str]]:
            updated_fields: list[str] = []

            # First, check if the entry exists
            row = await conn.fetchrow(
                f'SELECT id FROM context_entries WHERE id = {self._placeholder(1)}',
                context_id,
            )
            if not row:
                return False, []

            # Build update query dynamically based on provided fields
            update_parts: list[str] = []
            params: list[Any] = []

            if text_content is not None:
                update_parts.extend([
                    f'text_content = {self._placeholder(len(params) + 1)}',
                    f'content_hash = {self._placeholder(len(params) + 2)}',
                ])
                params.extend([text_content, compute_content_hash(text_content)])
                updated_fields.append('text_content')

            if metadata is not None:
                update_parts.append(f'metadata = {self._placeholder(len(params) + 1)}')
                params.append(metadata)
                updated_fields.append('metadata')

            if clear_summary:
                update_parts.append('summary = NULL')
                updated_fields.append('summary')
            elif summary is not None:
                update_parts.append(f'summary = {self._placeholder(len(params) + 1)}')
                params.append(summary)
                updated_fields.append('summary')

            # If no fields to update, return early
            if not update_parts:
                return False, []

            # Always update the updated_at timestamp
            update_parts.append('updated_at = CURRENT_TIMESTAMP')

            # Execute update
            query = f"UPDATE context_entries SET {', '.join(update_parts)} WHERE id = {self._placeholder(len(params) + 1)}"
            params.append(context_id)
            result = await conn.execute(query, *params)

            # Check if any rows were affected (asyncpg returns "UPDATE N")
            rows_affected = int(result.split()[-1]) if result else 0
            if rows_affected > 0:
                logger.debug(f'Updated context entry {context_id}, fields: {updated_fields}')
                return True, updated_fields

            return False, []

        if txn:
            return await _update_entry_postgresql(cast('asyncpg.Connection', txn.connection))
        return await self.backend.execute_write(_update_entry_postgresql)

    async def check_entry_exists(self, context_id: int) -> tuple[bool, str | None]:
        """Check if a context entry exists and return its source.

        Args:
            context_id: ID of the context entry

        Returns:
            Tuple of (exists, source). Source is None when entry does not exist.
            When exists=True, source is always 'user' or 'agent'.
        """
        if self.backend.backend_type == 'sqlite':

            def _check_exists_sqlite(conn: sqlite3.Connection) -> tuple[bool, str | None]:
                cursor = conn.cursor()
                cursor.execute(
                    f'SELECT source FROM context_entries WHERE id = {self._placeholder(1)} LIMIT 1',
                    (context_id,),
                )
                row = cursor.fetchone()
                if row is None:
                    return False, None
                return True, cast(str, row['source'])

            return await self.backend.execute_read(_check_exists_sqlite)

        # PostgreSQL
        async def _check_exists_postgresql(conn: asyncpg.Connection) -> tuple[bool, str | None]:
            row = await conn.fetchrow(
                f'SELECT source FROM context_entries WHERE id = {self._placeholder(1)} LIMIT 1',
                context_id,
            )
            if row is None:
                return False, None
            return True, cast(str, row['source'])

        return await self.backend.execute_read(_check_exists_postgresql)

    async def get_content_type(self, context_id: int) -> str | None:
        """Get the content type of a context entry.

        Args:
            context_id: ID of the context entry

        Returns:
            Content type ('text' or 'multimodal') or None if entry doesn't exist
        """
        if self.backend.backend_type == 'sqlite':

            def _get_content_type_sqlite(conn: sqlite3.Connection) -> str | None:
                cursor = conn.cursor()
                cursor.execute(
                    f'SELECT content_type FROM context_entries WHERE id = {self._placeholder(1)}',
                    (context_id,),
                )
                row = cursor.fetchone()
                return row['content_type'] if row else None

            return await self.backend.execute_read(_get_content_type_sqlite)

        # PostgreSQL
        async def _get_content_type_postgresql(conn: asyncpg.Connection) -> str | None:
            row = await conn.fetchrow(
                f'SELECT content_type FROM context_entries WHERE id = {self._placeholder(1)}',
                context_id,
            )
            return row['content_type'] if row else None

        return await self.backend.execute_read(_get_content_type_postgresql)

    async def update_content_type(
        self,
        context_id: int,
        content_type: str,
        txn: TransactionContext | None = None,
    ) -> bool:
        """Update the content type of a context entry.

        Args:
            context_id: ID of the context entry
            content_type: New content type ('text' or 'multimodal')
            txn: Optional transaction context for atomic multi-repository operations.
                When provided, uses the transaction's connection directly.
                When None, uses execute_write() for standalone operation.

        Returns:
            True if updated successfully, False otherwise
        """
        backend_type = txn.backend_type if txn else self.backend.backend_type

        if backend_type == 'sqlite':

            def _update_content_type_sqlite(conn: sqlite3.Connection) -> bool:
                cursor = conn.cursor()
                content_type_placeholder = self._placeholder(1)
                id_placeholder = self._placeholder(2)
                query = (
                    f'UPDATE context_entries SET content_type = {content_type_placeholder}, '
                    f'updated_at = CURRENT_TIMESTAMP WHERE id = {id_placeholder}'
                )
                cursor.execute(query, (content_type, context_id))
                return cursor.rowcount > 0

            if txn:
                return _update_content_type_sqlite(cast(sqlite3.Connection, txn.connection))
            return await self.backend.execute_write(_update_content_type_sqlite)

        # PostgreSQL
        async def _update_content_type_postgresql(conn: asyncpg.Connection) -> bool:
            content_type_placeholder = self._placeholder(1)
            id_placeholder = self._placeholder(2)
            query = (
                f'UPDATE context_entries SET content_type = {content_type_placeholder}, '
                f'updated_at = CURRENT_TIMESTAMP WHERE id = {id_placeholder}'
            )
            result = await conn.execute(query, content_type, context_id)
            # asyncpg returns "UPDATE N" where N is the count
            return int(result.split()[-1]) > 0 if result else False

        if txn:
            return await _update_content_type_postgresql(cast('asyncpg.Connection', txn.connection))
        return await self.backend.execute_write(_update_content_type_postgresql)

    async def patch_metadata(
        self,
        context_id: int,
        patch: dict[str, Any],
        txn: TransactionContext | None = None,
    ) -> tuple[bool, list[str]]:
        """Apply RFC 7396 JSON Merge Patch to metadata atomically.

        This method performs a partial update of the metadata field using database-native
        JSON patching functions for atomic, race-condition-free operations.

        RFC 7396 JSON Merge Patch Semantics:
        - New keys in patch are ADDED to existing metadata
        - Existing keys are REPLACED with new values
        - Keys with null values are DELETED from metadata

        IMPORTANT LIMITATIONS (RFC 7396):
        - Cannot set a value to null: null always means DELETE. If you need to store
          null values, use the full metadata replacement (metadata parameter) instead.
        - Array operations are replace-only: Arrays are replaced entirely, not merged.
          Individual array elements cannot be added, removed, or modified - the entire
          array is replaced with the new value.
        - Empty patch {} is a no-op for data but still updates the updated_at timestamp.

        Backend-specific implementation:
        - SQLite: Uses json_patch() function (available in SQLite 3.38.0+)
        - PostgreSQL: Uses custom jsonb_merge_patch() function for TRUE recursive deep merge.
          The function is created by migration app/migrations/add_jsonb_merge_patch_postgresql.sql
          and provides identical RFC 7396 semantics to SQLite's json_patch().

        Args:
            context_id: ID of the context entry to update
            patch: Dictionary containing the merge patch to apply
            txn: Optional transaction context for atomic multi-repository operations.
                When provided, uses the transaction's connection directly.
                When None, uses execute_write() for standalone operation.

        Returns:
            Tuple of (success, list_of_updated_fields).
            Updated fields will include 'metadata' if successful.
        """
        # Convert patch dict to JSON string for database operations
        patch_json = json.dumps(patch, ensure_ascii=False)
        backend_type = txn.backend_type if txn else self.backend.backend_type

        if backend_type == 'sqlite':

            def _patch_metadata_sqlite(conn: sqlite3.Connection) -> tuple[bool, list[str]]:
                cursor = conn.cursor()

                # Verify entry exists before attempting update
                cursor.execute(
                    f'SELECT id FROM context_entries WHERE id = {self._placeholder(1)}',
                    (context_id,),
                )
                if not cursor.fetchone():
                    return False, []

                # Apply JSON Merge Patch using SQLite's json_patch() function
                # json_patch() implements RFC 7396 semantics:
                # - COALESCE ensures null metadata is treated as empty object '{}'
                # - json_patch(target, patch) merges patch into target
                # - null values in patch DELETE keys from result
                cursor.execute(
                    f'''
                    UPDATE context_entries
                    SET metadata = json_patch(COALESCE(metadata, '{{}}'), {self._placeholder(1)}),
                        updated_at = CURRENT_TIMESTAMP
                    WHERE id = {self._placeholder(2)}
                    ''',
                    (patch_json, context_id),
                )

                if cursor.rowcount > 0:
                    logger.debug(f'Patched metadata for context entry {context_id}')
                    return True, ['metadata']

                return False, []

            if txn:
                return _patch_metadata_sqlite(cast(sqlite3.Connection, txn.connection))
            return await self.backend.execute_write(_patch_metadata_sqlite)

        # PostgreSQL implementation - RFC 7396 compliant using jsonb_merge_patch() function
        async def _patch_metadata_postgresql(conn: asyncpg.Connection) -> tuple[bool, list[str]]:
            # Import settings here to avoid circular import and ensure schema is retrieved at call time
            from app.settings import get_settings

            # Verify entry exists before attempting update
            row = await conn.fetchrow(
                f'SELECT id FROM context_entries WHERE id = {self._placeholder(1)}',
                context_id,
            )
            if not row:
                return False, []

            # RFC 7396 JSON Merge Patch Implementation for PostgreSQL
            #
            # Uses the custom jsonb_merge_patch() function that implements TRUE recursive
            # deep merge semantics as specified in RFC 7396:
            # - New keys in patch are ADDED to existing metadata
            # - Existing keys are REPLACED with new values from patch
            # - Keys with null values are DELETED from metadata
            # - Nested objects are RECURSIVELY merged (not replaced like || operator)
            #
            # The jsonb_merge_patch() function is created by the migration file:
            # app/migrations/add_jsonb_merge_patch_postgresql.sql
            #
            # This approach provides identical behavior to SQLite's json_patch() function,
            # ensuring consistent RFC 7396 semantics across both backends.
            #
            # IMPORTANT: Use schema-qualified function name to ensure the function is found
            # regardless of PostgreSQL search_path configuration (critical for Supabase).
            schema = get_settings().storage.postgresql_schema
            p1 = self._placeholder(1)
            p2 = self._placeholder(2)
            result = await conn.execute(
                f'''
                UPDATE context_entries
                SET metadata = {schema}.jsonb_merge_patch(COALESCE(metadata, '{{}}'::jsonb), {p1}::jsonb),
                    updated_at = CURRENT_TIMESTAMP
                WHERE id = {p2}
                ''',
                patch_json,
                context_id,
            )

            # asyncpg returns "UPDATE N" where N is the count
            rows_affected = int(result.split()[-1]) if result else 0
            if rows_affected > 0:
                logger.debug(f'Patched metadata for context entry {context_id}')
                return True, ['metadata']

            return False, []

        if txn:
            return await _patch_metadata_postgresql(cast('asyncpg.Connection', txn.connection))
        return await self.backend.execute_write(_patch_metadata_postgresql)

    @staticmethod
    def row_to_dict(row: sqlite3.Row) -> dict[str, Any]:
        """Convert a database row to a dictionary.

        Args:
            row: SQLite Row object

        Returns:
            Dictionary representation of the row
        """
        entry = dict(row)

        # Parse JSON metadata if present
        metadata_raw = entry.get('metadata')
        if metadata_raw is not None and hasattr(metadata_raw, 'strip'):
            try:
                entry['metadata'] = json.loads(str(metadata_raw))
            except (json.JSONDecodeError, ValueError, AttributeError):
                entry['metadata'] = None

        return entry

    async def get_ids_matching_batch_criteria(
        self,
        thread_ids: list[str] | None = None,
        source: str | None = None,
        older_than_days: int | None = None,
    ) -> list[int]:
        """Return context entry IDs matching batch deletion criteria.

        Builds the same WHERE clause as delete_contexts_batch but executes
        a SELECT instead of DELETE. Used to pre-query affected IDs for
        embedding cleanup on SQLite (where vec0 virtual tables lack CASCADE).

        Args:
            thread_ids: Filter by these thread IDs
            source: Filter by source ('user' or 'agent')
            older_than_days: Filter entries older than N days

        Returns:
            List of matching context entry IDs.
        """
        if self.backend.backend_type == 'sqlite':

            def _select_ids_sqlite(conn: sqlite3.Connection) -> list[int]:
                cursor = conn.cursor()
                conditions: list[str] = []
                params: list[Any] = []

                if thread_ids:
                    placeholders = ','.join([
                        self._placeholder(len(params) + i + 1) for i in range(len(thread_ids))
                    ])
                    conditions.append(f'thread_id IN ({placeholders})')
                    params.extend(thread_ids)

                if source:
                    conditions.append(f'source = {self._placeholder(len(params) + 1)}')
                    params.append(source)

                if older_than_days is not None:
                    conditions.append(
                        f"created_at < datetime('now', {self._placeholder(len(params) + 1)})",
                    )
                    params.append(f'-{older_than_days} days')

                if not conditions:
                    return []

                where_clause = ' AND '.join(conditions)
                query = f'SELECT id FROM context_entries WHERE {where_clause}'
                cursor.execute(query, tuple(params))
                return [row[0] for row in cursor.fetchall()]

            return await self.backend.execute_read(_select_ids_sqlite)

        # PostgreSQL: CASCADE handles embedding cleanup, so this method
        # returns an empty list (caller should not need it).
        return []

    async def delete_contexts_batch(
        self,
        context_ids: list[int] | None = None,
        thread_ids: list[str] | None = None,
        source: str | None = None,
        older_than_days: int | None = None,
    ) -> tuple[int, list[str]]:
        """Delete multiple context entries by various criteria.

        At least one criterion must be provided. Criteria can be combined
        for more targeted deletion. Cascading delete removes associated
        tags and images. On PostgreSQL, embeddings are also removed via CASCADE.
        On SQLite, vec0 virtual table rows (vec_context_embeddings) are not
        covered by CASCADE and require explicit cleanup via the embedding
        repository.

        Args:
            context_ids: Specific context entry IDs to delete
            thread_ids: Delete all entries in these threads
            source: Filter by source ('user' or 'agent') - combine with other criteria
            older_than_days: Delete entries older than N days

        Returns:
            Tuple of (deleted_count, list_of_criteria_used)
        """
        criteria_used: list[str] = []

        if self.backend.backend_type == 'sqlite':

            def _delete_batch_sqlite(conn: sqlite3.Connection) -> tuple[int, list[str]]:
                cursor = conn.cursor()
                conditions: list[str] = []
                params: list[Any] = []

                if context_ids:
                    placeholders = ','.join([self._placeholder(len(params) + i + 1) for i in range(len(context_ids))])
                    conditions.append(f'id IN ({placeholders})')
                    params.extend(context_ids)
                    criteria_used.append(f'context_ids: {len(context_ids)} IDs')

                if thread_ids:
                    placeholders = ','.join([
                        self._placeholder(len(params) + i + 1) for i in range(len(thread_ids))
                    ])
                    conditions.append(f'thread_id IN ({placeholders})')
                    params.extend(thread_ids)
                    criteria_used.append(f'thread_ids: {len(thread_ids)} threads')

                if source:
                    conditions.append(f'source = {self._placeholder(len(params) + 1)}')
                    params.append(source)
                    criteria_used.append(f'source: {source}')

                if older_than_days is not None:
                    conditions.append(
                        f"created_at < datetime('now', {self._placeholder(len(params) + 1)})",
                    )
                    params.append(f'-{older_than_days} days')
                    criteria_used.append(f'older_than_days: {older_than_days}')

                if not conditions:
                    return 0, criteria_used

                where_clause = ' AND '.join(conditions)
                query = f'DELETE FROM context_entries WHERE {where_clause}'
                cursor.execute(query, tuple(params))

                deleted_count = cursor.rowcount
                logger.info(f'Batch delete: removed {deleted_count} entries using criteria: {criteria_used}')
                return deleted_count, criteria_used

            return await self.backend.execute_write(_delete_batch_sqlite)

        # PostgreSQL
        async def _delete_batch_postgresql(conn: asyncpg.Connection) -> tuple[int, list[str]]:
            conditions: list[str] = []
            params: list[Any] = []

            if context_ids:
                placeholders = ','.join([self._placeholder(len(params) + i + 1) for i in range(len(context_ids))])
                conditions.append(f'id IN ({placeholders})')
                params.extend(context_ids)
                criteria_used.append(f'context_ids: {len(context_ids)} IDs')

            if thread_ids:
                placeholders = ','.join([self._placeholder(len(params) + i + 1) for i in range(len(thread_ids))])
                conditions.append(f'thread_id IN ({placeholders})')
                params.extend(thread_ids)
                criteria_used.append(f'thread_ids: {len(thread_ids)} threads')

            if source:
                conditions.append(f'source = {self._placeholder(len(params) + 1)}')
                params.append(source)
                criteria_used.append(f'source: {source}')

            if older_than_days is not None:
                conditions.append(
                    f"created_at < (NOW() - INTERVAL '{older_than_days} days')",
                )
                criteria_used.append(f'older_than_days: {older_than_days}')

            if not conditions:
                return 0, criteria_used

            where_clause = ' AND '.join(conditions)
            query = f'DELETE FROM context_entries WHERE {where_clause}'
            result = await conn.execute(query, *params)

            # asyncpg returns "DELETE N" where N is the count
            deleted_count = int(result.split()[-1]) if result else 0
            logger.info(f'Batch delete: removed {deleted_count} entries using criteria: {criteria_used}')
            return deleted_count, criteria_used

        return await self.backend.execute_write(_delete_batch_postgresql, validate_connection=True)
