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 TABLE IF NOT EXISTS board_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                repo_full_name TEXT NOT NULL,
                issue_number INTEGER NOT NULL,
                event_type TEXT NOT NULL,
                from_column TEXT,
                to_column TEXT,
                source TEXT NOT NULL DEFAULT 'ui',
                created_at TEXT NOT NULL DEFAULT (datetime('now'))
            );

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

            CREATE INDEX IF NOT EXISTS idx_events_repo
                ON board_events(repo_full_name, issue_number, created_at);
            """
        )
        _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: dict = {"closed_to_done": 0, "purged": 0,
                     "closed_numbers": [], "purged_numbers": []}
    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
                changed["purged_numbers"].append(number)
            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
                changed["closed_numbers"].append(number)
    return changed


# ── Event journal ──────────────────────────────────────────

EVENT_TYPES = ("created", "moved", "closed", "reopened", "commented")


def log_event(
    repo_full_name: str,
    issue_number: int,
    event_type: str,
    from_column: str | None = None,
    to_column: str | None = None,
    source: str = "ui",
    created_at: str | None = None,
) -> None:
    """Append one entry to the board's event journal.

    The journal is append-only: it is the memory board_state lacks
    (time-in-column, activity feed, throughput / cycle-time metrics).
    """
    if event_type not in EVENT_TYPES:
        raise ValueError(f"Invalid event type: {event_type}")
    with get_db() as conn:
        if created_at:
            conn.execute(
                "INSERT INTO board_events (repo_full_name, issue_number, "
                "event_type, from_column, to_column, source, created_at) "
                "VALUES (?, ?, ?, ?, ?, ?, ?)",
                (repo_full_name, issue_number, event_type,
                 from_column, to_column, source, created_at),
            )
        else:
            conn.execute(
                "INSERT INTO board_events (repo_full_name, issue_number, "
                "event_type, from_column, to_column, source) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                (repo_full_name, issue_number, event_type,
                 from_column, to_column, source),
            )


def get_events(
    repo_full_name: str | None = None, limit: int = 100
) -> list[dict]:
    """Most recent journal entries, newest first (one repo or all)."""
    with get_db() as conn:
        if repo_full_name:
            rows = conn.execute(
                "SELECT * FROM board_events WHERE repo_full_name = ? "
                "ORDER BY id DESC LIMIT ?",
                (repo_full_name, limit),
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT * FROM board_events ORDER BY id DESC LIMIT ?",
                (limit,),
            ).fetchall()
    return [dict(row) for row in rows]


def column_entry_times(repo_full_name: str) -> dict[int, str]:
    """When each issue entered its current column, per the journal.

    Latest event per issue, comments excluded (a comment does not move
    a card). Issues without journal entries are absent — callers fall
    back to board_state.updated_at, then to the issue's creation date.
    """
    with get_db() as conn:
        rows = conn.execute(
            "SELECT issue_number, MAX(created_at) AS entered_at "
            "FROM board_events WHERE repo_full_name = ? "
            "AND event_type != 'commented' GROUP BY issue_number",
            (repo_full_name,),
        ).fetchall()
    return {row["issue_number"]: row["entered_at"] for row in rows}


def get_metrics(repo_full_name: str | None = None, window_days: int = 30) -> dict:
    """Throughput and cycle time computed from the journal.

    Cycle time = time from entering in_progress (or from creation, when
    no in_progress entry exists) to the closing event, averaged over
    issues closed inside the window.
    """
    from datetime import datetime, timedelta, timezone

    def _parse(value: str):
        try:
            return datetime.fromisoformat(value).replace(tzinfo=timezone.utc)
        except (ValueError, TypeError):
            return None

    with get_db() as conn:
        if repo_full_name:
            rows = conn.execute(
                "SELECT repo_full_name, issue_number, event_type, to_column, "
                "created_at FROM board_events WHERE repo_full_name = ? "
                "ORDER BY created_at, id",
                (repo_full_name,),
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT repo_full_name, issue_number, event_type, to_column, "
                "created_at FROM board_events ORDER BY created_at, id"
            ).fetchall()

    cutoff = datetime.now(timezone.utc) - timedelta(days=window_days)
    counts: dict[str, int] = {}
    starts: dict[tuple, datetime] = {}
    cycles: list[float] = []
    for row in rows:
        at = _parse(row["created_at"])
        if at is None:
            continue
        if at >= cutoff:
            counts[row["event_type"]] = counts.get(row["event_type"], 0) + 1
        key = (row["repo_full_name"], row["issue_number"])
        if row["event_type"] == "created":
            starts.setdefault(key, at)
        elif row["event_type"] == "moved" and row["to_column"] == "in_progress":
            starts[key] = at  # cycle clock (re)starts on entering in_progress
        elif row["event_type"] == "closed" and key in starts and at >= cutoff:
            cycles.append((at - starts[key]).total_seconds() / 86400.0)
    return {
        "window_days": window_days,
        "created": counts.get("created", 0),
        "moved": counts.get("moved", 0),
        "closed": counts.get("closed", 0),
        "reopened": counts.get("reopened", 0),
        "avg_cycle_days": round(sum(cycles) / len(cycles), 2) if cycles else None,
        "cycle_samples": len(cycles),
    }


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