gitea_saas/gitea_client.py

@@ -0,0 +1,162 @@
+"""Gitea API client (mockable seam, env token).
+
+Shared by the factory front-end (issue #4) and the review gate (issue #5).
+Real HTTP only when a token is available; the token is read from
+``GITEA_TOKEN`` (or passed in) and is never printed or placed on a command
+line. Tests use an in-memory fake, so this module is never exercised over
+the network in CI.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.error
+import urllib.request
+from typing import Any, Optional
+
+
+class GiteaClient:
+    """Thin urllib client for the subset of the Gitea API the SaaS needs.
+
+    ``owner`` defaults to the ``GITEA_OWNER`` environment variable and then
+    to ``pbwraith`` (the self-hosted instance's user).  Repo-scoped calls
+    build ``/api/v1/repos/{owner}/{repo}/...``; org-repo creation uses
+    ``/api/v1/orgs/{owner}/repos`` and falls back to the authenticated
+    user's repos when ``owner`` is a user account (orgs endpoint 404s).
+    """
+
+    def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None,
+                 owner: Optional[str] = None):
+        self.base_url = (base_url or os.environ.get(
+            "GITEA_BASE_URL", "https://gitea.185.214.135.234.nip.io")).rstrip("/")
+        self.token = token if token is not None else os.environ.get("GITEA_TOKEN", "")
+        self.owner = (owner or os.environ.get("GITEA_OWNER") or "pbwraith").strip() or "pbwraith"
+
+    # -- low-level ---------------------------------------------------------
+
+    def _api(self, method: str, api_path: str, body: Optional[dict] = None) -> Any:
+        """Perform an API call under ``/api/v1/{api_path}`` and parse JSON."""
+        data = json.dumps(body).encode() if body is not None else None
+        url = f"{self.base_url}/api/v1/{api_path}"
+        req = urllib.request.Request(
+            url, data=data, method=method,
+            headers={"Authorization": "token " + self.token,
+                     "Accept": "application/json",
+                     "Content-Type": "application/json"})
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            raw = resp.read()
+            return json.loads(raw) if raw else None
+
+    def _request(self, method: str, path: str, body: Optional[dict] = None) -> Any:
+        """Repo-scoped call: ``path`` = ``<repo>/...`` under the owner."""
+        return self._api(method, f"repos/{self.owner}/{path}", body)
+
+    # -- API surface used by the factory front-end (issue #4) --------------
+
+    def repo_exists(self, name: str) -> bool:
+        """True when ``owner/name`` already exists on the instance.
+
+        A 404 means absent; any other HTTP error propagates (fail-closed:
+        we never claim a repo is missing when the check itself failed).
+        """
+        try:
+            self._request("GET", f"{name}")
+        except urllib.error.HTTPError as exc:
+            if exc.code == 404:
+                return False
+            raise
+        return True
+
+    def create_org_repo(self, name: str) -> dict:
+        """Create ``owner/name`` if missing; return the repo record.
+
+        Idempotent by construction: when the repo already exists the result
+        of ``repo_exists`` is returned immediately (no duplicate creation).
+        Uses ``orgs/{owner}/repos`` when ``owner`` is an organization and
+        falls back to the authenticated user's repos (``user/repos``) when
+        it is a user account -- the self-hosted default.
+        """
+        name = (name or "").strip()
+        if not name:
+            raise ValueError("repo name must be a non-empty string")
+        if self.repo_exists(name):
+            # Reuse: fetch the record so callers see the canonical shape.
+            return self._request("GET", f"{name}") or {}
+        payload: dict = {"name": name, "auto_init": False, "private": True}
+        try:
+            return self._api("POST", f"orgs/{self.owner}/repos", payload)
+        except urllib.error.HTTPError as exc:
+            if exc.code not in (404, 422):
+                raise
+        # ``owner`` is a user account (or lacks org-create rights): fall back
+        # to creating the repo under the authenticated token user.
+        return self._api("POST", "user/repos", payload) or {}
+
+    def list_labels(self, repo: str) -> list:
+        """List ``owner/repo``'s labels: [{id, name, ...}, ...]."""
+        return self._request("GET", f"{repo}/labels") or []
+
+    def label_id(self, repo: str, name: str) -> Optional[int]:
+        """Resolve a label name to its id in ``owner/repo`` (None if absent)."""
+        name = (name or "").strip()
+        for label in self.list_labels(repo):
+            if label.get("name") == name:
+                return label.get("id")
+        return None
+
+    def create_issue(self, repo: str, title: str, body: str,
+                     labels: Optional[list] = None) -> int:
+        """Create an issue in ``owner/repo``; return its number.
+
+        ``labels`` is a list of label NAMES; each is resolved to the repo's
+        label id and unknown names are skipped (the issue is still created).
+        """
+        title = (title or "").strip()
+        if not title:
+            raise ValueError("issue title must be a non-empty string")
+        label_ids = []
+        for name in labels or []:
+            if not isinstance(name, str) or not name.strip():
+                continue
+            label_id = self.label_id(repo, name)
+            if label_id is not None and label_id not in label_ids:
+                label_ids.append(label_id)
+        payload = {"title": title, "body": body or ""}
+        if label_ids:
+            payload["labels"] = label_ids
+        created = self._request("POST", f"{repo}/issues", payload) or {}
+        number = created.get("number")
+        if not isinstance(number, int):
+            raise RuntimeError(f"Gitea create_issue returned no issue number: {created!r}")
+        return number
+
+    # -- API surface used by the review gate (issue #5) --------------------
+
+    def get_issue(self, repo: str, number: int) -> dict:
+        """Fetch an issue: {title, body, state, labels: [...]}."""
+        return self._request("GET", f"{repo}/issues/{number}") or {}
+
+    def get_pr(self, repo: str, number: int) -> dict:
+        """Fetch a PR: {head_sha, changed_files: [filename, ...]}."""
+        pr = self._request("GET", f"{repo}/pulls/{number}") or {}
+        files = self._request("GET", f"{repo}/pulls/{number}/files") or []
+        return {
+            "head_sha": ((pr.get("head") or {}).get("sha")),
+            "changed_files": [f.get("filename") for f in files if f.get("filename")],
+        }
+
+    def get_pr_diff(self, repo: str, number: int) -> str:
+        """Return the PR's diff text reconstructed from per-file patches."""
+        files = self._request("GET", f"{repo}/pulls/{number}/files") or []
+        chunks = []
+        for f in files:
+            filename = f.get("filename") or "?"
+            chunks.append(f"diff --git a/{filename} b/{filename}\n"
+                          f"--- a/{filename}\n+++ b/{filename}\n"
+                          + (f.get("patch") or ""))
+        return "\n".join(chunks)
+
+    def post_comment(self, repo: str, number: int, body: str) -> dict:
+        """Post a comment on an issue or PR conversation."""
+        return self._request("POST", f"{repo}/issues/{number}/comments", {"body": body}) or {}

gitea_saas/intake_factory.py

@@ -0,0 +1,465 @@
+"""Factory front-end: client repo provisioning + unfunded issue gating.
+
+Option-1 front-end from ARCHITECTURE.md.  All gates live HERE, inside the
+SaaS service -- the lane survey/dispatcher scripts are never touched for
+SaaS concepts (owner rule 2026-08-25): unfunded client work is held in a
+local ``pending_issues`` queue and only becomes a Gitea issue after payment.
+
+Owns its schema additions against the SAME SQLite file as the billing core.
+``schema.sql`` / ``db.py`` stay owned by issue #1; the factory tables are
+created idempotently (``CREATE TABLE IF NOT EXISTS ...``) from this module
+so the feature can land without touching the core's files:
+
+    client_repos (client_id, repo, created_at)          -- repo->client map
+    pending_issues (id, client_id, repo, title, body_json, labels_json,
+                    created_at, status)                  -- queued|shipped
+    factory_locks (client_id, locked_at)                 -- explicit gating
+
+Money rule (fail-closed): a client is funded iff its available balance is
+>= the debit floor and no lock is set.  Any uncertainty (unknown client,
+core error) is treated as UNFUNDED so work is queued, never silently
+shipped without payment.
+"""
+
+from __future__ import annotations
+
+import hmac
+import json
+import os
+import re
+import sqlite3
+import threading
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+# Header carrying the API key.  The shared constant lives in the sibling
+# ``gitea_saas/config.py`` (issue #2/#6 auth), which lands in a later PR than
+# this one; until that module is merged this file falls back to the same
+# canonical value.  Keeping the single source of truth means the factory
+# routes use ``compare_digest`` on the same header name as ``server.py``.
+try:
+    from .config import API_KEY_HEADER  # shared with gitea_saas.server
+except ImportError:  # config.py not merged yet (sibling auth PRs)
+    API_KEY_HEADER = "X-API-Key"
+
+# Default floor for "funded": a client must hold at least this many minor
+# units (default = 5.00 USD) before intake will ship issues.
+DEFAULT_DEBIT_FLOOR_MINOR = 500
+
+# Label applied to every issue flushed by :meth:`IntakeFactory.unlock`.
+P1_LABEL = "p1"
+
+_STATUS_QUEUED = "queued"
+_STATUS_SHIPPED = "shipped"
+
+_REPO_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,99}$")
+
+# Idempotent by construction (same convention as the billing schema).
+_FACTORY_DDL = """
+CREATE TABLE IF NOT EXISTS client_repos (
+    client_id  INTEGER NOT NULL,
+    repo       TEXT    NOT NULL,
+    created_at TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    PRIMARY KEY (client_id, repo)
+);
+
+CREATE TABLE IF NOT EXISTS pending_issues (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    client_id   INTEGER NOT NULL,
+    repo        TEXT    NOT NULL,
+    title       TEXT    NOT NULL,
+    body_json   TEXT    NOT NULL,
+    labels_json TEXT    NOT NULL DEFAULT '[]',
+    created_at  TEXT    NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    status      TEXT    NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'shipped'))
+);
+
+CREATE TABLE IF NOT EXISTS factory_locks (
+    client_id INTEGER PRIMARY KEY,
+    locked_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+
+CREATE INDEX IF NOT EXISTS idx_pending_client_status ON pending_issues (client_id, status);
+"""
+
+
+def _now_iso() -> str:
+    """Current UTC time as an ISO-8601 string (matches billing core format)."""
+    return datetime.now(timezone.utc).isoformat()
+
+
+class FactoryError(Exception):
+    """Base class for factory domain errors."""
+
+
+class UnknownClientError(FactoryError):
+    """Raised when a client_id does not exist in the billing core."""
+
+
+class InvalidRequestError(FactoryError):
+    """Raised on malformed inputs (empty names, bad repo names, ...)."""
+
+
+class CoreContractError(FactoryError):
+    """Raised when the billing core violates its (ok, details) contract."""
+
+
+class IntakeFactory:
+    """Provision client repos and gate issue intake on funding.
+
+    Parameters
+    ----------
+    core:
+        The billing core (issue #1 contract): ``get_client(client_id,
+        include_balance=...) -> (ok, details)``.  Duck-typed; never
+        imported directly so the module stays composable.
+    gitea:
+        A :class:`gitea_saas.gitea_client.GiteaClient` (or in-memory mock):
+        ``create_org_repo(name)``, ``repo_exists(name)``,
+        ``create_issue(repo, title, body, labels)``.
+    config:
+        Configuration object exposing ``db_path`` (a ``pathlib.Path`` to the
+        billing SQLite file).  Optionally ``debit_floor_minor`` and/or the
+        ``SAAS_DEBIT_FLOOR_MINOR`` env var override the default floor.
+    """
+
+    def __init__(self, core: Any, gitea: Any, config: Any):
+        db_path = getattr(config, "db_path", None)
+        if db_path is None:
+            raw = os.environ.get("SAAS_DB_PATH", "").strip()
+            db_path = Path(raw).expanduser() if raw else None
+        if db_path is None:
+            raise InvalidRequestError(
+                "IntakeFactory requires config.db_path (or SAAS_DB_PATH): "
+                "the factory persists its queue in the billing SQLite file")
+        self._core = core
+        self._gitea = gitea
+        self._config = config
+        self._conn = sqlite3.connect(str(Path(db_path)), check_same_thread=False)
+        self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._conn.execute("PRAGMA foreign_keys=ON")
+        self._conn.executescript(_FACTORY_DDL)
+        self._lock = threading.RLock()
+
+    # ------------------------------------------------------------------
+    # Config
+    # ------------------------------------------------------------------
+
+    @property
+    def debit_floor_minor(self) -> int:
+        """The funded threshold in minor units (config attr > env > default)."""
+        value = getattr(self._config, "debit_floor_minor", None)
+        if value is None:
+            raw = os.environ.get("SAAS_DEBIT_FLOOR_MINOR", "").strip()
+            value = int(raw) if raw else DEFAULT_DEBIT_FLOOR_MINOR
+        if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+            raise InvalidRequestError(f"invalid debit floor: {value!r}")
+        return value
+
+    # ------------------------------------------------------------------
+    # Billing-core helpers (fail-closed)
+    # ------------------------------------------------------------------
+
+    def _client_record(self, client_id: int, include_balance: bool = False) -> Dict[str, Any]:
+        """Fetch a client record from the core; raise when unknown/broken."""
+        ok, details = self._core.get_client(client_id, include_balance=include_balance)
+        if not ok:
+            code = (details or {}).get("error", {}).get("code") if isinstance(details, dict) else None
+            if code and "UNKNOWN" in str(code).upper():
+                raise UnknownClientError(f"no client with id {client_id}")
+            raise CoreContractError(f"core.get_client({client_id}) failed: {details!r}")
+        if not isinstance(details, dict) or (include_balance and not isinstance(
+                details.get("balance"), int)):
+            raise CoreContractError(f"core.get_client({client_id}) violated its contract: {details!r}")
+        return details
+
+    def _client_balance(self, client_id: int) -> Optional[int]:
+        """Available balance, or None when funding cannot be confirmed."""
+        try:
+            details = self._client_record(client_id, include_balance=True)
+        except (UnknownClientError, CoreContractError):
+            return None
+        return details.get("balance")
+
+    def client_unfunded(self, client_id: int) -> bool:
+        """True when the client's balance is below the debit floor.
+
+        Fail-closed: an unknown client or a core error also reports
+        ``True`` -- intake queues instead of shipping unverified work.
+        """
+        balance = self._client_balance(client_id)
+        if balance is None:
+            return True
+        return balance < self.debit_floor_minor
+
+    def _require_client(self, client_id: int) -> Dict[str, Any]:
+        """Validate existence (no balance needed) and return the record."""
+        if isinstance(client_id, bool) or not isinstance(client_id, int) or client_id <= 0:
+            raise InvalidRequestError("client_id must be a positive integer")
+        return self._client_record(client_id, include_balance=False)
+
+    def _is_locked(self, client_id: int) -> bool:
+        row = self._conn.execute(
+            "SELECT 1 FROM factory_locks WHERE client_id = ?", (client_id,)).fetchone()
+        return row is not None
+
+    # ------------------------------------------------------------------
+    # Repo provisioning
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _default_repo_name(client_record: Dict[str, Any]) -> str:
+        """Derive a repo name from the client record (client name slug).
+
+        Accepts both the historical nested shape (``{client: {...}}``) and
+        the flat record now returned by :meth:`BillingCore.get_client`
+        (issue #16: ``{id, name, tier, status, meta, created_at, balance}``)
+        -- the factory composes with whichever contract the core exposes.
+        """
+        name = (client_record.get("client") or {}).get("name") or client_record.get("name") or ""
+        slug = re.sub(r"[^A-Za-z0-9._-]+", "-", name.strip().lower()).strip("-")
+        cid_fallback = (client_record.get("client_id")
+                        or client_record.get("id")
+                        or (client_record.get("client") or {}).get("id")
+                        or "")
+        return slug[:60] or f"client-{cid_fallback}"
+
+    def provision_client(self, client_id: int, repo_name: Optional[str] = None) -> str:
+        """Provision ``owner/<repo>`` for a client; returns the repo name.
+
+        Idempotent: a second call for the same client reuses the existing
+        mapping and never creates the repo twice.  When ``repo_name`` is
+        omitted it is derived from the client's name.
+        """
+        record = self._require_client(client_id)
+        if repo_name is not None:
+            repo_name = (repo_name or "").strip()
+            if not _REPO_NAME_RE.match(repo_name):
+                raise InvalidRequestError(
+                    f"invalid repo name: {repo_name!r} (letters, digits, . _ -; 1-100 chars)")
+        else:
+            repo_name = self._default_repo_name(record)
+        with self._lock:
+            mapped = self._conn.execute(
+                "SELECT repo FROM client_repos WHERE client_id = ? AND repo = ?",
+                (client_id, repo_name)).fetchone()
+            if mapped is None:
+                self._gitea.create_org_repo(repo_name)
+                self._conn.execute(
+                    "INSERT OR IGNORE INTO client_repos (client_id, repo, created_at)"
+                    " VALUES (?, ?, ?)", (client_id, repo_name, _now_iso()))
+                self._conn.commit()
+        return repo_name
+
+    # ------------------------------------------------------------------
+    # Issue intake (funded shipping vs unfunded queue)
+    # ------------------------------------------------------------------
+
+    def submit_issue(self, client_id: int, repo: str, title: str,
+                     acceptance_body: str, labels: Optional[List[str]] = None) -> Dict[str, Any]:
+        """Submit an issue for a client.
+
+        Funded + unlocked -> created in Gitea immediately,
+        ``{delivered: True, issue_number: N}``.
+        Unfunded or locked -> queued in ``pending_issues``,
+        ``{delivered: False, pending_id: N, reason: 'unfunded'|'locked'}``.
+        """
+        self._require_client(client_id)
+        repo = (repo or "").strip()
+        title = (title or "").strip()
+        if not _REPO_NAME_RE.match(repo):
+            raise InvalidRequestError(f"invalid repo name: {repo!r}")
+        if not title:
+            raise InvalidRequestError("issue title must be a non-empty string")
+        reason = "locked" if self._is_locked(client_id) else (
+            "unfunded" if self.client_unfunded(client_id) else None)
+        if reason is not None:
+            with self._lock:
+                cur = self._conn.execute(
+                    "INSERT INTO pending_issues (client_id, repo, title, body_json,"
+                    " labels_json, created_at, status) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    (client_id, repo, title,
+                     json.dumps(acceptance_body or ""),
+                     json.dumps([str(l) for l in (labels or [])]),
+                     _now_iso(), _STATUS_QUEUED))
+                self._conn.commit()
+                pending_id = cur.lastrowid
+            return {"delivered": False, "pending_id": pending_id, "reason": reason}
+        issue_number = self._gitea.create_issue(repo, title, acceptance_body or "", labels)
+        return {"delivered": True, "issue_number": issue_number}
+
+    # ------------------------------------------------------------------
+    # Lock / unlock
+    # ------------------------------------------------------------------
+
+    def lock(self, client_id: int) -> Dict[str, Any]:
+        """Stop shipping: future submits queue instead of creating issues.
+
+        Idempotent.  No label games and no dispatcher coupling -- the gate
+        is this service's local state.
+        """
+        self._require_client(client_id)
+        with self._lock:
+            self._conn.execute(
+                "INSERT OR REPLACE INTO factory_locks (client_id, locked_at)"
+                " VALUES (?, ?)", (client_id, _now_iso()))
+            self._conn.commit()
+        return {"client_id": client_id, "locked": True}
+
+    def unlock(self, client_id: int) -> Dict[str, Any]:
+        """Flush ALL queued issues for the client into Gitea (with ``p1``).
+
+        Each flushed issue is created with its original labels plus ``p1``
+        and marked ``shipped``.  Unlocking also clears an active lock so the
+        client ships again.  A second unlock with nothing queued is a
+        no-op: ``{flushed: 0, issues: []}``.
+        """
+        self._require_client(client_id)
+        with self._lock:
+            rows = self._conn.execute(
+                "SELECT * FROM pending_issues WHERE client_id = ? AND status = ?"
+                " ORDER BY id", (client_id, _STATUS_QUEUED)).fetchall()
+            numbers: List[int] = []
+            for row in rows:
+                labels = [str(l) for l in json.loads(row["labels_json"] or "[]")]
+                if P1_LABEL not in labels:
+                    labels.append(P1_LABEL)
+                number = self._gitea.create_issue(
+                    row["repo"], row["title"], json.loads(row["body_json"] or '""'), labels)
+                self._conn.execute(
+                    "UPDATE pending_issues SET status = ? WHERE id = ?",
+                    (_STATUS_SHIPPED, row["id"]))
+                numbers.append(number)
+            cleared = self._conn.execute(
+                "DELETE FROM factory_locks WHERE client_id = ?", (client_id,)).rowcount
+            self._conn.commit()
+        return {"flushed": len(numbers), "issues": numbers, "lock_cleared": cleared > 0}
+
+
+# ---------------------------------------------------------------------------
+# HTTP surface (mounted by the issue #2 Flask app; see register_factory_routes)
+# ---------------------------------------------------------------------------
+#
+# Routes (same auth discipline as gitea_saas.server -- API_KEY_HEADER via
+# hmac.compare_digest, fail-closed):
+#   POST /api/factory/provision           {"client_id": int, "repo_name"?: str}
+#   POST /api/factory/<client_id>/issues  {"repo", "title", "acceptance_body",
+#                                          "labels"?: [...]}
+#   POST /api/factory/<client_id>/unlock  {}
+#   POST /api/factory/<client_id>/lock    {}
+#
+# server.py is owned by issues #2/#6, so the mount point itself stays out of
+# this module; creating the app just calls::
+
+_HOOK_NOTES = """\
+    from gitea_saas.intake_factory import register_factory_routes
+    app = create_app(core, api_key)
+    register_factory_routes(app, core, gitea_client, cfg)
+"""
+
+
+def register_factory_routes(app: Any, core: Any, gitea: Any, config: Any) -> None:
+    """Mount the factory endpoints on a Flask app (idempotent).
+
+    The app keeps its own ``CORE``/``API_KEY`` config keys; this function
+    only adds routes that build an :class:`IntakeFactory` per request (the
+    factory is stateless apart from its own SQLite connection, which is
+    cheap to open and thread-safe via WAL + a per-instance lock).
+    """
+    from flask import jsonify, request
+
+    api_key = app.config.get("API_KEY") or getattr(config, "api_key", None) or ""
+
+    def _factory() -> IntakeFactory:
+        return IntakeFactory(core, gitea, config)
+
+    def _authorized() -> bool:
+        sent = request.headers.get(API_KEY_HEADER, "")
+        return bool(api_key) and hmac.compare_digest(sent, api_key)
+
+    def _require_auth():
+        if not _authorized():
+            resp = jsonify({"error": "missing or invalid API key", "code": "unauthorized"})
+            resp.status_code = 401
+            return resp
+        return None
+
+    @app.get("/api/factory/health")
+    def factory_health():
+        return jsonify({"ok": True})
+
+    @app.post("/api/factory/provision")
+    def factory_provision():
+        if (denied := _require_auth()) is not None:
+            return denied
+        body = request.get_json(silent=True) or {}
+        raw_id = body.get("client_id")
+        if isinstance(raw_id, bool) or not isinstance(raw_id, int) or raw_id <= 0:
+            return _json_error(400, "invalid_request", "client_id must be a positive integer")
+        client_id = raw_id
+        repo_name = body.get("repo_name")
+        try:
+            repo = _factory().provision_client(client_id, repo_name or None)
+        except Exception as exc:
+            return _map_factory_error(exc)
+        return jsonify({"repo": repo})
+
+    @app.post("/api/factory/<int:client_id>/issues")
+    def factory_submit(client_id: int):
+        if (denied := _require_auth()) is not None:
+            return denied
+        body = request.get_json(silent=True) or {}
+        repo = body.get("repo")
+        title = body.get("title")
+        acceptance_body = body.get("acceptance_body") or ""
+        labels = body.get("labels")
+        if not isinstance(labels, list):
+            labels = None
+        try:
+            result = _factory().submit_issue(
+                client_id, repo or "", title or "", acceptance_body,
+                [str(l) for l in labels] if labels else None)
+        except Exception as exc:
+            return _map_factory_error(exc)
+        resp = jsonify(result)
+        resp.status_code = 202 if not result["delivered"] else 201
+        return resp
+
+    @app.post("/api/factory/<int:client_id>/unlock")
+    def factory_unlock(client_id: int):
+        if (denied := _require_auth()) is not None:
+            return denied
+        try:
+            return jsonify(_factory().unlock(client_id))
+        except Exception as exc:
+            return _map_factory_error(exc)
+
+    @app.post("/api/factory/<int:client_id>/lock")
+    def factory_lock(client_id: int):
+        if (denied := _require_auth()) is not None:
+            return denied
+        try:
+            return jsonify(_factory().lock(client_id))
+        except Exception as exc:
+            return _map_factory_error(exc)
+
+
+def _json_error(status: int, code: str, message: str):
+    from flask import jsonify
+
+    resp = jsonify({"error": message, "code": code})
+    resp.status_code = status
+    return resp
+
+
+def _map_factory_error(exc: Exception):
+    """Map factory domain errors to HTTP responses (fail-closed on unknown)."""
+    if isinstance(exc, UnknownClientError):
+        return _json_error(404, "unknown_client", str(exc))
+    if isinstance(exc, (InvalidRequestError, ValueError)):
+        return _json_error(400, "invalid_request", str(exc))
+    if isinstance(exc, CoreContractError):
+        return _json_error(500, "core_contract", str(exc))
+    return _json_error(500, "factory_error", f"{type(exc).__name__}: {exc}")

tests/test_intake_factory.py

@@ -0,0 +1,528 @@
+"""Tests for ``gitea_saas.intake_factory`` (factory intake, issue #4).
+
+Hermetic by design: an in-memory ``MockGitea`` + a reference billing core
+over a temp SQLite DB.  No network, no Stripe, no Gitea, no dispatcher
+files.
+
+Covers the issue #4 acceptance checklist:
+
+1. provision creates repo once; second call reuses (idempotent)
+2. funded submit creates Gitea issue immediately, delivered=true
+3. unfunded submit queues (no Gitea issue created), delivered=false
+4. unlock flushes queued -> Gitea with p1 label, marks shipped, second
+   unlock is a no-op
+5. lock stops shipping: subsequent submits queue again
+6. client with balance below floor is ``client_unfunded() == True``
+7. all tests against MockGitea + tmp DB; no network
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import types
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+from gitea_saas.intake_factory import (
+    DEFAULT_DEBIT_FLOOR_MINOR,
+    IntakeFactory,
+    UnknownClientError,
+    register_factory_routes,
+)
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# ---------------------------------------------------------------------------
+# Reference billing core (issue #1 contract subset: (ok, details) tuples)
+# ---------------------------------------------------------------------------
+
+class ReferenceCore:
+    """Minimal billing core honouring the documented ``(ok, details)``
+    contract over a temp SQLite DB (subset used by the factory)."""
+
+    SCHEMA = """
+    CREATE TABLE IF NOT EXISTS clients (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        name TEXT NOT NULL UNIQUE,
+        tier TEXT NOT NULL DEFAULT 'standard',
+        status TEXT NOT NULL DEFAULT 'active',
+        meta_json TEXT,
+        created_at TEXT NOT NULL
+    );
+    CREATE TABLE IF NOT EXISTS ledger (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        client_id INTEGER NOT NULL REFERENCES clients(id),
+        kind TEXT NOT NULL CHECK (kind IN ('credit','debit','hold','release','refund')),
+        amount_minor INTEGER NOT NULL,
+        currency TEXT NOT NULL DEFAULT 'USD',
+        idempotency_key TEXT UNIQUE NOT NULL,
+        ref_type TEXT,
+        ref_id TEXT,
+        created_at TEXT NOT NULL
+    );
+    """
+
+    def __init__(self, db_path: Path):
+        self._conn = sqlite3.connect(str(db_path))
+        self._conn.row_factory = sqlite3.Row
+        self._conn.executescript(self.SCHEMA)
+        self._seq = 0
+
+    def _key(self) -> str:
+        self._seq += 1
+        return f"ref-{self._seq}"
+
+    def create_client(self, name: str, meta: dict | None = None):
+        try:
+            cur = self._conn.execute(
+                "INSERT INTO clients (name, meta_json, created_at) VALUES (?, ?, ?)",
+                (name, json.dumps(meta or {}), _now_iso()))
+        except sqlite3.IntegrityError:
+            return False, {"error": {"code": "DUPLICATE_CLIENT", "message": name}}
+        self._conn.commit()
+        return True, {"client_id": cur.lastrowid, "name": name}
+
+    def credit(self, client_id: int, amount_minor: int):
+        self._conn.execute(
+            "INSERT INTO ledger (client_id, kind, amount_minor, currency,"
+            " idempotency_key, created_at) VALUES (?, 'credit', ?, 'USD', ?, ?)",
+            (client_id, amount_minor, self._key(), _now_iso()))
+        self._conn.commit()
+        return True, {"client_id": client_id, "balance": self._balance(client_id)}
+
+    def _balance(self, client_id: int) -> int:
+        row = self._conn.execute(
+            "SELECT COALESCE(SUM(CASE WHEN kind='credit' THEN amount_minor"
+            " ELSE -amount_minor END), 0) FROM ledger WHERE client_id = ?",
+            (client_id,)).fetchone()
+        return int(row[0])
+
+    def get_client(self, client_id: int, include_balance: bool = True):
+        row = self._conn.execute(
+            "SELECT * FROM clients WHERE id = ?", (client_id,)).fetchone()
+        if row is None:
+            return False, {"error": {"code": "UNKNOWN_CLIENT",
+                                     "message": f"no client with id {client_id}"}}
+        client = dict(row)
+        client["meta"] = json.loads(client.pop("meta_json") or "{}")
+        details = {"client": client}
+        if include_balance:
+            details["balance"] = self._balance(client_id)
+        return True, details
+
+    def get_balance(self, client_id: int):
+        row = self._conn.execute(
+            "SELECT id FROM clients WHERE id = ?", (client_id,)).fetchone()
+        if row is None:
+            return False, {"error": {"code": "UNKNOWN_CLIENT", "message": "no client"}}
+        return True, {"client_id": client_id, "balance": self._balance(client_id)}
+
+
+# ---------------------------------------------------------------------------
+# MockGitea (in-memory fake of gitea_saas.gitea_client.GiteaClient)
+# ---------------------------------------------------------------------------
+
+class MockGitea:
+    """In-memory Gitea: repos, per-repo labels (names -> ids) and issues."""
+
+    def __init__(self):
+        self.repos: dict = {}
+        self.labels: dict = {}
+        self.created_repo_calls: list = []
+        self._label_seq = 0
+
+    # -- repo provisioning -------------------------------------------------
+
+    def repo_exists(self, name: str) -> bool:
+        return name in self.repos
+
+    def create_org_repo(self, name: str) -> dict:
+        self.created_repo_calls.append(name)
+        if name not in self.repos:
+            self.repos[name] = {"name": name, "issues": []}
+            self.labels[name] = {"p1": self._next_label_id()}
+        return {"name": name}
+
+    # -- labels ------------------------------------------------------------
+
+    def _next_label_id(self) -> int:
+        self._label_seq += 1
+        return self._label_seq
+
+    def add_label(self, repo: str, name: str) -> int:
+        if name not in self.labels.setdefault(repo, {}):
+            self.labels[repo][name] = self._next_label_id()
+        return self.labels[repo][name]
+
+    def list_labels(self, repo: str) -> list:
+        return [{"id": i, "name": n} for n, i in self.labels.get(repo, {}).items()]
+
+    def label_id(self, repo: str, name: str) -> int | None:
+        return self.labels.get(repo, {}).get(name)
+
+    # -- issues ------------------------------------------------------------
+
+    def create_issue(self, repo: str, title: str, body: str,
+                     labels: list | None = None) -> int:
+        rec = self.repos[repo]
+        number = len(rec["issues"]) + 1
+        label_ids = []
+        for name in labels or []:
+            lid = self.label_id(repo, name)
+            if lid is not None and lid not in label_ids:
+                label_ids.append(lid)
+        rec["issues"].append({
+            "number": number, "title": title, "body": body, "labels": label_ids,
+        })
+        return number
+
+    def issue(self, repo: str, number: int) -> dict:
+        return self.repos[repo]["issues"][number - 1]
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture()
+def env(tmp_path: Path, monkeypatch):
+    core = ReferenceCore(tmp_path / "billing.db")
+    gitea = MockGitea()
+    cfg = types.SimpleNamespace(db_path=tmp_path / "billing.db",
+                                api_key="test-api-key",
+                                debit_floor_minor=DEFAULT_DEBIT_FLOOR_MINOR)
+    factory = IntakeFactory(core, gitea, cfg)
+    return types.SimpleNamespace(core=core, gitea=gitea, cfg=cfg, factory=factory,
+                                 db_path=tmp_path / "billing.db")
+
+
+def _client(env, name: str = "Acme Corp", amount_minor: int = 0) -> int:
+    ok, details = env.core.create_client(name)
+    assert ok, details
+    client_id = details["client_id"]
+    if amount_minor:
+        ok, _ = env.core.credit(client_id, amount_minor)
+        assert ok
+    return client_id
+
+
+def _queued_rows(db_path: Path, client_id: int) -> list:
+    conn = sqlite3.connect(str(db_path))
+    try:
+        conn.row_factory = sqlite3.Row
+        rows = conn.execute(
+            "SELECT * FROM pending_issues WHERE client_id = ? ORDER BY id",
+            (client_id,)).fetchall()
+        return [dict(r) for r in rows]
+    finally:
+        conn.close()
+
+
+def _mapping_rows(db_path: Path, client_id: int) -> list:
+    conn = sqlite3.connect(str(db_path))
+    try:
+        conn.row_factory = sqlite3.Row
+        rows = conn.execute(
+            "SELECT * FROM client_repos WHERE client_id = ?", (client_id,)).fetchall()
+        return [dict(r) for r in rows]
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------------
+# 1. provision: creates once, reuses on second call (idempotent)
+# ---------------------------------------------------------------------------
+
+def test_provision_creates_repo_once_and_reuses(env):
+    cid = _client(env, "Acme Corp")
+    repo = env.factory.provision_client(cid)
+    assert repo == "acme-corp"
+    assert env.gitea.repo_exists("acme-corp")
+    assert env.gitea.created_repo_calls == ["acme-corp"]
+
+    # Second provision: same repo, no duplicate creation, one mapping row.
+    again = env.factory.provision_client(cid)
+    assert again == "acme-corp"
+    assert env.gitea.created_repo_calls == ["acme-corp"]
+    assert len(_mapping_rows(env.db_path, cid)) == 1
+
+
+def test_provision_explicit_repo_name_and_cross_client_isolation(env):
+    cid = _client(env, "Globex")
+    repo = env.factory.provision_client(cid, "globex-main")
+    assert repo == "globex-main"
+    assert env.gitea.repo_exists("globex-main")
+    # A different client gets its own mapping and repo.
+    cid2 = _client(env, "Initech")
+    repo2 = env.factory.provision_client(cid2)
+    assert repo2 == "initech"
+    assert len(env.gitea.created_repo_calls) == 2
+    assert len(_mapping_rows(env.db_path, cid)) == 1
+    assert len(_mapping_rows(env.db_path, cid2)) == 1
+
+
+def test_provision_rejects_bad_repo_name_and_unknown_client(env):
+    cid = _client(env)
+    with pytest.raises(Exception):
+        env.factory.provision_client(cid, "bad name with spaces")
+    with pytest.raises(UnknownClientError):
+        env.factory.provision_client(9999, "whatever-repo")
+
+
+# ---------------------------------------------------------------------------
+# 2. funded submit: Gitea issue immediately, delivered=true
+# ---------------------------------------------------------------------------
+
+def test_funded_submit_delivers_immediately(env):
+    cid = _client(env, "Acme Corp", amount_minor=10_000)
+    env.factory.provision_client(cid)
+    env.gitea.add_label("acme-corp", "p2")
+    result = env.factory.submit_issue(
+        cid, "acme-corp", "Fix the bug",
+        "## Acceptance\n- [ ] it works", labels=["p2"])
+    assert result == {"delivered": True, "issue_number": 1}
+    issue = env.gitea.issue("acme-corp", 1)
+    assert issue["title"] == "Fix the bug"
+    assert issue["body"] == "## Acceptance\n- [ ] it works"
+    assert issue["labels"] == [env.gitea.label_id("acme-corp", "p2")]
+    # Nothing queued for a funded direct delivery.
+    assert _queued_rows(env.db_path, cid) == []
+
+
+def test_funded_submit_without_labels(env):
+    cid = _client(env, "Acme Corp", amount_minor=10_000)
+    env.factory.provision_client(cid)
+    result = env.factory.submit_issue(cid, "acme-corp", "No labels", "")
+    assert result["delivered"] is True
+    assert env.gitea.issue("acme-corp", 1)["labels"] == []
+
+
+# ---------------------------------------------------------------------------
+# 3. unfunded submit: queued, no Gitea issue, delivered=false
+# ---------------------------------------------------------------------------
+
+def test_unfunded_submit_queues(env):
+    cid = _client(env, "Acme Corp", amount_minor=0)
+    env.factory.provision_client(cid)
+    result = env.factory.submit_issue(
+        cid, "acme-corp", "Queued work", "## Acceptance\n- [ ] later")
+    assert result["delivered"] is False
+    assert result["reason"] == "unfunded"
+    assert isinstance(result["pending_id"], int)
+    assert env.gitea.repos["acme-corp"]["issues"] == []
+    rows = _queued_rows(env.db_path, cid)
+    assert len(rows) == 1
+    row = rows[0]
+    assert row["status"] == "queued"
+    assert row["title"] == "Queued work"
+    assert json.loads(row["body_json"]) == "## Acceptance\n- [ ] later"
+    assert json.loads(row["labels_json"]) == []
+
+
+# ---------------------------------------------------------------------------
+# 4. unlock: flush queued -> Gitea with p1 label, shipped, second no-op
+# ---------------------------------------------------------------------------
+
+def test_unlock_flushes_queued_with_p1_and_is_idempotent(env):
+    cid = _client(env, "Acme Corp", amount_minor=0)
+    env.factory.provision_client(cid)
+    env.gitea.add_label("acme-corp", "p3")
+    r1 = env.factory.submit_issue(cid, "acme-corp", "Task A", "body A", labels=["p3"])
+    r2 = env.factory.submit_issue(cid, "acme-corp", "Task B", "body B")
+    assert r1["delivered"] is False and r2["delivered"] is False
+
+    flushed = env.factory.unlock(cid)
+    assert flushed["flushed"] == 2
+    assert flushed["issues"] == [1, 2]
+    p1 = env.gitea.label_id("acme-corp", "p1")
+    p3 = env.gitea.label_id("acme-corp", "p3")
+    a = env.gitea.issue("acme-corp", 1)
+    b = env.gitea.issue("acme-corp", 2)
+    assert a["title"] == "Task A" and a["body"] == "body A"
+    assert p1 in a["labels"] and p3 in a["labels"]
+    assert b["labels"] == [p1]
+    # Rows marked shipped.
+    rows = _queued_rows(env.db_path, cid)
+    assert len(rows) == 2 and all(r["status"] == "shipped" for r in rows)
+
+    # Second unlock: nothing queued -> no-op.
+    again = env.factory.unlock(cid)
+    assert again == {"flushed": 0, "issues": [], "lock_cleared": False}
+    assert len(env.gitea.repos["acme-corp"]["issues"]) == 2
+
+
+# ---------------------------------------------------------------------------
+# 5. lock: future submits queue even when funded; unlock clears + flushes
+# ---------------------------------------------------------------------------
+
+def test_lock_queues_funded_submits_and_unlock_resumes(env):
+    cid = _client(env, "Acme Corp", amount_minor=10_000)
+    env.factory.provision_client(cid)
+    assert env.factory.lock(cid) == {"client_id": cid, "locked": True}
+
+    # Funded but locked -> queued, no Gitea issue.
+    result = env.factory.submit_issue(cid, "acme-corp", "Blocked by lock", "")
+    assert result == {"delivered": False, "pending_id": 1, "reason": "locked"}
+    assert env.gitea.repos["acme-corp"]["issues"] == []
+
+    # unlock flushes the queued issue AND clears the lock.
+    flushed = env.factory.unlock(cid)
+    assert flushed["flushed"] == 1 and flushed["lock_cleared"] is True
+    assert env.gitea.repos["acme-corp"]["issues"][0]["title"] == "Blocked by lock"
+
+    # Lock cleared -> shipping resumes.
+    again = env.factory.submit_issue(cid, "acme-corp", "Now shipping", "")
+    assert again == {"delivered": True, "issue_number": 2}
+
+    # lock() twice is idempotent; lock() on unknown client raises.
+    assert env.factory.lock(cid)["locked"] is True
+    with pytest.raises(UnknownClientError):
+        env.factory.lock(9999)
+
+
+# ---------------------------------------------------------------------------
+# 6. client_unfunded: below the debit floor (and fail-closed)
+# ---------------------------------------------------------------------------
+
+def test_client_unfunded_boundary(env):
+    cid = _client(env, "Acme Corp", amount_minor=0)
+    assert env.factory.client_unfunded(cid) is True
+    env.core.credit(cid, DEFAULT_DEBIT_FLOOR_MINOR - 1)          # 499
+    assert env.factory.client_unfunded(cid) is True
+    env.core.credit(cid, 1)                                       # 500 == floor
+    assert env.factory.client_unfunded(cid) is False
+    env.core.credit(cid, 1000)                                    # 1500
+    assert env.factory.client_unfunded(cid) is False
+
+
+def test_client_unfunded_fail_closed_on_unknown_client(env):
+    # Unknown client -> cannot confirm funding -> unfunded (fail-closed).
+    assert env.factory.client_unfunded(9999) is True
+    with pytest.raises(UnknownClientError):
+        env.factory.submit_issue(9999, "acme-corp", "Nope", "")
+
+
+def test_floor_override_via_config(env):
+    cid = _client(env, "Acme Corp", amount_minor=200)
+    env.cfg.debit_floor_minor = 100
+    assert env.factory.client_unfunded(cid) is False   # 200 >= 100
+    env.cfg.debit_floor_minor = 300
+    assert env.factory.client_unfunded(cid) is True    # 200 < 300
+
+
+# ---------------------------------------------------------------------------
+# HTTP surface (register_factory_routes), same X-API-Key discipline
+# ---------------------------------------------------------------------------
+
+@pytest.fixture()
+def app_client(env):
+    from flask import Flask
+
+    app = Flask(__name__)
+    app.config["API_KEY"] = env.cfg.api_key
+    register_factory_routes(app, env.core, env.gitea, env.cfg)
+    return app.test_client()
+
+
+def _post(app, path, payload=None, key="test-api-key"):
+    headers = {"X-API-Key": key} if key else {}
+    return app.post(path, json=payload or {}, headers=headers)
+
+
+def test_routes_require_api_key(app_client):
+    resp = app_client.post("/api/factory/provision", json={"client_id": 1})
+    assert resp.status_code == 401
+    resp = app_client.post("/api/factory/1/unlock")
+    assert resp.status_code == 401
+
+
+def test_route_provision_submit_unlock_lock_flow(env, app_client):
+    cid = _client(env, "Acme Corp", amount_minor=0)
+    resp = _post(app_client, "/api/factory/provision", {"client_id": cid})
+    assert resp.status_code == 200
+    assert resp.get_json() == {"repo": "acme-corp"}
+
+    # Unfunded submit -> 202 queued.
+    resp = _post(app_client, f"/api/factory/{cid}/issues",
+                 {"repo": "acme-corp", "title": "Queued via API",
+                  "acceptance_body": "## Acceptance"})
+    assert resp.status_code == 202
+    body = resp.get_json()
+    assert body["delivered"] is False and body["reason"] == "unfunded"
+
+    # Unlock -> 200 flushed with p1.
+    resp = _post(app_client, f"/api/factory/{cid}/unlock")
+    assert resp.status_code == 200
+    assert resp.get_json()["flushed"] == 1
+
+    # Lock -> 200; funded submit afterwards still queues.
+    env.core.credit(cid, 10_000)
+    resp = _post(app_client, f"/api/factory/{cid}/lock")
+    assert resp.get_json() == {"client_id": cid, "locked": True}
+    resp = _post(app_client, f"/api/factory/{cid}/issues",
+                 {"repo": "acme-corp", "title": "Locked via API"})
+    assert resp.get_json()["reason"] == "locked"
+
+
+def test_route_unknown_client_404(env, app_client):
+    resp = _post(app_client, "/api/factory/provision", {"client_id": 9999})
+    assert resp.status_code == 404
+    assert resp.get_json()["code"] == "unknown_client"
+
+
+def test_route_validation_400(app_client):
+    resp = _post(app_client, "/api/factory/provision", {"client_id": -1})
+    assert resp.status_code == 400
+
+
+# ---------------------------------------------------------------------------
+# 7. Hermeticity guard
+# ---------------------------------------------------------------------------
+
+def test_hermetic_no_real_gitea_or_network_used(env):
+    """The whole suite runs against mocks; a real GiteaClient would need a
+    token and a network.  This test pins the factory's seams: every gitea
+    interaction goes through the injected mock object."""
+    assert isinstance(env.gitea, MockGitea)
+    assert isinstance(env.factory._gitea, MockGitea)
+
+
+def test_real_billing_core_composition(tmp_path: Path):
+    """The factory composes with the REAL billing core (issue #1, merged).
+
+    Uses the actual ``gitea_saas.billing_core.BillingCore`` over a temp DB
+    -- still no network: Gitea stays mocked.
+    """
+    from gitea_saas.billing_core import BillingCore
+
+    core = BillingCore(tmp_path / "real.db")
+    gitea = MockGitea()
+    cfg = types.SimpleNamespace(db_path=tmp_path / "real.db", api_key="k",
+                                debit_floor_minor=DEFAULT_DEBIT_FLOOR_MINOR)
+    factory = IntakeFactory(core, gitea, cfg)
+
+    ok, details = core.create_client("Acme Corp")
+    assert ok
+    cid = details["id"]
+    assert core.credit(cid, 10_000, "idem-1")[0] is True
+    assert factory.provision_client(cid) == "acme-corp"
+    result = factory.submit_issue(cid, "acme-corp", "Funded with real core", "body")
+    assert result == {"delivered": True, "issue_number": 1}
+
+    ok, details2 = core.create_client("Poor Corp")
+    assert ok
+    cid2 = details2["id"]
+    assert factory.provision_client(cid2) == "poor-corp"
+    queued = factory.submit_issue(cid2, "poor-corp", "Queued", "body")
+    assert queued["delivered"] is False and queued["reason"] == "unfunded"
+
+    flushed = factory.unlock(cid2)
+    assert flushed["flushed"] == 1
+    p1 = gitea.label_id("poor-corp", "p1")
+    assert p1 in gitea.issue("poor-corp", 1)["labels"]