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, FirmwareMetadataResponse, UpdateType COLLECTION = "firmware_versions" VALID_HW_TYPES = {"vesper", "vesper_plus", "vesper_pro", "chronos", "chronos_pro", "agnus", "agnus_mini"} 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", ""), update_type=data.get("update_type", UpdateType.mandatory), min_fw_version=data.get("min_fw_version"), uploaded_at=uploaded_str, notes=data.get("notes"), is_latest=data.get("is_latest", False), ) def _fw_to_metadata_response(fw: FirmwareVersion) -> FirmwareMetadataResponse: download_url = f"/api/firmware/{fw.hw_type}/{fw.channel}/{fw.version}/firmware.bin" return FirmwareMetadataResponse( hw_type=fw.hw_type, channel=fw.channel, version=fw.version, size_bytes=fw.size_bytes, sha256=fw.sha256, update_type=fw.update_type, min_fw_version=fw.min_fw_version, download_url=download_url, uploaded_at=fw.uploaded_at, notes=fw.notes, ) def upload_firmware( hw_type: str, channel: str, version: str, file_bytes: bytes, update_type: UpdateType = UpdateType.mandatory, min_fw_version: str | None = None, 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(sorted(VALID_HW_TYPES))}") if channel not in VALID_CHANNELS: raise HTTPException(status_code=400, detail=f"Invalid channel. Must be one of: {', '.join(sorted(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, "update_type": update_type.value, "min_fw_version": min_fw_version, "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) -> FirmwareMetadataResponse: 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") return _fw_to_metadata_response(_doc_to_firmware_version(docs[0])) def get_version_info(hw_type: str, channel: str, version: str) -> FirmwareMetadataResponse: """Fetch metadata for a specific version. Used by devices resolving upgrade chains.""" 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("version", "==", version) .limit(1) .stream() ) if not docs: raise NotFoundError("Firmware version") return _fw_to_metadata_response(_doc_to_firmware_version(docs[0])) 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})