PostgresStore.cs

using Npgsql;
using NpgsqlTypes;
using System;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;

namespace Mythosia.VectorDb.Postgres;

/// <summary>
/// PostgreSQL (pgvector) implementation of <see cref="IVectorStore"/>.
/// Uses a single shared table with a <c>metadata</c> JSONB column for all filtering
/// including logical isolation via metadata conditions.
/// </summary>
public class PostgresStore : IVectorStore, IDisposable
{
    private readonly PostgresOptions _options;
    private readonly string _qualifiedTable;
    private readonly SemaphoreSlim _schemaLock = new SemaphoreSlim(1, 1);
    private volatile bool _schemaEnsured;

    /// <summary>
    /// Creates a new <see cref="PostgresStore"/>.
    /// </summary>
    /// <param name="options">Configuration options. Validated on construction.</param>
    public PostgresStore(PostgresOptions options)
    {
        options.Validate();
        _options = options;
        _qualifiedTable = $"\"{_options.SchemaName}\".\"{_options.TableName}\"";
    }

    #region IVectorStore — Upsert

    public async Task UpsertAsync(VectorRecord record, CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        using var conn = await OpenConnectionAsync(cancellationToken);
        using var cmd = conn.CreateCommand();
        cmd.CommandText = BuildUpsertSql();
        AddUpsertParameters(cmd.Parameters, record);
        await cmd.ExecuteNonQueryAsync(cancellationToken);
    }

    public async Task UpsertBatchAsync(IEnumerable<VectorRecord> records, CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        using var conn = await OpenConnectionAsync(cancellationToken);
        await using var tx = await conn.BeginTransactionAsync(cancellationToken);
        using var batch = new NpgsqlBatch(conn) { Transaction = tx };

        foreach (var record in records)
        {
            cancellationToken.ThrowIfCancellationRequested();

            var cmd = new NpgsqlBatchCommand(BuildUpsertSql());
            AddUpsertParameters(cmd.Parameters, record);
            batch.BatchCommands.Add(cmd);
        }

        if (batch.BatchCommands.Count > 0)
            await batch.ExecuteNonQueryAsync(cancellationToken);

        await tx.CommitAsync(cancellationToken);
    }

    #endregion

    #region IVectorStore — Replace

    public async Task ReplaceByFilterAsync(VectorFilter filter, IReadOnlyList<VectorRecord> records, CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        using var conn = await OpenConnectionAsync(cancellationToken);
        await using var tx = await conn.BeginTransactionAsync(cancellationToken);

        // DELETE by filter
        var (whereClause, filterParams) = BuildFilterWhere(filter, includeMinScore: false, scoreExpression: string.Empty);

        using (var deleteCmd = conn.CreateCommand())
        {
            deleteCmd.Transaction = tx;
            deleteCmd.CommandText = $"DELETE FROM {_qualifiedTable} WHERE 1=1{whereClause}";
            foreach (var p in filterParams)
                deleteCmd.Parameters.Add(p);
            await deleteCmd.ExecuteNonQueryAsync(cancellationToken);
        }

        // INSERT new records
        if (records.Count > 0)
        {
            using var batch = new NpgsqlBatch(conn) { Transaction = tx };
            foreach (var record in records)
            {
                cancellationToken.ThrowIfCancellationRequested();
                var cmd = new NpgsqlBatchCommand(BuildUpsertSql());
                AddUpsertParameters(cmd.Parameters, record);
                batch.BatchCommands.Add(cmd);
            }
            await batch.ExecuteNonQueryAsync(cancellationToken);
        }

        await tx.CommitAsync(cancellationToken);
    }

    #endregion

    #region IVectorStore — Get / Delete

    public async Task<VectorRecord?> GetAsync(string id, VectorFilter? filter = null, CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        var (whereClause, filterParams) = filter != null
            ? BuildFilterWhere(filter, includeMinScore: false, scoreExpression: string.Empty)
            : (string.Empty, new List<NpgsqlParameter>());

        using var conn = await OpenConnectionAsync(cancellationToken);
        using var cmd = conn.CreateCommand();
        cmd.CommandText = $@"
SELECT id, content, metadata, embedding::text
FROM {_qualifiedTable}
WHERE id = @id{whereClause}";
        cmd.Parameters.AddWithValue("@id", id);
        foreach (var p in filterParams)
            cmd.Parameters.Add(p);

        using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
        if (!await reader.ReadAsync(cancellationToken))
            return null;

        return ReadRecord(reader);
    }

    public async Task DeleteAsync(string id, VectorFilter? filter = null, CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        var (whereClause, filterParams) = filter != null
            ? BuildFilterWhere(filter, includeMinScore: false, scoreExpression: string.Empty)
            : (string.Empty, new List<NpgsqlParameter>());

        using var conn = await OpenConnectionAsync(cancellationToken);
        using var cmd = conn.CreateCommand();
        cmd.CommandText = $"DELETE FROM {_qualifiedTable} WHERE id = @id{whereClause}";
        cmd.Parameters.AddWithValue("@id", id);
        foreach (var p in filterParams)
            cmd.Parameters.Add(p);
        await cmd.ExecuteNonQueryAsync(cancellationToken);
    }

    public async Task DeleteByFilterAsync(VectorFilter filter, CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        var (whereClause, parameters) = BuildFilterWhere(filter, includeMinScore: false, scoreExpression: string.Empty);

        using var conn = await OpenConnectionAsync(cancellationToken);
        using var cmd = conn.CreateCommand();
        cmd.CommandText = $"DELETE FROM {_qualifiedTable} WHERE 1=1{whereClause}";
        foreach (var p in parameters)
            cmd.Parameters.Add(p);

        await cmd.ExecuteNonQueryAsync(cancellationToken);
    }

    #endregion

    #region IVectorStore — Get Batch

    public async Task<IReadOnlyList<VectorRecord>> GetBatchAsync(
        IEnumerable<string> ids,
        VectorFilter? filter = null,
        CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        var idList = ids.ToList();
        if (idList.Count == 0)
            return Array.Empty<VectorRecord>();

        var (whereClause, filterParams) = filter != null
            ? BuildFilterWhere(filter, includeMinScore: false, scoreExpression: string.Empty)
            : (string.Empty, new List<NpgsqlParameter>());

        using var conn = await OpenConnectionAsync(cancellationToken);
        using var cmd = conn.CreateCommand();
        cmd.CommandText = $@"
SELECT id, content, metadata, embedding::text
FROM {_qualifiedTable}
WHERE id = ANY(@ids){whereClause}";
        cmd.Parameters.Add(new NpgsqlParameter("@ids", NpgsqlDbType.Array | NpgsqlDbType.Text) { Value = idList.ToArray() });
        foreach (var p in filterParams)
            cmd.Parameters.Add(p);

        var results = new List<VectorRecord>();
        using var reader = await cmd.ExecuteReaderAsync(cancellationToken);
        while (await reader.ReadAsync(cancellationToken))
            results.Add(ReadRecord(reader));

        return results;
    }

    #endregion

    #region IVectorStore — Search

    public async Task<IReadOnlyList<VectorSearchResult>> SearchAsync(
        float[] queryVector,
        int topK = 5,
        VectorFilter? filter = null,
        CancellationToken cancellationToken = default)
        => await SearchAsync(queryVector, topK, filter, runtimeOptions: null, cancellationToken);

    /// <summary>
    /// Performs a similarity search with optional per-request runtime tuning.
    /// </summary>
    public async Task<IReadOnlyList<VectorSearchResult>> SearchAsync(
        float[] queryVector,
        int topK,
        VectorFilter? filter,
        VectorSearchRuntimeOptions? runtimeOptions,
        CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        var distanceOperator = GetDistanceOperator();
        var scoreExpression = GetScoreExpression();

        var (whereClause, filterParams) = filter != null
            ? BuildFilterWhere(filter, includeMinScore: true, scoreExpression)
            : ("", new List<NpgsqlParameter>());

        using var conn = await OpenConnectionAsync(cancellationToken);
        await using var tx = await conn.BeginTransactionAsync(cancellationToken);

        await ApplySearchRuntimeSettingsAsync(conn, tx, runtimeOptions, cancellationToken);

        var results = new List<VectorSearchResult>();

        using (var cmd = conn.CreateCommand())
        {
            cmd.Transaction = tx;

            cmd.CommandText = $@"
SELECT id, content, metadata, embedding::text,
       {scoreExpression} AS score
FROM {_qualifiedTable}
WHERE 1=1{whereClause}
ORDER BY embedding {distanceOperator} @q::vector
LIMIT @topK";

            cmd.Parameters.AddWithValue("@q", VectorToString(queryVector));
            cmd.Parameters.AddWithValue("@topK", topK);

            foreach (var p in filterParams)
                cmd.Parameters.Add(p);

            using (var reader = await cmd.ExecuteReaderAsync(cancellationToken))
            {
                while (await reader.ReadAsync(cancellationToken))
                {
                    var record = ReadRecord(reader);
                    var score = reader.GetDouble(reader.GetOrdinal("score"));
                    results.Add(new VectorSearchResult(record, score));
                }
            }
        }

        await tx.CommitAsync(cancellationToken);

        return results;
    }

    #endregion

    #region IVectorStore — Hybrid Search

    /// <summary>
    /// Performs a hybrid search combining dense vector similarity with PostgreSQL full-text search (tsvector/tsquery).
    /// Both searches run in parallel, then results are merged using Reciprocal Rank Fusion (RRF).
    /// </summary>
    public async Task<IReadOnlyList<VectorSearchResult>> HybridSearchAsync(
        float[] denseVector,
        string query,
        int topK,
        VectorFilter? filter = null,
        CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        var distanceOperator = GetDistanceOperator();
        var expandedTopK = topK * 2;
        const int rrfK = 60;
        const float vectorWeight = 0.5f;
        const float keywordWeight = 0.5f;

        var (whereClause, filterParams) = filter != null
            ? BuildFilterWhere(filter, includeMinScore: false, string.Empty)
            : ("", new List<NpgsqlParameter>());

        var (bm25WhereClause, bm25FilterParams) = filter != null
            ? BuildFilterWhere(filter, includeMinScore: false, scoreExpression: string.Empty)
            : ("", new List<NpgsqlParameter>());

        using var conn = await OpenConnectionAsync(cancellationToken);
        await using var tx = await conn.BeginTransactionAsync(cancellationToken);

        if (_options.TextSearchMode == TextSearchMode.Trigram)
        {
            using (var thresholdCmd = conn.CreateCommand())
            {
                thresholdCmd.Transaction = tx;
                thresholdCmd.CommandText = "SET LOCAL pg_trgm.word_similarity_threshold = 0.1";
                await thresholdCmd.ExecuteNonQueryAsync(cancellationToken);
            }
        }

        using (var cmd = conn.CreateCommand())
        {
            cmd.Transaction = tx;
            cmd.CommandText = $@"
WITH vector_candidates AS (
    SELECT id, content, metadata, embedding::text AS embedding
    FROM {_qualifiedTable}
    WHERE 1=1{whereClause}
    ORDER BY embedding {distanceOperator} @q::vector
    LIMIT @candidateTopK
),
vector_results AS (
    SELECT id, content, metadata, embedding,
           ROW_NUMBER() OVER () AS rank_idx
    FROM vector_candidates
),
{BuildTextCandidatesCte(bm25WhereClause)},
text_results AS (
    SELECT id, content, metadata, embedding,
           ROW_NUMBER() OVER (ORDER BY text_score DESC) AS rank_idx
    FROM text_candidates
),
rrf_scores AS (
    SELECT id,
           @vectorWeight * (1.0 / (@rrfK + rank_idx)) AS rrf_score
    FROM vector_results
    UNION ALL
    SELECT id,
           @keywordWeight * (1.0 / (@rrfK + rank_idx)) AS rrf_score
    FROM text_results
),
final_scores AS (
    SELECT id, SUM(rrf_score) * (@rrfK + 1) AS score
    FROM rrf_scores
    GROUP BY id
    ORDER BY score DESC
    LIMIT @topK
),
base_records AS (
    SELECT id, content, metadata, embedding FROM vector_results
    UNION
    SELECT id, content, metadata, embedding FROM text_results
)
SELECT br.id, br.content, br.metadata, br.embedding,
       fs.score
FROM final_scores fs
JOIN base_records br ON br.id = fs.id
ORDER BY fs.score DESC";

            cmd.Parameters.AddWithValue("@q", VectorToString(denseVector));
            cmd.Parameters.AddWithValue("@query", query);
            cmd.Parameters.AddWithValue("@candidateTopK", expandedTopK);
            cmd.Parameters.AddWithValue("@topK", topK);
            cmd.Parameters.AddWithValue("@rrfK", rrfK);
            cmd.Parameters.AddWithValue("@vectorWeight", vectorWeight);
            cmd.Parameters.AddWithValue("@keywordWeight", keywordWeight);

            foreach (var p in filterParams)
                cmd.Parameters.Add(p);

            foreach (var p in bm25FilterParams)
            {
                if (!cmd.Parameters.Contains(p.ParameterName))
                    cmd.Parameters.Add(p);
            }

            var results = new List<VectorSearchResult>();

            using (var reader = await cmd.ExecuteReaderAsync(cancellationToken))
            {
                while (await reader.ReadAsync(cancellationToken))
                {
                    var record = ReadRecord(reader);
                    var score = reader.GetDouble(reader.GetOrdinal("score"));
                    results.Add(new VectorSearchResult(record, score));
                }
            }

            await tx.CommitAsync(cancellationToken);

            return results;
        }
    }

    #endregion

    #region Schema Management

    private async Task EnsureSchemaIfNeededAsync(CancellationToken cancellationToken)
    {
        if (_schemaEnsured)
            return;

        await _schemaLock.WaitAsync(cancellationToken);
        try
        {
            if (_schemaEnsured)
                return;

            // Always migrate legacy schema if detected, regardless of EnsureSchema setting
            using (var conn = await OpenConnectionAsync(cancellationToken))
            {
                await MigrateLegacySchemaAsync(conn, cancellationToken);
            }

            if (_options.EnsureSchema)
            {
                await CreateSchemaAsync(cancellationToken);
            }
            else
            {
                await VerifySchemaExistsAsync(cancellationToken);
            }

            _schemaEnsured = true;
        }
        finally
        {
            _schemaLock.Release();
        }
    }

    private async Task CreateSchemaAsync(CancellationToken cancellationToken)
    {
        using var conn = await OpenConnectionAsync(cancellationToken);

        var sql = $@"
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE IF NOT EXISTS {_qualifiedTable} (
    id          text        NOT NULL PRIMARY KEY,
    content     text        NULL,
    content_tsv tsvector    NOT NULL DEFAULT ''::tsvector,
    metadata    jsonb       NOT NULL DEFAULT '{{}}'::jsonb,
    embedding   vector({_options.Dimension}) NOT NULL,
    created_at  timestamptz NOT NULL DEFAULT now(),
    updated_at  timestamptz NOT NULL DEFAULT now()
);

COMMENT ON COLUMN {_qualifiedTable}.content IS 'Nullable to support customer policies that prohibit storing original text while still allowing hybrid search via content_tsv.';

CREATE INDEX IF NOT EXISTS idx_{_options.TableName}_metadata
    ON {_qualifiedTable} USING gin (metadata);

CREATE INDEX IF NOT EXISTS idx_{_options.TableName}_content_tsv
    ON {_qualifiedTable} USING gin (content_tsv);
";
        using var cmd = conn.CreateCommand();
        cmd.CommandText = sql;
        await cmd.ExecuteNonQueryAsync(cancellationToken);

        await TryCreateVectorIndexAsync(conn, cancellationToken);
        await TryCreateTrigramIndexAsync(conn, cancellationToken);
    }

    /// <summary>
    /// Automatically migrates a legacy schema (v10.6.x and earlier) where <c>namespace</c> and <c>scope</c>
    /// were stored as dedicated columns. Merges their values into <c>metadata</c> JSONB,
    /// resolves any duplicate IDs across namespaces by prefixing with the namespace value,
    /// changes the PK to <c>(id)</c>, and drops the legacy columns.
    /// All operations run in a single transaction — on failure, the schema remains unchanged.
    /// </summary>
    private async Task MigrateLegacySchemaAsync(NpgsqlConnection conn, CancellationToken cancellationToken)
    {
        bool hasLegacySchema;
        using (var checkCmd = conn.CreateCommand())
        {
            checkCmd.CommandText = @"
SELECT 1 FROM information_schema.columns
WHERE table_schema = @schema AND table_name = @table AND column_name = 'namespace'";
            checkCmd.Parameters.AddWithValue("@schema", _options.SchemaName);
            checkCmd.Parameters.AddWithValue("@table", _options.TableName);
            hasLegacySchema = await checkCmd.ExecuteScalarAsync(cancellationToken) != null;
        }

        if (!hasLegacySchema)
            return;

        await using var tx = await conn.BeginTransactionAsync(cancellationToken);

        // 1. Merge namespace/scope values into metadata JSONB
        using (var cmd = conn.CreateCommand())
        {
            cmd.Transaction = tx;
            cmd.CommandText = $@"
UPDATE {_qualifiedTable}
SET metadata = metadata
    || jsonb_build_object('namespace', namespace)
    || CASE WHEN scope IS NOT NULL THEN jsonb_build_object('scope', scope) ELSE '{{}}'::jsonb END";
            await cmd.ExecuteNonQueryAsync(cancellationToken);
        }

        // 2. Resolve duplicate IDs across namespaces by prefixing with namespace
        using (var cmd = conn.CreateCommand())
        {
            cmd.Transaction = tx;
            cmd.CommandText = $@"
UPDATE {_qualifiedTable} t
SET id = t.namespace || ':' || t.id
FROM (SELECT id FROM {_qualifiedTable} GROUP BY id HAVING COUNT(*) > 1) dups
WHERE t.id = dups.id";
            await cmd.ExecuteNonQueryAsync(cancellationToken);
        }

        // 3. Look up existing PK constraint name
        string? pkConstraintName;
        using (var cmd = conn.CreateCommand())
        {
            cmd.Transaction = tx;
            cmd.CommandText = @"
SELECT constraint_name FROM information_schema.table_constraints
WHERE table_schema = @schema AND table_name = @table AND constraint_type = 'PRIMARY KEY'";
            cmd.Parameters.AddWithValue("@schema", _options.SchemaName);
            cmd.Parameters.AddWithValue("@table", _options.TableName);
            pkConstraintName = (string?)await cmd.ExecuteScalarAsync(cancellationToken);
        }

        // 4. Drop old PK and create new PK
        if (pkConstraintName != null)
        {
            using var cmd = conn.CreateCommand();
            cmd.Transaction = tx;
            cmd.CommandText = $@"
ALTER TABLE {_qualifiedTable} DROP CONSTRAINT ""{pkConstraintName}"";
ALTER TABLE {_qualifiedTable} ADD PRIMARY KEY (id)";
            await cmd.ExecuteNonQueryAsync(cancellationToken);
        }

        // 5. Drop legacy columns
        using (var cmd = conn.CreateCommand())
        {
            cmd.Transaction = tx;
            cmd.CommandText = $@"
ALTER TABLE {_qualifiedTable} DROP COLUMN namespace;
ALTER TABLE {_qualifiedTable} DROP COLUMN scope";
            await cmd.ExecuteNonQueryAsync(cancellationToken);
        }

        // 6. Drop old namespace+scope index
        using (var cmd = conn.CreateCommand())
        {
            cmd.Transaction = tx;
            cmd.CommandText = $"DROP INDEX IF EXISTS idx_{_options.TableName}_ns_scope";
            await cmd.ExecuteNonQueryAsync(cancellationToken);
        }

        await tx.CommitAsync(cancellationToken);
    }

    private async Task TryCreateVectorIndexAsync(NpgsqlConnection conn, CancellationToken cancellationToken)
    {
        if (_options.Index is NoIndexOptions)
            return;

        try
        {
            using var cmd = conn.CreateCommand();
            var operatorClass = GetVectorOperatorClass();
            cmd.CommandText = _options.Index switch
            {
                HnswIndexOptions hnsw => $@"
CREATE INDEX IF NOT EXISTS idx_{_options.TableName}_embedding
    ON {_qualifiedTable} USING hnsw (embedding {operatorClass}) WITH (m = {hnsw.M}, ef_construction = {hnsw.EfConstruction})",
                IvfFlatIndexOptions ivfFlat => $@"
CREATE INDEX IF NOT EXISTS idx_{_options.TableName}_embedding
    ON {_qualifiedTable} USING ivfflat (embedding {operatorClass}) WITH (lists = {ivfFlat.Lists})",
                _ => throw new InvalidOperationException($"Unsupported index options type: {_options.Index.GetType().Name}")
            };

            await cmd.ExecuteNonQueryAsync(cancellationToken);
        }
        catch when (!_options.FailFastOnIndexCreationFailure)
        {
            // Non-fail-fast mode: allow startup to continue even if index creation fails.
        }
    }

    private async Task TryCreateTrigramIndexAsync(NpgsqlConnection conn, CancellationToken cancellationToken)
    {
        if (_options.TextSearchMode != TextSearchMode.Trigram)
            return;

        try
        {
            using var cmd = conn.CreateCommand();
            cmd.CommandText = $@"
CREATE EXTENSION IF NOT EXISTS pg_trgm;

CREATE INDEX IF NOT EXISTS idx_{_options.TableName}_content_trgm
    ON {_qualifiedTable} USING gin (content gin_trgm_ops);";
            await cmd.ExecuteNonQueryAsync(cancellationToken);
        }
        catch when (!_options.FailFastOnIndexCreationFailure)
        {
            // pg_trgm extension may not be available on all PostgreSQL installations.
        }
    }

    private async Task VerifySchemaExistsAsync(CancellationToken cancellationToken)
    {
        using var conn = await OpenConnectionAsync(cancellationToken);
        using var cmd = conn.CreateCommand();
        cmd.CommandText = @"
SELECT 1
FROM information_schema.tables
WHERE table_schema = @schema AND table_name = @table";
        cmd.Parameters.AddWithValue("@schema", _options.SchemaName);
        cmd.Parameters.AddWithValue("@table", _options.TableName);

        var result = await cmd.ExecuteScalarAsync(cancellationToken);
        if (result == null)
        {
            throw new InvalidOperationException(
                $"Table \"{_options.SchemaName}\".\"{_options.TableName}\" does not exist. " +
                $"Create the table manually or set EnsureSchema = true for automatic provisioning.");
        }
    }

    #endregion

    #region IVectorStore — Count

    public async Task<long> CountAsync(VectorFilter? filter = null, CancellationToken cancellationToken = default)
    {
        await EnsureSchemaIfNeededAsync(cancellationToken);

        using var conn = await OpenConnectionAsync(cancellationToken);
        using var cmd = conn.CreateCommand();

        var sb = new StringBuilder($"SELECT COUNT(*) FROM {_qualifiedTable} WHERE 1=1");

        if (filter != null)
        {
            var (whereClause, filterParams) = BuildFilterWhere(filter, includeMinScore: false, string.Empty);
            sb.Append(whereClause);
            foreach (var p in filterParams)
                cmd.Parameters.Add(p);
        }

        cmd.CommandText = sb.ToString();
        var result = await cmd.ExecuteScalarAsync(cancellationToken);
        return Convert.ToInt64(result);
    }

    #endregion

    #region Connection Verification

    public async Task VerifyConnectionAsync(CancellationToken cancellationToken = default)
    {
        using var conn = await OpenConnectionAsync(cancellationToken);
    }

    #endregion

    #region Private Helpers

    private async Task<NpgsqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
    {
        var conn = new NpgsqlConnection(_options.ConnectionString);
        await conn.OpenAsync(cancellationToken);
        return conn;
    }

    private string BuildUpsertSql()
    {
        return $@"
INSERT INTO {_qualifiedTable} (id, content, content_tsv, metadata, embedding)
VALUES (@id, @content, to_tsvector('{_options.TextSearchConfig}', coalesce(@content_normalized, '')), @metadata, @embedding::vector)
ON CONFLICT (id) DO UPDATE SET
    content    = EXCLUDED.content,
    content_tsv = EXCLUDED.content_tsv,
    metadata   = EXCLUDED.metadata,
    embedding  = EXCLUDED.embedding,
    updated_at = now()";
    }

    private static void AddUpsertParameters(NpgsqlParameterCollection parameters, VectorRecord record)
    {
        parameters.AddWithValue("@id", record.Id);
        parameters.AddWithValue("@content", (object?)record.Content ?? DBNull.Value);
        parameters.AddWithValue("@content_normalized", (object?)NormalizeScriptBoundaries(record.Content) ?? DBNull.Value);
        parameters.Add(CreateJsonbParameter("@metadata", record.Metadata));
        parameters.AddWithValue("@embedding", VectorToString(record.Vector));
    }

    private static (string whereClause, List<NpgsqlParameter> parameters) BuildFilterWhere(
        VectorFilter filter,
        bool includeMinScore,
        string scoreExpression)
    {
        var sb = new StringBuilder();
        var parameters = new List<NpgsqlParameter>();
        int paramIdx = 0;

        if (filter.Conditions.Count > 0)
        {
            sb.Append(" AND ");
            AppendConditionGroup(sb, parameters, filter.Conditions, FilterLogic.And, ref paramIdx);
        }

        if (includeMinScore && filter.MinScore.HasValue)
        {
            sb.Append($" AND ({scoreExpression}) >= @min_score");
            parameters.Add(new NpgsqlParameter("@min_score", filter.MinScore.Value));
        }

        return (sb.ToString(), parameters);
    }

    private static void AppendConditionGroup(
        StringBuilder sb,
        List<NpgsqlParameter> parameters,
        IReadOnlyList<FilterCondition> conditions,
        FilterLogic logic,
        ref int paramIdx)
    {
        var joiner = logic == FilterLogic.And ? " AND " : " OR ";
        sb.Append('(');
        for (int i = 0; i < conditions.Count; i++)
        {
            if (i > 0) sb.Append(joiner);
            AppendSingleCondition(sb, parameters, conditions[i], ref paramIdx);
        }
        sb.Append(')');
    }

    private static void AppendSingleCondition(
        StringBuilder sb,
        List<NpgsqlParameter> parameters,
        FilterCondition condition,
        ref int paramIdx)
    {
        if (condition is MetadataCondition mc)
        {
            AppendMetadataCondition(sb, parameters, mc, ref paramIdx);
        }
        else if (condition is FilterGroup group)
        {
            AppendConditionGroup(sb, parameters, group.Conditions, group.Logic, ref paramIdx);
        }
    }

    private static void AppendMetadataCondition(
        StringBuilder sb,
        List<NpgsqlParameter> parameters,
        MetadataCondition mc,
        ref int paramIdx)
    {
        int idx = paramIdx++;
        var pKey = $"@mf_k{idx}";
        var pVal = $"@mf_v{idx}";

        switch (mc.Operator)
        {
            case FilterOperator.Eq:
            {
                // JSONB containment: uses GIN index and safely embeds key via JsonSerializer
                var json = JsonSerializer.Serialize(
                    new System.Collections.Generic.Dictionary<string, string> { { mc.Key, mc.Value! } });
                sb.Append($"metadata @> {pVal}");
                parameters.Add(CreateJsonbParameter(pVal, json));
                break;
            }
            case FilterOperator.Ne:
                // NULL from missing key propagates as false in WHERE — records without the key do not match
                sb.Append($"metadata->>{pKey} != {pVal}");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                parameters.Add(new NpgsqlParameter(pVal, mc.Value!));
                break;

            case FilterOperator.Gt:
                sb.Append($"metadata->>{pKey} > {pVal}");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                parameters.Add(new NpgsqlParameter(pVal, mc.Value!));
                break;

            case FilterOperator.Gte:
                sb.Append($"metadata->>{pKey} >= {pVal}");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                parameters.Add(new NpgsqlParameter(pVal, mc.Value!));
                break;

            case FilterOperator.Lt:
                sb.Append($"metadata->>{pKey} < {pVal}");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                parameters.Add(new NpgsqlParameter(pVal, mc.Value!));
                break;

            case FilterOperator.Lte:
                sb.Append($"metadata->>{pKey} <= {pVal}");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                parameters.Add(new NpgsqlParameter(pVal, mc.Value!));
                break;

            case FilterOperator.In:
                sb.Append($"metadata->>{pKey} = ANY({pVal})");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                parameters.Add(new NpgsqlParameter(pVal, NpgsqlTypes.NpgsqlDbType.Array | NpgsqlTypes.NpgsqlDbType.Text)
                {
                    Value = ToArray(mc.Values!)
                });
                break;

            case FilterOperator.NotIn:
                sb.Append($"NOT (metadata->>{pKey} = ANY({pVal}))");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                parameters.Add(new NpgsqlParameter(pVal, NpgsqlTypes.NpgsqlDbType.Array | NpgsqlTypes.NpgsqlDbType.Text)
                {
                    Value = ToArray(mc.Values!)
                });
                break;

            case FilterOperator.Like:
                sb.Append($"metadata->>{pKey} LIKE {pVal}");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                parameters.Add(new NpgsqlParameter(pVal, mc.Value!));
                break;

            case FilterOperator.Exists:
                sb.Append($"jsonb_exists(metadata, {pKey})");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                break;

            case FilterOperator.NotExists:
                sb.Append($"NOT jsonb_exists(metadata, {pKey})");
                parameters.Add(new NpgsqlParameter(pKey, mc.Key));
                break;
        }
    }

    private static string[] ToArray(IReadOnlyList<string> values)
    {
        var arr = new string[values.Count];
        for (int i = 0; i < values.Count; i++) arr[i] = values[i];
        return arr;
    }

    /// <summary>
    /// Builds the <c>text_candidates</c> CTE for hybrid search based on <see cref="TextSearchMode"/>.
    /// </summary>
    private string BuildTextCandidatesCte(string filterWhereClause)
    {
        if (_options.TextSearchMode == TextSearchMode.Trigram)
        {
            return $@"text_candidates AS (
    SELECT id, content, metadata, embedding::text AS embedding,
           word_similarity(@query, content) AS text_score
    FROM {_qualifiedTable}
    WHERE 1=1{filterWhereClause}
      AND content IS NOT NULL
      AND length(@query) > 0
      AND @query <% content
    ORDER BY @query <%> content
    LIMIT @candidateTopK
)";
        }

        var cfg = _options.TextSearchConfig;
        return $@"text_candidates AS (
    SELECT id, content, metadata, embedding::text AS embedding,
           ts_rank(content_tsv, to_tsquery('{cfg}',
               regexp_replace(regexp_replace(trim(@query), '[^\w\s]', '', 'g'), '\s+', ' | ', 'g')
           )) AS text_score
    FROM {_qualifiedTable}
    WHERE 1=1{filterWhereClause}
      AND length(trim(@query)) > 0
      AND content_tsv @@ to_tsquery('{cfg}',
              regexp_replace(regexp_replace(trim(@query), '[^\w\s]', '', 'g'), '\s+', ' | ', 'g')
          )
    ORDER BY text_score DESC
    LIMIT @candidateTopK
)";
    }

    private static VectorRecord ReadRecord(NpgsqlDataReader reader)
    {
        var record = new VectorRecord
        {
            Id = reader.GetString(reader.GetOrdinal("id")),
            Content = reader.IsDBNull(reader.GetOrdinal("content"))
                ? string.Empty
                : reader.GetString(reader.GetOrdinal("content")),
        };

        // Read embedding as text (cast in SQL: embedding::text → '[0.1,0.2,...]')
        var embeddingText = reader.GetString(reader.GetOrdinal("embedding"));
        record.Vector = ParseVectorString(embeddingText);

        var metadataJson = reader.IsDBNull(reader.GetOrdinal("metadata"))
            ? "{}"
            : reader.GetString(reader.GetOrdinal("metadata"));
        record.Metadata = JsonSerializer.Deserialize<Dictionary<string, string>>(metadataJson)
                          ?? new Dictionary<string, string>();

        return record;
    }

    /// <summary>
    /// Converts a float array to pgvector literal format: '[0.1,0.2,0.3]'.
    /// </summary>
    public static string VectorToString(float[] values)
    {
        var sb = new StringBuilder("[");
        for (int i = 0; i < values.Length; i++)
        {
            if (i > 0) sb.Append(',');
            sb.Append(values[i].ToString("G", CultureInfo.InvariantCulture));
        }
        sb.Append(']');
        return sb.ToString();
    }

    /// <summary>
    /// Parses pgvector text representation '[0.1,0.2,0.3]' back to float[].
    /// </summary>
    public static float[] ParseVectorString(string text)
    {
        if (string.IsNullOrEmpty(text))
            return Array.Empty<float>();

        // Remove surrounding brackets
        var trimmed = text.Trim();
        if (trimmed.StartsWith("[")) trimmed = trimmed.Substring(1);
        if (trimmed.EndsWith("]")) trimmed = trimmed.Substring(0, trimmed.Length - 1);

        if (string.IsNullOrWhiteSpace(trimmed))
            return Array.Empty<float>();

        var parts = trimmed.Split(',');
        var result = new float[parts.Length];
        for (int i = 0; i < parts.Length; i++)
        {
            result[i] = float.Parse(parts[i].Trim(), CultureInfo.InvariantCulture);
        }
        return result;
    }

    /// <summary>
    /// Inserts a space at every script boundary so that PostgreSQL's
    /// <c>to_tsvector</c> tokenises mixed-script words correctly.
    /// <para>Examples: "event테이블에" → "event 테이블에", "データ테이블" → "データ 테이블"</para>
    /// </summary>
    internal static string NormalizeScriptBoundaries(string? text)
    {
        if (string.IsNullOrEmpty(text)) return text!;

        var sb = new StringBuilder(text.Length + 16);
        var prevScript = ScriptKind.Common;

        for (int i = 0; i < text.Length; i++)
        {
            var ch = text[i];
            var script = ClassifyScript(ch);

            if (script != ScriptKind.Common
                && prevScript != ScriptKind.Common
                && script != prevScript)
            {
                sb.Append(' ');
            }

            sb.Append(ch);

            if (script != ScriptKind.Common)
                prevScript = script;
        }

        return sb.ToString();
    }

    private enum ScriptKind : byte
    {
        Common,     // whitespace, punctuation, symbols, digits
        Latin,
        Hangul,
        CJK,
        Hiragana,
        Katakana,
    }

    private static ScriptKind ClassifyScript(char ch)
    {
        if ((ch >= 'A' && ch <= 'Z') || (ch >= 'a' && ch <= 'z'))
            return ScriptKind.Latin;

        // Hangul Syllables (가-힣), Jamo (ㄱ-ㅎ, ㅏ-ㅣ), Compatibility Jamo
        if ((ch >= 0xAC00 && ch <= 0xD7AF)
            || (ch >= 0x3130 && ch <= 0x318F)
            || (ch >= 0x1100 && ch <= 0x11FF))
            return ScriptKind.Hangul;

        // CJK Unified Ideographs + Extension A + Compatibility
        if ((ch >= 0x4E00 && ch <= 0x9FFF)
            || (ch >= 0x3400 && ch <= 0x4DBF)
            || (ch >= 0xF900 && ch <= 0xFAFF))
            return ScriptKind.CJK;

        // Hiragana
        if (ch >= 0x3040 && ch <= 0x309F)
            return ScriptKind.Hiragana;

        // Katakana + Katakana Phonetic Extensions
        if ((ch >= 0x30A0 && ch <= 0x30FF)
            || (ch >= 0x31F0 && ch <= 0x31FF))
            return ScriptKind.Katakana;

        // Latin Extended (accented characters: é, ñ, ü, etc.)
        if ((ch >= 0x00C0 && ch <= 0x024F))
            return ScriptKind.Latin;

        return ScriptKind.Common;
    }

    private static NpgsqlParameter CreateJsonbParameter(string name, object value)
    {
        string json = value is string s ? s : JsonSerializer.Serialize(value);
        return new NpgsqlParameter(name, NpgsqlDbType.Jsonb) { Value = json };
    }

    private async Task ApplySearchRuntimeSettingsAsync(
        NpgsqlConnection conn,
        NpgsqlTransaction tx,
        VectorSearchRuntimeOptions? runtimeOptions,
        CancellationToken cancellationToken)
    {
        var (profileIvfflatProbes, profileHnswEfSearch) = GetProfileDefaults(runtimeOptions?.Profile);

        if (_options.Index is NoIndexOptions)
            return;

        if (_options.Index is IvfFlatIndexOptions ivfFlat)
        {
            if (runtimeOptions is not null && runtimeOptions is not IvfFlatSearchRuntimeOptions)
                throw new ArgumentException("Use IvfFlatSearchRuntimeOptions when Index is IvfFlatIndexOptions.", nameof(runtimeOptions));

            var runtimeIvf = runtimeOptions as IvfFlatSearchRuntimeOptions;
            var probes = runtimeIvf?.Probes ?? profileIvfflatProbes ?? ivfFlat.Probes;

            // Block-scoped using ensures the command is fully disposed before returning,
            // preventing "A command is already in progress" errors from Npgsql.
            // SET LOCAL does not support parameterized queries ($1) in PostgreSQL.
            // The value is always a validated integer, so interpolation is safe.
            using (var cmd = conn.CreateCommand())
            {
                cmd.Transaction = tx;
                cmd.CommandText = $"SET LOCAL ivfflat.probes = {probes}";
                await cmd.ExecuteNonQueryAsync(cancellationToken);
            }
            return;
        }

        if (_options.Index is HnswIndexOptions hnsw)
        {
            if (runtimeOptions is not null && runtimeOptions is not HnswSearchRuntimeOptions)
                throw new ArgumentException("Use HnswSearchRuntimeOptions when Index is HnswIndexOptions.", nameof(runtimeOptions));

            var runtimeHnsw = runtimeOptions as HnswSearchRuntimeOptions;
            var efSearch = runtimeHnsw?.EfSearch ?? profileHnswEfSearch ?? hnsw.EfSearch;

            // Block-scoped using ensures the command is fully disposed before returning,
            // preventing "A command is already in progress" errors from Npgsql.
            // SET LOCAL does not support parameterized queries ($1) in PostgreSQL.
            // The value is always a validated integer, so interpolation is safe.
            using (var cmd = conn.CreateCommand())
            {
                cmd.Transaction = tx;
                cmd.CommandText = $"SET LOCAL hnsw.ef_search = {efSearch}";
                await cmd.ExecuteNonQueryAsync(cancellationToken);
            }
            return;
        }

        throw new InvalidOperationException($"Unsupported index options type: {_options.Index.GetType().Name}");
    }

    private static (int? ivfflatProbes, int? hnswEfSearch) GetProfileDefaults(SearchProfile? profile)
    {
        return profile switch
        {
            SearchProfile.Fast => (4, 16),
            SearchProfile.Balanced => (10, 40),
            SearchProfile.HighRecall => (32, 120),
            _ => (null, null)
        };
    }

    private string GetDistanceOperator()
    {
        return _options.DistanceStrategy switch
        {
            DistanceStrategy.Cosine => "<=>",
            DistanceStrategy.Euclidean => "<->",
            DistanceStrategy.InnerProduct => "<#>",
            _ => throw new InvalidOperationException($"Unsupported distance strategy: {_options.DistanceStrategy}")
        };
    }

    private string GetVectorOperatorClass()
    {
        return _options.DistanceStrategy switch
        {
            DistanceStrategy.Cosine => "vector_cosine_ops",
            DistanceStrategy.Euclidean => "vector_l2_ops",
            DistanceStrategy.InnerProduct => "vector_ip_ops",
            _ => throw new InvalidOperationException($"Unsupported distance strategy: {_options.DistanceStrategy}")
        };
    }

    private string GetScoreExpression()
    {
        return _options.DistanceStrategy switch
        {
            DistanceStrategy.Cosine => "1 - (embedding <=> @q::vector)",
            DistanceStrategy.Euclidean => "1 / (1 + (embedding <-> @q::vector))",
            DistanceStrategy.InnerProduct => "-(embedding <#> @q::vector)",
            _ => throw new InvalidOperationException($"Unsupported distance strategy: {_options.DistanceStrategy}")
        };
    }

    #endregion

    #region IDisposable

    public void Dispose()
    {
        _schemaLock.Dispose();
    }

    #endregion
}