bellsystems-cp/backend/firmware/service.py

import hashlib
import uuid
from datetime import datetime, timezone
from pathlib import Path

from fastapi import HTTPException

from config import settings
from shared.firebase import get_db
from shared.exceptions import NotFoundError
from firmware.models import FirmwareVersion, FirmwareLatestResponse

COLLECTION = "firmware_versions"

VALID_HW_TYPES = {"vs", "vp", "vx"}
VALID_CHANNELS = {"stable", "beta", "alpha", "testing"}


def _storage_path(hw_type: str, channel: str, version: str) -> Path:
    return Path(settings.firmware_storage_path) / hw_type / channel / version / "firmware.bin"


def _doc_to_firmware_version(doc) -> FirmwareVersion:
    data = doc.to_dict() or {}
    uploaded_raw = data.get("uploaded_at")
    if isinstance(uploaded_raw, datetime):
        uploaded_str = uploaded_raw.strftime("%Y-%m-%dT%H:%M:%SZ")
    else:
        uploaded_str = str(uploaded_raw) if uploaded_raw else ""

    return FirmwareVersion(
        id=doc.id,
        hw_type=data.get("hw_type", ""),
        channel=data.get("channel", ""),
        version=data.get("version", ""),
        filename=data.get("filename", "firmware.bin"),
        size_bytes=data.get("size_bytes", 0),
        sha256=data.get("sha256", ""),
        uploaded_at=uploaded_str,
        notes=data.get("notes"),
        is_latest=data.get("is_latest", False),
    )


def upload_firmware(
    hw_type: str,
    channel: str,
    version: str,
    file_bytes: bytes,
    notes: str | None = None,
) -> FirmwareVersion:
    if hw_type not in VALID_HW_TYPES:
        raise HTTPException(status_code=400, detail=f"Invalid hw_type. Must be one of: {', '.join(VALID_HW_TYPES)}")
    if channel not in VALID_CHANNELS:
        raise HTTPException(status_code=400, detail=f"Invalid channel. Must be one of: {', '.join(VALID_CHANNELS)}")

    dest = _storage_path(hw_type, channel, version)
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_bytes(file_bytes)

    sha256 = hashlib.sha256(file_bytes).hexdigest()
    now = datetime.now(timezone.utc)
    doc_id = str(uuid.uuid4())

    db = get_db()

    # Mark previous latest for this hw_type+channel as no longer latest
    prev_docs = (
        db.collection(COLLECTION)
        .where("hw_type", "==", hw_type)
        .where("channel", "==", channel)
        .where("is_latest", "==", True)
        .stream()
    )
    for prev in prev_docs:
        prev.reference.update({"is_latest": False})

    doc_ref = db.collection(COLLECTION).document(doc_id)
    doc_ref.set({
        "hw_type": hw_type,
        "channel": channel,
        "version": version,
        "filename": "firmware.bin",
        "size_bytes": len(file_bytes),
        "sha256": sha256,
        "uploaded_at": now,
        "notes": notes,
        "is_latest": True,
    })

    return _doc_to_firmware_version(doc_ref.get())


def list_firmware(
    hw_type: str | None = None,
    channel: str | None = None,
) -> list[FirmwareVersion]:
    db = get_db()
    query = db.collection(COLLECTION)
    if hw_type:
        query = query.where("hw_type", "==", hw_type)
    if channel:
        query = query.where("channel", "==", channel)

    docs = list(query.stream())
    items = [_doc_to_firmware_version(doc) for doc in docs]
    items.sort(key=lambda x: x.uploaded_at, reverse=True)
    return items


def get_latest(hw_type: str, channel: str) -> FirmwareLatestResponse:
    if hw_type not in VALID_HW_TYPES:
        raise HTTPException(status_code=400, detail=f"Invalid hw_type '{hw_type}'")
    if channel not in VALID_CHANNELS:
        raise HTTPException(status_code=400, detail=f"Invalid channel '{channel}'")

    db = get_db()
    docs = list(
        db.collection(COLLECTION)
        .where("hw_type", "==", hw_type)
        .where("channel", "==", channel)
        .where("is_latest", "==", True)
        .limit(1)
        .stream()
    )
    if not docs:
        raise NotFoundError("Firmware")

    fw = _doc_to_firmware_version(docs[0])
    download_url = f"/api/firmware/{hw_type}/{channel}/{fw.version}/firmware.bin"
    return FirmwareLatestResponse(
        hw_type=fw.hw_type,
        channel=fw.channel,
        version=fw.version,
        size_bytes=fw.size_bytes,
        sha256=fw.sha256,
        download_url=download_url,
        uploaded_at=fw.uploaded_at,
        notes=fw.notes,
    )


def get_firmware_path(hw_type: str, channel: str, version: str) -> Path:
    path = _storage_path(hw_type, channel, version)
    if not path.exists():
        raise NotFoundError("Firmware binary")
    return path


def delete_firmware(doc_id: str) -> None:
    db = get_db()
    doc_ref = db.collection(COLLECTION).document(doc_id)
    doc = doc_ref.get()
    if not doc.exists:
        raise NotFoundError("Firmware")

    data = doc.to_dict()
    hw_type = data.get("hw_type", "")
    channel = data.get("channel", "")
    version = data.get("version", "")
    was_latest = data.get("is_latest", False)

    # Delete the binary file
    path = _storage_path(hw_type, channel, version)
    if path.exists():
        path.unlink()
    # Remove the version directory if empty
    try:
        path.parent.rmdir()
    except OSError:
        pass

    doc_ref.delete()

    # If we deleted the latest, promote the next most recent as latest
    if was_latest:
        remaining = list(
            db.collection(COLLECTION)
            .where("hw_type", "==", hw_type)
            .where("channel", "==", channel)
            .order_by("uploaded_at", direction="DESCENDING")
            .limit(1)
            .stream()
        )
        if remaining:
            remaining[0].reference.update({"is_latest": True})