backend/core/subscription/api_keys.py

"""Developer API key management."""
from __future__ import annotations

import hashlib
import secrets
import uuid
from datetime import datetime, timezone
from typing import Any

from sqlalchemy.orm import Session

from core.subscription.models import DeveloperApiKey


def _hash_key(raw: str) -> str:
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def _prefix(raw: str) -> str:
    return raw[:8]


def create_api_key(db: Session, user_id: str, label: str) -> tuple[str, dict[str, Any]]:
    raw = f"gf_{secrets.token_urlsafe(32)}"
    row = DeveloperApiKey(
        id=str(uuid.uuid4()),
        user_id=user_id,
        label=label.strip(),
        key_prefix=_prefix(raw),
        key_hash=_hash_key(raw),
    )
    db.add(row)
    db.commit()
    db.refresh(row)
    meta = {
        "id": row.id,
        "label": row.label,
        "key_prefix": row.key_prefix,
        "created_at": row.created_at.isoformat() if row.created_at else None,
    }
    return raw, meta


def list_api_keys(db: Session, user_id: str) -> list[dict[str, Any]]:
    rows = (
        db.query(DeveloperApiKey)
        .filter(DeveloperApiKey.user_id == user_id, DeveloperApiKey.revoked_at.is_(None))
        .order_by(DeveloperApiKey.created_at.desc())
        .all()
    )
    return [
        {
            "id": r.id,
            "label": r.label,
            "key_prefix": r.key_prefix,
            "created_at": r.created_at.isoformat() if r.created_at else None,
            "last_used_at": r.last_used_at.isoformat() if r.last_used_at else None,
        }
        for r in rows
    ]


def verify_api_key_hash(db: Session, raw_key: str) -> dict[str, Any] | None:
    digest = _hash_key(raw_key)
    row = (
        db.query(DeveloperApiKey)
        .filter(DeveloperApiKey.key_hash == digest, DeveloperApiKey.revoked_at.is_(None))
        .first()
    )
    if not row:
        return None
    row.last_used_at = datetime.now(timezone.utc)
    db.commit()
    return {"user_id": row.user_id, "key_id": row.id, "label": row.label}


def revoke_api_key(db: Session, key_id: str, user_id: str) -> bool:
    row = (
        db.query(DeveloperApiKey)
        .filter(DeveloperApiKey.id == key_id, DeveloperApiKey.user_id == user_id)
        .first()
    )
    if not row or row.revoked_at:
        return False
    row.revoked_at = datetime.now(timezone.utc)
    db.commit()
    return True