kanbante/app/models.py

"""
SQLite models for Kanbante — lightweight local state.

Local state only stores what Gitea does not: the column of each issue,
its position inside the column, and per-project notes. The database path
honours the KANBANTE_DB_PATH environment variable (see app.config) —
that is the path the Dockerfile / docker-compose mount their volume on.

Migrations are additive and idempotent: init_db() creates fresh schemas
and upgrades older databases in place (position column, notes
updated_at, cleanup of rows carrying an unknown column name).
"""

import os
import sqlite3
from contextlib import contextmanager
from typing import Iterator

KANBAN_COLUMNS = ["backlog", "todo", "in_progress", "review", "done"]


def db_path() -> str:
    """Resolve the database path from the environment (per call)."""
    return os.path.expanduser(
        os.getenv("KANBANTE_DB_PATH", "~/.kanbante/kanbante.db")
    )


@contextmanager
def get_db() -> Iterator[sqlite3.Connection]:
    path = db_path()
    directory = os.path.dirname(path)
    if directory:
        os.makedirs(directory, exist_ok=True)
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()


def init_db() -> None:
    with get_db() as conn:
        conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS board_state (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                repo_full_name TEXT NOT NULL,
                issue_number INTEGER NOT NULL,
                column_name TEXT NOT NULL DEFAULT 'backlog',
                position REAL NOT NULL DEFAULT 0,
                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
                UNIQUE(repo_full_name, issue_number)
            );

            CREATE TABLE IF NOT EXISTS project_notes (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                repo_full_name TEXT NOT NULL,
                note TEXT NOT NULL,
                created_at TEXT NOT NULL DEFAULT (datetime('now')),
                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
                UNIQUE(repo_full_name)
            );

            CREATE INDEX IF NOT EXISTS idx_board_repo
                ON board_state(repo_full_name, column_name);
            """
        )
        _migrate(conn)


def _migrate(conn: sqlite3.Connection) -> None:
    """Upgrade schemas created by Kanbante <= 0.1.0."""
    board_cols = [r[1] for r in conn.execute("PRAGMA table_info(board_state)")]
    if board_cols and "position" not in board_cols:
        conn.execute(
            "ALTER TABLE board_state ADD COLUMN position REAL NOT NULL DEFAULT 0"
        )
    notes_cols = [r[1] for r in conn.execute("PRAGMA table_info(project_notes)")]
    if notes_cols and "updated_at" not in notes_cols:
        # SQLite rejects non-constant defaults in ADD COLUMN: add the
        # column plainly, then backfill from created_at.
        conn.execute("ALTER TABLE project_notes ADD COLUMN updated_at TEXT")
        conn.execute(
            "UPDATE project_notes SET updated_at = created_at "
            "WHERE updated_at IS NULL"
        )
    # Rows written under a column name that no longer exists used to crash
    # get_board_state() with a KeyError. Drop them once, here.
    placeholders = ",".join("?" for _ in KANBAN_COLUMNS)
    conn.execute(
        f"DELETE FROM board_state WHERE column_name NOT IN ({placeholders})",
        KANBAN_COLUMNS,
    )


def ping_db() -> bool:
    """Cheap check used by /api/readiness."""
    try:
        with get_db() as conn:
            conn.execute("SELECT 1").fetchone()
        return True
    except Exception:
        return False


def get_board_state(repo_full_name: str) -> dict[str, list[dict]]:
    """Return board columns with their issues, ordered by position."""
    with get_db() as conn:
        rows = conn.execute(
            "SELECT issue_number, column_name, position, updated_at "
            "FROM board_state WHERE repo_full_name = ? "
            "ORDER BY position ASC, issue_number ASC",
            (repo_full_name,),
        ).fetchall()

    columns: dict[str, list[dict]] = {col: [] for col in KANBAN_COLUMNS}
    for row in rows:
        column = row["column_name"]
        if column not in columns:
            # Defensive: never let one bad row break the whole board.
            continue
        columns[column].append(
            {
                "issue_number": row["issue_number"],
                "position": row["position"],
                "updated_at": row["updated_at"],
            }
        )
    return columns


def move_issue(repo_full_name: str, issue_number: int, to_column: str) -> None:
    """Move an issue to a different column (appended at the end)."""
    if to_column not in KANBAN_COLUMNS:
        raise ValueError(f"Invalid column: {to_column}")
    with get_db() as conn:
        row = conn.execute(
            "SELECT COALESCE(MAX(position), -1) + 1 AS next_pos FROM board_state "
            "WHERE repo_full_name = ? AND column_name = ?",
            (repo_full_name, to_column),
        ).fetchone()
        conn.execute(
            """INSERT INTO board_state
                   (repo_full_name, issue_number, column_name, position, updated_at)
               VALUES (?, ?, ?, ?, datetime('now'))
               ON CONFLICT(repo_full_name, issue_number)
               DO UPDATE SET column_name = excluded.column_name,
                             position = excluded.position,
                             updated_at = excluded.updated_at""",
            (repo_full_name, issue_number, to_column, row["next_pos"]),
        )


def set_column_order(
    repo_full_name: str, column_name: str, ordered_issue_numbers: list[int]
) -> None:
    """Persist the exact order of the cards in one column.

    Each listed card keeps (or gains) this column and its position becomes
    its index in the list. Used after a drag & drop so manual ordering
    survives reloads.
    """
    if column_name not in KANBAN_COLUMNS:
        raise ValueError(f"Invalid column: {column_name}")
    with get_db() as conn:
        for index, issue_number in enumerate(ordered_issue_numbers):
            conn.execute(
                """INSERT INTO board_state
                       (repo_full_name, issue_number, column_name, position, updated_at)
                   VALUES (?, ?, ?, ?, datetime('now'))
                   ON CONFLICT(repo_full_name, issue_number)
                   DO UPDATE SET column_name = excluded.column_name,
                                 position = excluded.position,
                                 updated_at = excluded.updated_at""",
                (repo_full_name, issue_number, column_name, float(index)),
            )


def reconcile_board_state(
    repo_full_name: str, open_numbers: set[int], closed_numbers: set[int]
) -> dict[str, int]:
    """Reconcile local board state with the truth in Gitea.

    - Issues closed in Gitea move to the "done" column.
    - Rows for issues that no longer exist in Gitea are deleted.
    Returns counters of what changed.
    """
    changed = {"closed_to_done": 0, "purged": 0}
    known = open_numbers | closed_numbers
    with get_db() as conn:
        rows = conn.execute(
            "SELECT issue_number, column_name FROM board_state "
            "WHERE repo_full_name = ?",
            (repo_full_name,),
        ).fetchall()
        for row in rows:
            number = row["issue_number"]
            if number not in known:
                conn.execute(
                    "DELETE FROM board_state "
                    "WHERE repo_full_name = ? AND issue_number = ?",
                    (repo_full_name, number),
                )
                changed["purged"] += 1
            elif number in closed_numbers and row["column_name"] != "done":
                conn.execute(
                    "UPDATE board_state SET column_name = 'done', "
                    "updated_at = datetime('now') "
                    "WHERE repo_full_name = ? AND issue_number = ?",
                    (repo_full_name, number),
                )
                changed["closed_to_done"] += 1
    return changed


def set_notes(repo_full_name: str, note: str) -> None:
    with get_db() as conn:
        conn.execute(
            """INSERT INTO project_notes (repo_full_name, note)
               VALUES (?, ?)
               ON CONFLICT(repo_full_name)
               DO UPDATE SET note = excluded.note,
                             updated_at = datetime('now')""",
            (repo_full_name, note),
        )


def get_notes(repo_full_name: str) -> str | None:
    with get_db() as conn:
        row = conn.execute(
            "SELECT note FROM project_notes WHERE repo_full_name = ?",
            (repo_full_name,),
        ).fetchone()
    return row["note"] if row else None


def get_stats() -> dict:
    """Get overall stats across all projects."""
    with get_db() as conn:
        rows = conn.execute(
            "SELECT repo_full_name, column_name, COUNT(*) as cnt "
            "FROM board_state GROUP BY repo_full_name, column_name"
        ).fetchall()

    projects: dict[str, dict[str, int]] = {}
    for row in rows:
        if row["column_name"] not in KANBAN_COLUMNS:
            continue
        name = row["repo_full_name"]
        if name not in projects:
            projects[name] = {col: 0 for col in KANBAN_COLUMNS}
        projects[name][row["column_name"]] = row["cnt"]
    return projects