db.py

"""DuckDB connection lifecycle, schema ownership, and FastAPI Depends helpers.

This project is the single READ_WRITE writer for the shared DuckDB file.
Schema creation and seeding live here; the MCP reader project only reads.
"""

from __future__ import annotations

import asyncio
import os
from contextlib import asynccontextmanager
from typing import Annotated, Any

import duckdb
from fastapi import Depends, Request

_DB_PATH_DEFAULT = "./enterprise.duckdb"

# ---------------------------------------------------------------------------
# Schema DDL
# ---------------------------------------------------------------------------

_DDL = """
CREATE SEQUENCE IF NOT EXISTS entry_seq START 1;
CREATE SEQUENCE IF NOT EXISTS tag_seq   START 1;

CREATE TABLE IF NOT EXISTS categories (
    id          INTEGER PRIMARY KEY,
    name        VARCHAR NOT NULL UNIQUE,
    description TEXT
);

CREATE TABLE IF NOT EXISTS roles (
    id   INTEGER PRIMARY KEY,
    name VARCHAR NOT NULL UNIQUE
);

CREATE TABLE IF NOT EXISTS knowledge_entries (
    id          INTEGER PRIMARY KEY DEFAULT nextval('entry_seq'),
    title       VARCHAR NOT NULL,
    content     TEXT    NOT NULL,
    category_id INTEGER NOT NULL REFERENCES categories(id),
    created_at  TIMESTAMP DEFAULT now(),
    updated_at  TIMESTAMP DEFAULT now()
);

CREATE TABLE IF NOT EXISTS entry_roles (
    entry_id INTEGER NOT NULL REFERENCES knowledge_entries(id),
    role_id  INTEGER NOT NULL REFERENCES roles(id),
    PRIMARY KEY (entry_id, role_id)
);

CREATE TABLE IF NOT EXISTS tags (
    id   INTEGER PRIMARY KEY DEFAULT nextval('tag_seq'),
    name VARCHAR NOT NULL UNIQUE
);

CREATE TABLE IF NOT EXISTS entry_tags (
    entry_id INTEGER NOT NULL REFERENCES knowledge_entries(id),
    tag_id   INTEGER NOT NULL REFERENCES tags(id),
    PRIMARY KEY (entry_id, tag_id)
);
"""

# ---------------------------------------------------------------------------
# Initial seed data
# ---------------------------------------------------------------------------

_CATEGORIES = [
    (1, "development_process", "General software development standards and processes"),
    (2, "cicd",               "CI/CD pipeline standards, scanning, and quality gates"),
    (3, "security",           "Security policies, compliance, and secrets management"),
    (4, "production_readiness", "Production readiness checklist, monitoring, and deployment"),
    (5, "git_pr",             "Git branching strategy, PR conventions, and code review"),
    (6, "architecture",       "Architecture guidelines, patterns, and technology radar"),
]

_ROLES = [
    (1, "developer"),
    (2, "devops"),
    (3, "architect"),
    (4, "manager"),
]


def _seed_lookups(con: duckdb.DuckDBPyConnection) -> None:
    """Insert categories and roles if the tables are empty."""
    if con.execute("SELECT COUNT(*) FROM categories").fetchone()[0] == 0:  # type: ignore[index]
        con.executemany(
            "INSERT INTO categories (id, name, description) VALUES (?, ?, ?)",
            _CATEGORIES,
        )
    if con.execute("SELECT COUNT(*) FROM roles").fetchone()[0] == 0:  # type: ignore[index]
        con.executemany("INSERT INTO roles (id, name) VALUES (?, ?)", _ROLES)


# ---------------------------------------------------------------------------
# FastAPI lifespan
# ---------------------------------------------------------------------------


@asynccontextmanager
async def lifespan(app: Any):
    """Open a READ_WRITE DuckDB connection for the lifetime of the server."""
    db_path = os.environ.get("ENTERPRISE_DB_PATH", _DB_PATH_DEFAULT)
    con = duckdb.connect(db_path)
    con.execute(_DDL)
    _seed_lookups(con)
    lock = asyncio.Lock()
    app.state.db = con
    app.state.db_lock = lock
    try:
        yield
    finally:
        con.close()


# ---------------------------------------------------------------------------
# Depends helpers
# ---------------------------------------------------------------------------


def get_db(request: Request) -> duckdb.DuckDBPyConnection:
    return request.app.state.db


def get_lock(request: Request) -> asyncio.Lock:
    return request.app.state.db_lock


DBDep   = Annotated[duckdb.DuckDBPyConnection, Depends(get_db)]
LockDep = Annotated[asyncio.Lock,              Depends(get_lock)]