"""{{ display_name }} MCP server — generated by mcpify. Do not hand-edit.

Single data-access layer against mcp_store.db — both the relational
`endpoints` table and the `semantic_endpoints` vec0 virtual table
(REQ-2.4.1's single-store consolidation). This module never creates
tables — mcpify's shared pipeline (Story 5/6) already wrote
mcp_store.db's schema before any generated code runs.
"""

from __future__ import annotations

import asyncio
import json
from contextlib import asynccontextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any, AsyncIterator

import aiosqlite
import sqlite_vec

_ENDPOINT_COLUMNS = (
    "operation_id, path, method, summary, description, "
    "input_schema, output_schema, auth_scheme_ref"
)

# mcpify:versions:begin
VERSION_STORE_FILES: dict[str, str] = {
{%- for v in version_entries %}
    "{{ v.label }}": "{{ v.db_file }}",
{%- endfor %}
}
# mcpify:versions:end


def resolve_store_path(api_version: str) -> Path:
    """Resolves the active `api_version` (from the config cascade) to its
    store file — the default version's data always lives in the
    project-root `mcp_store.db` this target has always used; every other
    version lives in its own `mcp_store_v<label>.db` sibling, added by a
    later `mcpify add-version` call.
    """
    file_name = VERSION_STORE_FILES.get(api_version)
    if file_name is None:
        raise ValueError(
            f"unknown api_version '{api_version}' — run the 'versions' command to see what's available"
        )
    return Path(file_name)


async def _load_vec_extension(conn: aiosqlite.Connection) -> None:
    await conn.enable_load_extension(True)
    await conn.load_extension(sqlite_vec.loadable_path())
    await conn.enable_load_extension(False)


@asynccontextmanager
async def open_store(path: Path) -> AsyncIterator[aiosqlite.Connection]:
    """Opens `mcp_store.db` read-only: this project only ever reads it,
    except `populate_embeddings` (`services/populate_embeddings.py`),
    which uses `open_store_read_write` instead.
    """
    conn = await aiosqlite.connect(f"file:{path}?mode=ro", uri=True)
    try:
        conn.row_factory = aiosqlite.Row
        await _load_vec_extension(conn)
        yield conn
    finally:
        await conn.close()


@asynccontextmanager
async def open_store_read_write(path: Path) -> AsyncIterator[aiosqlite.Connection]:
    """Read-write counterpart to `open_store` — for
    `services/populate_embeddings.py`, the one script that actually writes
    to `mcp_store.db` (backfilling `semantic_endpoints`, whose table every
    other caller only ever reads from).
    """
    conn = await aiosqlite.connect(path)
    try:
        conn.row_factory = aiosqlite.Row
        await _load_vec_extension(conn)
        yield conn
    finally:
        await conn.close()


_cached_connections: dict[str, aiosqlite.Connection] = {}
_cached_connections_lock = asyncio.Lock()


async def cached_store_connection(api_version: str) -> aiosqlite.Connection:
    """Returns a process-wide, in-memory-backed connection for
    `api_version`: on first access, opens the on-disk file read-only via
    `open_store`, copies its entire contents into a fresh `:memory:`
    connection (via SQLite's backup API), then closes the on-disk
    connection immediately — so its file handle/lock is held only for
    that brief copy, not for the process's lifetime. This means an
    external process (a fresh `populate_embeddings` run, a deployment
    replacing the file) can update `mcp_store.db` without hitting
    "database is locked" against a live server, at the cost of the
    running process only picking up such an update on its next restart.

    Unlike `open_store`, this is *not* an `@asynccontextmanager` — the
    returned connection is cached and reused for the rest of the
    process's lifetime, so callers must not close it.
    """
    async with _cached_connections_lock:
        cached = _cached_connections.get(api_version)
        if cached is not None:
            return cached

        path = resolve_store_path(api_version)
        mem_conn = await aiosqlite.connect(":memory:")
        mem_conn.row_factory = aiosqlite.Row
        await _load_vec_extension(mem_conn)

        async with open_store(path) as disk_conn:
            await disk_conn.backup(mem_conn)

        _cached_connections[api_version] = mem_conn
        return mem_conn


async def close_cached_connections() -> None:
    """Closes every connection `cached_store_connection` has handed out and
    clears the cache. `aiosqlite.Connection` runs a dedicated (non-daemon)
    worker thread per connection, so a cached connection left open past
    the last `await` in a one-shot process (`search`/`get`/`call`, and the
    real embedding e2e test) blocks CPython's interpreter shutdown
    forever waiting to join that thread — unlike the long-running
    harness server, where the OS simply kills the process, these short-
    lived callers must close what they opened. Every one-shot CLI command
    and the e2e test call this once they're done; the persistent harness
    server calls it from its own shutdown `finally` block alongside
    `shutdown_tracing`.
    """
    async with _cached_connections_lock:
        for conn in _cached_connections.values():
            await conn.close()
        _cached_connections.clear()


@dataclass
class EndpointRecord:
    operation_id: str
    path: str
    method: str
    summary: str | None
    description: str | None
    input_schema: Any
    output_schema: Any
    auth_scheme_ref: str | None


def _row_to_endpoint(row: aiosqlite.Row) -> EndpointRecord:
    return EndpointRecord(
        operation_id=row["operation_id"],
        path=row["path"],
        method=row["method"],
        summary=row["summary"],
        description=row["description"],
        input_schema=json.loads(row["input_schema"]),
        output_schema=json.loads(row["output_schema"]),
        auth_scheme_ref=row["auth_scheme_ref"],
    )


async def get_endpoint(conn: aiosqlite.Connection, operation_id: str) -> EndpointRecord | None:
    cursor = await conn.execute(
        f"SELECT {_ENDPOINT_COLUMNS} FROM endpoints WHERE operation_id = ?", (operation_id,)
    )
    row = await cursor.fetchone()
    return _row_to_endpoint(row) if row is not None else None


async def list_endpoints(conn: aiosqlite.Connection) -> list[EndpointRecord]:
    cursor = await conn.execute(f"SELECT {_ENDPOINT_COLUMNS} FROM endpoints")
    rows = await cursor.fetchall()
    return [_row_to_endpoint(row) for row in rows]


@dataclass
class SearchResult:
    operation_id: str
    summary: str | None
    similarity: float


async def search_endpoints(
    conn: aiosqlite.Connection, query_embedding: list[float], limit: int
) -> list[SearchResult]:
    """k-nearest-neighbor search over `semantic_endpoints` (sqlite-vec's
    documented `MATCH ... AND k = ...` query form). `query_embedding` must
    come from the same model as the vectors it's compared against — see
    `services.embedding_service`.

    `similarity` is a simple `1 - distance` transform of sqlite-vec's L2
    distance, not a true cosine-similarity computation; since embeddings
    are normalized (unit vectors), L2-distance ordering already matches
    cosine-similarity ordering, so result *ranking* is correct even though
    the displayed score is an approximation.
    """
    blob = sqlite_vec.serialize_float32(query_embedding)
    cursor = await conn.execute(
        "SELECT e.operation_id, e.summary, s.distance "
        "FROM semantic_endpoints s "
        "JOIN endpoints e ON e.operation_id = s.operation_id "
        "WHERE s.embedding MATCH ? AND k = ? "
        "ORDER BY s.distance",
        (blob, limit),
    )
    rows = await cursor.fetchall()
    return [
        SearchResult(
            operation_id=row["operation_id"],
            summary=row["summary"],
            similarity=1.0 - row["distance"],
        )
        for row in rows
    ]
