rfc-app/backend/app/collections.py

"""§22 collection grain — resolution helpers beneath the project tier.

In S1 each project has exactly one collection (the default). These helpers
recover the collection for a project and read the per-corpus fields (`type`,
`initial_state`) that moved down from `projects` in migration 029. Project-grain
authz (auth.py) recovers a row's project by joining `collections` on
`collection_id`.
"""
from __future__ import annotations

import json

from . import db

DEFAULT_COLLECTION_ID = "default"

# §22.4a item (2): the displayed noun for an entry is a type-driven label, a
# framework concept (like role names), not deployment content. The chrome reads
# this from the API rather than hardcoding "RFC", so a `specification` collection
# says "Spec" and a `bdd` collection says "Feature" with no per-deployment config.
ENTRY_NOUN = {
    "document": "RFC",
    "specification": "Spec",
    "bdd": "Feature",
}
_DEFAULT_ENTRY_NOUN = "RFC"


def entry_noun(collection_type: str) -> str:
    """The §22.4a entry noun for a collection type. Unknown types fall back to
    the generic 'RFC' so a future type is never label-less."""
    return ENTRY_NOUN.get(collection_type, _DEFAULT_ENTRY_NOUN)


def _enabled_models_from_config(config_json: str | None) -> list[str] | None:
    """§22.12 per-collection enabled_models from a `config_json` blob, or None
    when unset (the collection inherits its project's universe)."""
    if not config_json:
        return None
    try:
        cfg = json.loads(config_json)
    except (json.JSONDecodeError, TypeError):
        return None
    em = cfg.get("enabled_models") if isinstance(cfg, dict) else None
    return [str(m) for m in em] if isinstance(em, list) else None


def default_collection_id(project_id: str) -> str:
    """The id of a project's default (S1: sole) collection. Falls back to the
    literal 'default' when the project has no collection row yet."""
    row = db.conn().execute(
        "SELECT id FROM collections WHERE project_id = ? ORDER BY created_at, id LIMIT 1",
        (project_id,),
    ).fetchone()
    return row["id"] if row else DEFAULT_COLLECTION_ID


def project_of_collection(collection_id: str) -> str | None:
    """The project a collection belongs to, or None if unknown."""
    row = db.conn().execute(
        "SELECT project_id FROM collections WHERE id = ?", (collection_id,)
    ).fetchone()
    return row["project_id"] if row else None


def collection_initial_state(collection_id: str) -> str:
    """§22.4b landing state for new entries in a collection. 'super-draft'
    default for an unknown/unset row (today's safe flow)."""
    row = db.conn().execute(
        "SELECT initial_state FROM collections WHERE id = ?", (collection_id,)
    ).fetchone()
    if row is None or not row["initial_state"]:
        return "super-draft"
    return row["initial_state"]


def collection_type(collection_id: str) -> str:
    """The collection's immutable §22.4a type. 'document' default for unknown."""
    row = db.conn().execute(
        "SELECT type FROM collections WHERE id = ?", (collection_id,)
    ).fetchone()
    return row["type"] if row and row["type"] else "document"


def subfolder_of(collection_id: str) -> str:
    """The content-repo subfolder a collection lives under (§22.3). Empty string
    for the default collection (entries at the repo root `rfcs/`)."""
    row = db.conn().execute(
        "SELECT subfolder FROM collections WHERE id = ?", (collection_id,)
    ).fetchone()
    return (row["subfolder"] if row else "") or ""


def get_collection(collection_id: str) -> dict | None:
    """The full collection row as a dict, or None if unknown. `enabled_models`
    (§22.12) is unpacked from `config_json` as a list, or None when unset."""
    row = db.conn().execute(
        "SELECT id, project_id, type, subfolder, initial_state, visibility, name, "
        "config_json FROM collections WHERE id = ?",
        (collection_id,),
    ).fetchone()
    if row is None:
        return None
    out = dict(row)
    out["enabled_models"] = _enabled_models_from_config(out.pop("config_json", None))
    out["entry_noun"] = entry_noun(out["type"])
    return out


def list_collections(project_id: str, include_unlisted: bool = False) -> list[dict]:
    """Collections in a project, the default first then by name (§22.5). `unlisted`
    is omitted from enumeration unless include_unlisted (a direct-id read or the
    corpus mirror, which serves every collection)."""
    rows = db.conn().execute(
        "SELECT id, project_id, type, subfolder, initial_state, visibility, name "
        "FROM collections WHERE project_id = ? ORDER BY (id != 'default'), name, id",
        (project_id,),
    ).fetchall()
    out: list[dict] = []
    for r in rows:
        if not include_unlisted and r["visibility"] == "unlisted":
            continue
        item = dict(r)
        item["entry_noun"] = entry_noun(item["type"])
        out.append(item)
    return out