app/backend/shared/persimmons.py

from __future__ import annotations

import json
import re
import secrets
from datetime import datetime, timezone
from typing import Any, Awaitable, Callable

LEGACY_REPO = "matcha/persimmons"
META_PREFIX = "PERSIM_META_"
SECRET_PREFIX = "PERSIM_SECRET_"
JUICE_PREFIX = "PERSIM_JUICE_"
ID_RE = re.compile(r"[A-F0-9]{12}")
REPO_RE = re.compile(r"[A-Za-z0-9_.-]{1,100}/[A-Za-z0-9_.-]{1,100}")
SECRET_NAME_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
LEGACY_ID_PREFIX = "secret:"
GiteaRequest = Callable[..., Awaitable[Any]]


def utc_now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def connection_id(value: str) -> str:
    value = str(value or "").strip().upper()
    if not ID_RE.fullmatch(value):
        raise ValueError("invalid connection id")
    return value

def secret_name(value: str) -> str:
    value = re.sub(r"[^A-Za-z0-9_]+", "_", str(value or "").strip().upper()).strip("_")
    if value and value[0].isdigit():
        value = f"P_{value}"
    value = value[:100]
    if not value or not SECRET_NAME_RE.fullmatch(value):
        raise ValueError("valid secret name is required")
    return value


def metadata_secret_name(metadata: dict[str, Any], cid: str) -> str:
    stored = str(metadata.get("secret_name") or "").strip()
    if stored:
        return secret_name(stored)
    return f"{SECRET_PREFIX}{connection_id(cid)}"



def repository_name(value: str) -> str:
    value = str(value or "").strip()
    if not REPO_RE.fullmatch(value):
        raise ValueError("invalid repository")
    return value


def _items(data: Any, key: str) -> list[dict[str, Any]]:
    if isinstance(data, list):
        return [item for item in data if isinstance(item, dict)]
    if isinstance(data, dict) and isinstance(data.get(key), list):
        return [item for item in data[key] if isinstance(item, dict)]
    return []


def _scopes(value: Any) -> list[str]:
    raw = value.split(",") if isinstance(value, str) else value if isinstance(value, list) else []
    result = [str(item).strip() for item in raw if str(item).strip()]
    if len(result) > 40 or any(len(item) > 100 for item in result):
        raise ValueError("invalid scopes")
    return result


def vault_from_input(data: dict[str, Any]) -> dict[str, Any]:
    raw_kind = str(data.get("vault_kind") or "").strip().lower()
    if not raw_kind:
        return {"kind": "repository", "repository": LEGACY_REPO, "implicit_legacy": True}
    if raw_kind == "user":
        return {"kind": "user"}
    if raw_kind == "repository":
        return {"kind": "repository", "repository": repository_name(str(data.get("repository") or ""))}
    raise ValueError("invalid vault")


def clean_vault(vault: dict[str, Any]) -> dict[str, Any]:
    if vault["kind"] == "user":
        return {"kind": "user"}
    return {"kind": "repository", "repository": vault["repository"]}


def metadata_from_input(data: dict[str, Any], *, vault: dict[str, Any], created_at: str | None = None) -> dict[str, Any]:
    provider = str(data.get("provider") or "").strip()
    credential_type = str(data.get("credential_type") or "").strip()
    account = str(data.get("account") or "").strip()
    name = str(data.get("name") or "").strip() or provider
    notes = str(data.get("notes") or "").strip()
    expires = str(data.get("expires") or "").strip() or None
    if not provider or not credential_type:
        raise ValueError("provider and credential type are required")
    for value in (name, provider, account, credential_type):
        if len(value) > 200:
            raise ValueError("field too long")
    if len(notes) > 1000:
        raise ValueError("notes too long")
    now = utc_now()
    return {
        "kind": "credential",
        "name": name,
        "provider": provider,
        "account": account,
        "credential_type": credential_type,
        "scopes": _scopes(data.get("scopes")),
        "created_at": created_at or now,
        "updated_at": now,
        "expires": expires,
        "notes": notes,
        "status": "active",
        "vault": clean_vault(vault),
    }


def _metadata(item: dict[str, Any]) -> dict[str, Any]:
    raw = item.get("data") if item.get("data") is not None else item.get("value")
    try:
        value = json.loads(str(raw or "{}"))
    except json.JSONDecodeError as exc:
        raise ValueError("invalid Persimmons metadata") from exc
    if not isinstance(value, dict):
        raise ValueError("invalid Persimmons metadata")
    return value


def metadata_vault(metadata: dict[str, Any], fallback_repository: str | None = None) -> dict[str, Any]:
    vault = metadata.get("vault")
    if isinstance(vault, dict) and vault.get("kind") == "user":
        return {"kind": "user"}
    if isinstance(vault, dict) and vault.get("kind") == "repository":
        return {"kind": "repository", "repository": repository_name(str(vault.get("repository") or ""))}
    return {"kind": "repository", "repository": fallback_repository or LEGACY_REPO}


def public_connection(cid: str, metadata: dict[str, Any], *, available: bool, fallback_repository: str | None = None) -> dict[str, Any]:
    vault = metadata_vault(metadata, fallback_repository)
    stored_name = metadata_secret_name(metadata, cid)
    display_name = str(metadata.get("name") or metadata.get("provider") or stored_name)
    return {
        "id": cid,
        "kind": str(metadata.get("kind") or "credential"),
        "name": display_name,
        "secret_name": stored_name,
        "provider": str(metadata.get("provider") or ""),
        "account": str(metadata.get("account") or ""),
        "credential_type": str(metadata.get("credential_type") or ""),
        "scopes": metadata.get("scopes") if isinstance(metadata.get("scopes"), list) else [],
        "created_at": metadata.get("created_at"),
        "updated_at": metadata.get("updated_at"),
        "expires": metadata.get("expires") or None,
        "notes": str(metadata.get("notes") or ""),
        "status": "active" if available else "revoked",
        "available": bool(available),
        "vault_kind": vault["kind"],
        "repository": vault.get("repository"),
    }


def legacy_secret_name(value: str) -> str | None:
    raw = str(value or "")
    if not raw.startswith(LEGACY_ID_PREFIX):
        return None
    name = raw[len(LEGACY_ID_PREFIX):]
    if not SECRET_NAME_RE.fullmatch(name):
        raise ValueError("invalid legacy secret")
    return name


def public_legacy_secret(
    name: str,
    vault: dict[str, Any],
    *,
    available: bool = True,
    created_at: str | None = None,
    description: str = "",
) -> dict[str, Any]:
    if not SECRET_NAME_RE.fullmatch(name):
        raise ValueError("invalid legacy secret")
    return {
        "id": f"{LEGACY_ID_PREFIX}{name}",
        "kind": "credential",
        "name": name,
        "provider": "",
        "account": "",
        "credential_type": "Existing Gitea secret",
        "scopes": [],
        "created_at": created_at,
        "updated_at": created_at,
        "expires": None,
        "notes": description,
        "status": "active" if available else "revoked",
        "available": bool(available),
        "vault_kind": vault["kind"],
        "repository": vault.get("repository"),
        "legacy": True,
        "secret_name": name,
    }


def _prefix(vault: dict[str, Any]) -> str:
    return "user/actions" if vault["kind"] == "user" else f"repos/{vault['repository']}/actions"


async def _validate_repository(request: GiteaRequest, repository: str) -> dict[str, Any]:
    repo = await request("GET", f"repos/{repository}")
    if not isinstance(repo, dict):
        raise ValueError("repository unavailable")
    permissions = repo.get("permissions") if isinstance(repo.get("permissions"), dict) else {}
    if permissions and not (permissions.get("admin") or permissions.get("push")):
        raise ValueError("repository is not writable")
    return repo


async def list_repositories(request: GiteaRequest) -> dict[str, Any]:
    repos: list[dict[str, Any]] = []
    for page in range(1, 6):
        data = await request("GET", "user/repos", params={"limit": 100, "page": page})
        batch = _items(data, "data")
        if not batch:
            break
        for repo in batch:
            full_name = str(repo.get("full_name") or "")
            permissions = repo.get("permissions") if isinstance(repo.get("permissions"), dict) else {}
            if full_name and (not permissions or permissions.get("admin") or permissions.get("push")):
                repos.append({
                    "full_name": full_name,
                    "private": bool(repo.get("private")),
                    "owner": str((repo.get("owner") or {}).get("login") or ""),
                })
        if len(batch) < 100:
            break
    repos.sort(key=lambda item: item["full_name"].lower())
    return {"repositories": repos}


async def _variable_get(request: GiteaRequest, vault: dict[str, Any], cid: str, *, prefix: str = META_PREFIX) -> dict[str, Any] | None:
    try:
        return await request("GET", f"{_prefix(vault)}/variables/{prefix}{cid}")
    except RuntimeError as exc:
        if "HTTP 404" in str(exc):
            return None
        raise


async def _variable_put(request: GiteaRequest, vault: dict[str, Any], cid: str, metadata: dict[str, Any], *, exists: bool, prefix: str = META_PREFIX) -> None:
    name = f"{prefix}{cid}"
    path = f"{_prefix(vault)}/variables/{name}"
    body = {
        "value": json.dumps(metadata, separators=(",", ":"), ensure_ascii=False),
        "description": "Persimmons object metadata. Contains no protected value.",
    }
    await request("PUT" if exists else "POST", path, json=body)


async def _list_vault(request: GiteaRequest, vault: dict[str, Any]) -> list[dict[str, Any]]:
    if vault["kind"] == "repository" and vault["repository"] != LEGACY_REPO:
        await _validate_repository(request, vault["repository"])
    variable_data = await request("GET", f"{_prefix(vault)}/variables", params={"limit": 100})

    secret_items: list[dict[str, Any]] = []
    secret_listing_supported = True
    try:
        secret_data = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100})
        secret_items = _items(secret_data, "secrets")
    except RuntimeError as exc:
        if vault["kind"] == "user" and "HTTP 404" in str(exc):
            secret_listing_supported = False
        else:
            raise
    secret_names = {
        str(item.get("name") or "")
        for item in secret_items
        if str(item.get("name") or "")
    }

    result: list[dict[str, Any]] = []
    metadata_secret_names: set[str] = set()
    for item in _items(variable_data, "variables"):
        name = str(item.get("name") or "")
        if not name.startswith(META_PREFIX):
            continue
        cid = name[len(META_PREFIX):]
        if not ID_RE.fullmatch(cid):
            continue
        try:
            metadata = _metadata(item)
        except ValueError:
            continue
        stored_name = metadata_secret_name(metadata, cid)
        metadata_secret_names.add(stored_name)
        available = (
            stored_name in secret_names
            if secret_listing_supported
            else str(metadata.get("status") or "active") == "active"
        )
        result.append(
            public_connection(
                cid,
                metadata,
                available=available,
                fallback_repository=vault.get("repository"),
            )
        )

    for item in secret_items:
        name = str(item.get("name") or "")
        if not name or name in metadata_secret_names or not SECRET_NAME_RE.fullmatch(name):
            continue
        result.append(
            public_legacy_secret(
                name,
                clean_vault(vault),
                available=True,
                created_at=str(item.get("created_at") or "") or None,
                description=str(item.get("description") or ""),
            )
        )
    return result


async def list_connections(
    request: GiteaRequest,
    *,
    repository: str | None = None,
    include_legacy: bool = True,
    include_user: bool = True,
) -> dict[str, Any]:
    result: list[dict[str, Any]] = []
    if include_user:
        result.extend(await _list_vault(request, {"kind": "user"}))
    target = repository_name(repository) if repository else None
    if target:
        result.extend(await _list_vault(request, {"kind": "repository", "repository": target}))
    elif include_legacy:
        try:
            result.extend(await _list_vault(request, {"kind": "repository", "repository": LEGACY_REPO}))
        except (RuntimeError, ValueError):
            pass
    result.sort(key=lambda item: str(item.get("updated_at") or item.get("created_at") or ""), reverse=True)
    return {"connections": result}


async def create_connection(request: GiteaRequest, data: dict[str, Any]) -> dict[str, Any]:
    credential = str(data.get("credential") or "")
    if not credential or len(credential) > 16384:
        raise ValueError("valid credential value is required")
    vault = vault_from_input(data)
    if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
        await _validate_repository(request, vault["repository"])
    cid = secrets.token_hex(6).upper()
    metadata = metadata_from_input(data, vault=vault)
    stored_name = secret_name(metadata["name"])
    metadata["name"] = stored_name
    metadata["secret_name"] = stored_name

    existing_data = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100})
    existing_names = {
        str(item.get("name") or "")
        for item in _items(existing_data, "secrets")
        if str(item.get("name") or "")
    }
    if stored_name in existing_names:
        raise ValueError(f"secret {stored_name} already exists")

    secret_path = f"{_prefix(vault)}/secrets/{stored_name}"
    await request("PUT", secret_path, json={"data": credential, "description": "Persimmons write-only credential"})
    try:
        await _variable_put(request, vault, cid, metadata, exists=False)
    except Exception:
        try:
            await request("DELETE", secret_path)
        except Exception:
            pass
        raise
    return public_connection(cid, metadata, available=True)


async def _existing(request: GiteaRequest, cid: str, vault: dict[str, Any]) -> dict[str, Any]:
    if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
        await _validate_repository(request, vault["repository"])
    variable = await _variable_get(request, vault, cid)
    if variable is None:
        raise ValueError("connection not found")
    metadata = _metadata(variable)
    if metadata_vault(metadata, vault.get("repository")) != clean_vault(vault):
        raise ValueError("connection vault does not match")
    return metadata


async def rotate_connection(request: GiteaRequest, cid: str, credential: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]:
    credential = str(credential or "")
    if not credential or len(credential) > 16384:
        raise ValueError("valid credential value is required")
    vault = vault_from_input(vault_data or {})
    legacy_name = legacy_secret_name(cid)
    if legacy_name:
        if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
            await _validate_repository(request, vault["repository"])
        await request(
            "PUT",
            f"{_prefix(vault)}/secrets/{legacy_name}",
            json={"data": credential, "description": "Existing Gitea secret managed by Persimmons"},
        )
        return public_legacy_secret(legacy_name, clean_vault(vault), available=True)
    cid = connection_id(cid)
    metadata = await _existing(request, cid, vault)
    stored_name = metadata_secret_name(metadata, cid)
    await request("PUT", f"{_prefix(vault)}/secrets/{stored_name}", json={"data": credential, "description": "Persimmons write-only credential"})
    metadata["updated_at"] = utc_now()
    metadata["status"] = "active"
    await _variable_put(request, vault, cid, metadata, exists=True)
    return public_connection(cid, metadata, available=True)


async def revoke_connection(request: GiteaRequest, cid: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]:
    vault = vault_from_input(vault_data or {})
    legacy_name = legacy_secret_name(cid)
    if legacy_name:
        if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
            await _validate_repository(request, vault["repository"])
        try:
            await request("DELETE", f"{_prefix(vault)}/secrets/{legacy_name}")
        except RuntimeError as exc:
            if "HTTP 404" not in str(exc):
                raise
        return public_legacy_secret(legacy_name, clean_vault(vault), available=False)
    cid = connection_id(cid)
    metadata = await _existing(request, cid, vault)
    stored_name = metadata_secret_name(metadata, cid)
    try:
        await request("DELETE", f"{_prefix(vault)}/secrets/{stored_name}")
    except RuntimeError as exc:
        if "HTTP 404" not in str(exc):
            raise
    metadata["updated_at"] = utc_now()
    metadata["status"] = "revoked"
    await _variable_put(request, vault, cid, metadata, exists=True)
    return public_connection(cid, metadata, available=False)


async def delete_connection(request: GiteaRequest, cid: str, vault_data: dict[str, Any] | None = None) -> dict[str, Any]:
    vault = vault_from_input(vault_data or {})
    legacy_name = legacy_secret_name(cid)
    if legacy_name:
        if vault["kind"] == "repository" and not vault.get("implicit_legacy"):
            await _validate_repository(request, vault["repository"])
        try:
            await request("DELETE", f"{_prefix(vault)}/secrets/{legacy_name}")
        except RuntimeError as exc:
            if "HTTP 404" not in str(exc):
                raise
        return {
            "id": cid,
            "deleted": True,
            "vault_kind": vault["kind"],
            "repository": vault.get("repository"),
            "legacy": True,
        }
    cid = connection_id(cid)
    metadata = await _existing(request, cid, vault)
    stored_name = metadata_secret_name(metadata, cid)
    for path in (
        f"{_prefix(vault)}/secrets/{stored_name}",
        f"{_prefix(vault)}/variables/{META_PREFIX}{cid}",
    ):
        try:
            await request("DELETE", path)
        except RuntimeError as exc:
            if "HTTP 404" not in str(exc):
                raise
    return {"id": cid, "deleted": True, "vault_kind": vault["kind"], "repository": vault.get("repository")}


def juice_vault(data: dict[str, Any]) -> dict[str, Any]:
    """Juice defaults to My vault, never the implicit legacy repository."""
    repository = data.get("repository")
    kind = data.get("vault_kind") or ("repository" if repository else "user")
    if kind == "user" and repository:
        raise ValueError("Juice vault does not match repository")
    return vault_from_input({"vault_kind": kind, "repository": repository})


def _juice_policy(data: dict[str, Any]) -> dict[str, Any]:
    subject = data.get("subject")
    scopes = data.get("scopes")
    if not isinstance(subject, str) or not subject.strip() or len(subject) > 200:
        raise ValueError("valid subject is required (maximum 200 characters)")
    if not isinstance(scopes, list) or not scopes or any(
        not isinstance(scope, str) or not scope.strip() for scope in scopes
    ):
        raise ValueError("nonempty scopes array is required")
    expires = data.get("expires_at")
    if expires is not None:
        try:
            if not isinstance(expires, str) or len(expires) > 40:
                raise ValueError
            parsed = datetime.fromisoformat(expires.replace("Z", "+00:00"))
            if parsed.tzinfo is None:
                raise ValueError
            expires = parsed.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
        except ValueError as exc:
            raise ValueError("expires_at must be a timestamp with timezone or null") from exc
    return {"subject": subject.strip(), "scopes": _scopes(scopes), "expires_at": expires}


def public_juice(jid: str, metadata: dict[str, Any], vault: dict[str, Any]) -> dict[str, Any]:
    if metadata.get("vault") != clean_vault(vault):
        raise ValueError("Juice vault does not match")
    if metadata.get("kind") != "juice" or metadata.get("status") not in {"active", "revoked"}:
        raise ValueError("invalid Juice metadata")
    source = metadata.get("source")
    if not isinstance(source, str) or not source or len(source) > 200:
        raise ValueError("invalid Juice source")
    # Explicit projection: never return arbitrary stored fields or source metadata.
    return {
        "id": jid, "kind": "juice",
        "source_id": connection_id(metadata.get("source_id")), "source": source,
        **_juice_policy(metadata), "created_at": metadata.get("created_at"),
        "status": metadata["status"], "revocable": True,
        "vault": clean_vault(vault), "vault_kind": vault["kind"],
        "repository": vault.get("repository"),
    }


async def create_juice(request: GiteaRequest, data: dict[str, Any]) -> dict[str, Any]:
    vault = juice_vault(data)
    cid = connection_id(data.get("source_id"))
    policy = _juice_policy(data)
    if policy["expires_at"] is not None and datetime.fromisoformat(
        policy["expires_at"].replace("Z", "+00:00")
    ) <= datetime.now(timezone.utc):
        raise ValueError("expires_at must be in the future")
    source = await _existing(request, cid, vault)
    if source.get("status", "active") != "active":
        raise ValueError("source connection is revoked")
    # Gitea's collection returns names only; no individual secret GET is allowed.
    try:
        names = await request("GET", f"{_prefix(vault)}/secrets", params={"limit": 100})
        if metadata_secret_name(source, cid) not in {item.get("name") for item in _items(names, "secrets")}:
            raise ValueError("source connection is unavailable")
    except RuntimeError as exc:
        # Older Gitea user APIs lack secret listing; use the saved active metadata.
        if vault["kind"] != "user" or "HTTP 404" not in str(exc):
            raise
    jid = secrets.token_hex(6).upper()
    metadata = public_juice(jid, {
        "kind": "juice", "source_id": cid, "source": source.get("provider"),
        **policy, "created_at": utc_now(), "status": "active", "vault": clean_vault(vault),
    }, vault)
    await _variable_put(request, vault, jid, metadata, exists=False, prefix=JUICE_PREFIX)
    return metadata


async def list_juice(
    request: GiteaRequest, *, repository: str | None = None, source_id: str | None = None,
) -> dict[str, Any]:
    vault = juice_vault({"repository": repository})
    cid = connection_id(source_id) if source_id is not None else None
    if vault["kind"] == "repository":
        await _validate_repository(request, vault["repository"])
    variables = await request("GET", f"{_prefix(vault)}/variables", params={"limit": 100})
    result = []
    for item in _items(variables, "variables"):
        name = str(item.get("name") or "")
        if not name.startswith(JUICE_PREFIX) or not ID_RE.fullmatch(name[len(JUICE_PREFIX):]):
            continue
        try:
            juice = public_juice(name[len(JUICE_PREFIX):], _metadata(item), vault)
        except ValueError:
            continue
        if cid is None or juice["source_id"] == cid:
            result.append(juice)
    result.sort(key=lambda item: str(item.get("created_at") or ""), reverse=True)
    return {"juice": result}


async def revoke_juice(
    request: GiteaRequest, jid: str, vault_data: dict[str, Any] | None = None,
) -> dict[str, Any]:
    jid = connection_id(jid)
    vault = juice_vault(vault_data or {})
    if vault["kind"] == "repository":
        await _validate_repository(request, vault["repository"])
    variable = await _variable_get(request, vault, jid, prefix=JUICE_PREFIX)
    if variable is None:
        raise ValueError("Juice not found")
    metadata = public_juice(jid, _metadata(variable), vault)
    metadata["status"] = "revoked"
    await _variable_put(request, vault, jid, metadata, exists=True, prefix=JUICE_PREFIX)
    return metadata