Source code for widgets.widgets_registry

# File: widgets_registry.py
from __future__ import annotations

import asyncio
import atexit
import io
import logging
import os
import re
import tarfile
import urllib.parse
from collections import OrderedDict
from dataclasses import dataclass
from pathlib import Path
from typing import Any

import aiohttp
from aiohttp import ClientSession, ClientTimeout, TCPConnector
from bson import ObjectId
from motor.motor_asyncio import AsyncIOMotorClient

from widgets.widgets_registry_metadata import (
    artefact_latest_from_metadata as _artefact_latest_from_metadata_impl,
)
from widgets.widgets_registry_metadata import (
    npm_latest_from_metadata as _npm_latest_from_metadata_impl,
)
from widgets.widgets_registry_metadata import (
    npm_metadata_url as _npm_metadata_url_impl,
)
from widgets.widgets_registry_metadata import (
    resolve_artefact_requested_version as _resolve_artefact_requested_meta,
)
from widgets.widgets_registry_metadata import (
    resolve_npm_requested_version as _resolve_npm_requested_version_impl,
)
from widgets.widgets_registry_models import WidgetDoc, _derive_key
from widgets.widgets_registry_parsing import parse_source as _parse_source_impl
from widgets.widgets_registry_security import (
    check_url_allowed as _check_url_allowed_impl,
)
from widgets.widgets_registry_security import (
    ensure_pkg_allowed_scope as _ensure_pkg_allowed_scope_impl,
)
from widgets.widgets_registry_security import (
    extract_pkg_from_cdn_path as _extract_pkg_from_cdn_path_impl,
)
from widgets.widgets_registry_sources import build_sources_payload
from widgets.widgets_registry_utils import (
    artefact_metadata_url as _artefact_metadata_url_impl,
)
from widgets.widgets_registry_utils import (
    artefact_tarball_url as _artefact_tarball_url_impl,
)
from widgets.widgets_registry_utils import is_newer as _is_newer_impl
from widgets.widgets_registry_utils import (
    parse_seed_list as _parse_seed_list_impl,
)
from widgets.widgets_registry_utils import (
    scrub_mongo_uri as _scrub_mongo_uri_impl,
)
from widgets.widgets_registry_utils import semver_tuple as _semver_tuple_impl
from widgets.widgets_registry_utils import sha256_hex as _sha256_impl
from widgets.widgets_registry_utils import utc_now_iso as _utc_now_iso_impl

# ───────────── env & globals ─────────────
MONGO_URI = os.getenv("MONGO_URI", "mongodb://mongo:27017")
MONGO_DB = os.getenv("MONGO_DB", "octopus")
WIDGET_COLLECTION = os.getenv("WIDGET_COLLECTION", "widget_versions")
WIDGET_CACHE_DIR = Path(os.getenv("WIDGET_CACHE_DIR", "widgets-cache"))

# Allowed hosts – can be narrowed via env (e.g. "cdn.jsdelivr.net")
WIDGET_ALLOWED_HOSTS = {
    h.strip()
    for h in (os.getenv("WIDGET_ALLOWED_HOSTS", "")).split(",")
    if h.strip()
} or {"cdn.jsdelivr.net", "unpkg.com", "artefact.skao.int"}

# Only allow npm packages under these scopes (default hardens to SKA org)
WIDGET_ALLOWED_NPM_SCOPES = {
    s.strip()
    for s in (
        os.getenv("WIDGET_ALLOWED_NPM_SCOPES", "@ska-octopus-widgets")
    ).split(",")
    if s.strip()
}

ARTEFACT_REGISTRY_BASE = os.getenv(
    "WIDGET_ARTEFACT_BASE", "https://artefact.skao.int/repository/npm-internal"
)

WIDGET_SEED = os.getenv("WIDGET_SEED", None)

# In-memory LRU cache (hot path) controls
_MEM_MAX = int(os.getenv("WIDGET_MEM_CACHE_MAX", "64"))  # number of bundles
_PREWARM = os.getenv("WIDGET_PREWARM", "true").lower() in (
    "1",
    "true",
    "yes",
    "on",
)
_PREWARM_CONC = int(os.getenv("WIDGET_PREWARM_CONCURRENCY", "6"))

_client = AsyncIOMotorClient(MONGO_URI)
_coll = _client[MONGO_DB][WIDGET_COLLECTION]
_ready_once: asyncio.Event | None = None

# Shared HTTP session (connection pooled)
_session: ClientSession | None = None
_log = logging.getLogger(__name__)


def _scrub_mongo_uri(uri: str) -> str:
    return _scrub_mongo_uri_impl(uri)


# In-memory LRU cache and in-flight fetch deduplication
@dataclass
class _MemEntry:
    version: str  # 'latest' or concrete semver
    code: str  # JS source
    integrity: str | None


_MEM: OrderedDict[str, _MemEntry] = OrderedDict()
_INFLIGHT: dict[str, asyncio.Task[tuple[str, str | None]]] = {}


# ───────────── helpers ─────────────
def _utc_now_iso() -> str:
    return _utc_now_iso_impl()


def _semver_tuple(v: str | None) -> tuple[int, int, int]:
    return _semver_tuple_impl(v)


def _is_newer(latest: str | None, current: str | None) -> bool:
    return _is_newer_impl(latest, current)


def _sha256(data: bytes) -> str:
    return _sha256_impl(data)


def _artefact_metadata_url(pkg: str) -> str:
    return _artefact_metadata_url_impl(ARTEFACT_REGISTRY_BASE, pkg)


def _artefact_tarball_url(pkg: str, version: str) -> str:
    return _artefact_tarball_url_impl(ARTEFACT_REGISTRY_BASE, pkg, version)


def _parse_seed_list(raw: str) -> list[str]:
    return _parse_seed_list_impl(raw)


def _mem_key(doc_id: str, version: str | None, updated_at: str | None) -> str:
    """
    Build a memory cache key including updated_at to invalidate on changes.
    """
    return f"{doc_id}:{version or 'latest'}:{updated_at or ''}"


def _mem_get(key: str) -> _MemEntry | None:
    entry = _MEM.pop(key, None)
    if entry is not None:
        # move to MRU
        _MEM[key] = entry
    return entry


def _mem_put(key: str, entry: _MemEntry) -> None:
    if key in _MEM:
        del _MEM[key]
    _MEM[key] = entry
    # evict LRU
    while len(_MEM) > _MEM_MAX:
        _MEM.popitem(last=False)


def _invalidate_mem_for(doc_id: str) -> None:
    # remove all entries for this id
    to_del = [k for k in _MEM.keys() if k.startswith(f"{doc_id}:")]
    for k in to_del:
        _MEM.pop(k, None)


async def _get_session() -> (
    ClientSession
):  # pragma: no cover - network session setup
    global _session
    if _session is None:
        connector = TCPConnector(limit=40, limit_per_host=8, ttl_dns_cache=300)
        timeout = ClientTimeout(total=30)
        _session = aiohttp.ClientSession(connector=connector, timeout=timeout)
    return _session


# ───────────── scope/org enforcement ─────────────
def _ensure_pkg_allowed_scope(pkg: str) -> None:
    _ensure_pkg_allowed_scope_impl(pkg, WIDGET_ALLOWED_NPM_SCOPES)


def _extract_pkg_from_cdn_path(host: str, path: str) -> str | None:
    return _extract_pkg_from_cdn_path_impl(host, path)


def _check_url_allowed(url: str) -> None:
    _check_url_allowed_impl(
        url,
        WIDGET_ALLOWED_HOSTS,
        WIDGET_ALLOWED_NPM_SCOPES,
    )


# ───────────── model ─────────────
# ───────────── fetchers ─────────────
async def _get(
    session: aiohttp.ClientSession, url: str
) -> bytes:  # pragma: no cover - network I/O
    # Allow only known hosts and org-scoped paths for direct URL loads
    _check_url_allowed(url)
    async with session.get(url) as r:
        r.raise_for_status()
        return await r.read()


def _npm_metadata_url(pkg: str) -> str:
    return _npm_metadata_url_impl(pkg)


def _npm_latest_from_metadata(data: dict[str, Any]) -> str | None:
    return _npm_latest_from_metadata_impl(data)


def _resolve_npm_requested_version(
    data: dict[str, Any], requested: str
) -> str | None:
    return _resolve_npm_requested_version_impl(data, requested)


async def _npm_metadata(
    session: aiohttp.ClientSession, pkg: str
) -> dict[str, Any] | None:
    # Guard: only allowed scopes
    try:
        _ensure_pkg_allowed_scope(pkg)
    except Exception:
        return None

    url = _npm_metadata_url(pkg)
    _log.info("npm metadata lookup for %s via %s", pkg, url)
    try:
        async with session.get(url) as r:
            if r.status != 200:
                _log.warning(
                    "npm metadata request failed for %s (status %s)",
                    pkg,
                    r.status,
                )
                return None
            data = await r.json()
            if isinstance(data, dict):
                return data
            return None
    except Exception:
        _log.exception("npm metadata request error for %s", pkg)
        return None


async def _npm_latest(
    session: aiohttp.ClientSession, pkg: str
) -> str | None:  # pragma: no cover - network I/O
    data = await _npm_metadata(session, pkg)
    if not data:
        return None
    return _npm_latest_from_metadata(data)


def _artefact_latest_from_metadata(data: dict[str, Any]) -> str | None:
    return _artefact_latest_from_metadata_impl(data)


def _resolve_artefact_requested_version(
    data: dict[str, Any], requested: str
) -> str | None:
    return _resolve_artefact_requested_meta(data, requested)


async def _artefact_metadata(
    session: aiohttp.ClientSession, pkg: str
) -> dict[str, Any] | None:
    try:
        _ensure_pkg_allowed_scope(pkg)
    except Exception:
        return None

    url = _artefact_metadata_url(pkg)
    _log.info("artefact metadata lookup for %s via %s", pkg, url)
    try:
        async with session.get(url) as resp:
            if resp.status != 200:
                _log.warning(
                    "Artefact metadata fetch failed for %s (status %s)",
                    pkg,
                    resp.status,
                )
                return None
            data = await resp.json()
            if isinstance(data, dict):
                return data
            return None
    except Exception:
        _log.exception("Artefact metadata fetch error for %s", pkg)
        return None


async def _artefact_latest(
    session: aiohttp.ClientSession, pkg: str
) -> str | None:  # pragma: no cover - network I/O
    data = await _artefact_metadata(session, pkg)
    if not data:
        return None
    latest = _artefact_latest_from_metadata(data)
    if latest:
        _log.info("Artefact latest for %s -> %s", pkg, latest)
    else:
        _log.info("Artefact metadata missing versions for %s", pkg)
    return latest


async def _resolve_source_version(
    session: aiohttp.ClientSession,
    source_type: str,
    pkg: str,
    requested_version: str,
) -> str:
    """
    Resolve a requested version/dist-tag to an exact version and reject
    unknown versions so invalid sources are never persisted.
    """
    if source_type == "npm":
        data = await _npm_metadata(session, pkg)
        if not data:
            raise ValueError(
                f"Could not verify package '{pkg}' in npm registry"
            )
        resolved = _resolve_npm_requested_version(data, requested_version)
        latest = _npm_latest_from_metadata(data)
    elif source_type == "artefact":
        data = await _artefact_metadata(session, pkg)
        if not data:
            raise ValueError(
                f"Could not verify package '{pkg}' in artefact registry"
            )
        resolved = _resolve_artefact_requested_version(data, requested_version)
        latest = _artefact_latest_from_metadata(data)
    else:
        raise ValueError(
            "Version validation supports only npm or artefact widgets"
        )

    if resolved:
        return resolved

    if latest:
        raise ValueError(
            f"Version '{requested_version}' does not exist for package "
            f"'{pkg}'. Latest available is '{latest}'."
        )
    raise ValueError(
        f"Version '{requested_version}' does not exist for package '{pkg}'."
    )


async def _pick_iife(  # pragma: no cover - network + tar processing
    session: aiohttp.ClientSession, pkg: str, ver: str
) -> tuple[str, bytes]:
    _ensure_pkg_allowed_scope(pkg)

    # 1. Try jsDelivr
    url = f"https://cdn.jsdelivr.net/npm/{pkg}@{ver}/dist/index.iife.js"
    _log.info("bundle lookup jsDelivr %s@%s via %s", pkg, ver, url)
    try:
        data = await _get(session, url)
        return url, data
    except Exception:
        pass

    # 2. Try unpkg
    url2 = f"https://unpkg.com/{pkg}@{ver}/dist/index.iife.js"
    _log.info("bundle lookup unpkg %s@%s via %s", pkg, ver, url2)
    try:
        data = await _get(session, url2)
        return url2, data
    except Exception:
        pass

    # 3. Try generic npm tarball URLs for registries (GitLab, Nexus, etc.).
    # Example:
    #   https://gitlab.com/api/v4/projects/<id>/packages/npm/<pkg>/-/
    #   <pkg>-<ver>.tgz
    for host in WIDGET_ALLOWED_HOSTS:
        pkg_leaf = pkg.split("/")[-1]
        tgz_url = f"https://{host}/{pkg}/-/{pkg_leaf}-{ver}.tgz"
        _log.info("bundle lookup tarball %s@%s via %s", pkg, ver, tgz_url)
        try:
            tgz_bytes = await _get(session, tgz_url)
            with tarfile.open(
                fileobj=io.BytesIO(tgz_bytes), mode="r:gz"
            ) as tar:
                member = next(
                    m
                    for m in tar.getmembers()
                    if m.name.endswith("dist/index.iife.js")
                )
                file_bytes = tar.extractfile(member).read()
                return tgz_url, file_bytes
        except Exception:
            continue

    raise ValueError(f"Could not resolve IIFE bundle for {pkg}@{ver}")


async def _artefact_pick_iife(
    session: aiohttp.ClientSession, pkg: str, ver: str
) -> tuple[str, bytes]:  # pragma: no cover - network I/O
    _ensure_pkg_allowed_scope(pkg)
    if not ver:
        raise ValueError("Artefact widgets require a pinned version")

    url = _artefact_tarball_url(pkg, ver)
    _log.info("artefact.bundle fetch %s@%s via %s", pkg, ver, url)
    data = await _get(session, url)
    return url, data


# ───────────── persistence / lifecycle ─────────────


async def ensure_ready() -> None:  # pragma: no cover - DB + background prewarm
    """
    Create indexes, seed from WIDGET_SEED once per process, and start
    best-effort prewarming of pinned npm bundles.
    """
    global _ready_once
    if _ready_once is None:
        _ready_once = asyncio.Event()

        # connectivity check (fail-fast with logging)
        _log.info(
            "widgets_registry connecting to Mongo at %s (db=%s, coll=%s)",
            _scrub_mongo_uri(MONGO_URI),
            MONGO_DB,
            WIDGET_COLLECTION,
        )
        try:
            await asyncio.wait_for(_client.admin.command("ping"), timeout=5)
        except Exception:
            _log.exception("Mongo connectivity check failed")
            raise

        # indices
        await _coll.create_index("origin", unique=True)
        await _coll.create_index("updated_at")

        # cache dir
        WIDGET_CACHE_DIR.mkdir(parents=True, exist_ok=True)

        # Seed only if the collection is empty (DB state beats env seed).
        try:
            count = await _coll.count_documents({})
        except Exception:
            count = 0
        if count == 0 and WIDGET_SEED:
            for src in _parse_seed_list(WIDGET_SEED):
                try:
                    await _seed_one(src)
                except Exception:
                    continue  # best-effort

        # background prewarm (do not block app startup)
        if _PREWARM:
            asyncio.create_task(_prewarm_pinned_bundles())

        _ready_once.set()
    await _ready_once.wait()





async def shutdown() -> None:
    """Close shared HTTP session on app shutdown."""
    global _session
    s, _session = _session, None
    if s is not None:
        try:
            await s.close()
        except Exception:
            pass



def _shutdown_sync():  # pragma: no cover - defensive cleanup at process exit
    try:
        # Try to close any leftover session synchronously at interpreter exit
        s = globals().get("_session")
        globals()["_session"] = None
        if s is not None:
            try:
                loop = asyncio.get_event_loop()
                if loop.is_running():
                    # Create a private loop just to close the session
                    asyncio.run(s.close())
                else:
                    loop.run_until_complete(s.close())
            except Exception:
                try:
                    asyncio.run(s.close())
                except Exception:
                    pass
    except Exception:
        pass


atexit.register(_shutdown_sync)


def _parse_source(source: str) -> dict[str, Any]:
    return _parse_source_impl(
        source,
        ensure_pkg_allowed_scope=_ensure_pkg_allowed_scope,
        check_url_allowed=_check_url_allowed,
    )


async def _seed_one(source: str) -> None:
    spec = _parse_source(source)
    existing = await _coll.find_one({"origin": spec["origin"]})
    label = _derive_key(spec["pkg"]) if spec.get("pkg") else None
    doc = {
        **spec,
        "label": label,
        "integrity": None,
        "updated_at": _utc_now_iso(),
    }
    if existing:
        return  # unique by origin
    await _coll.insert_one(doc)


async def _fetch_bundle_bytes(d: dict, ver: str | None) -> tuple[str, bytes]:
    session = await _get_session()
    if d.get("type") == "npm":
        url, data = await _pick_iife(session, d.get("pkg") or "", ver or "")
    elif d.get("type") == "artefact":
        url, data = await _artefact_pick_iife(
            session, d.get("pkg") or "", ver or ""
        )
    else:
        url = str((d.get("origin") or "")[4:])
        data = await _get(session, url)

    # Detect npm/GitLab/Nexus tarball downloads
    if url.endswith(".tgz") or data[:2] == b"\x1f\x8b":  # gzip magic header
        try:
            with tarfile.open(fileobj=io.BytesIO(data), mode="r:gz") as tar:
                # Look for index.iife.js inside package/
                candidates = [
                    m
                    for m in tar.getmembers()
                    if m.name.endswith("dist/index.iife.js")
                ]
                if not candidates:
                    raise ValueError("index.iife.js not found in tarball")
                member = candidates[0]
                file_bytes = tar.extractfile(member).read()
                return url, file_bytes
        except Exception as e:
            raise ValueError(
                f"Failed to extract index.iife.js from tarball {url}: {e}"
            )

    # Normal case: already a JS file
    return url, data


async def _write_cache_and_update(
    doc_id: ObjectId, code: str, raw: bytes
) -> str:
    integ = _sha256(raw)
    await _coll.update_one(
        {"_id": doc_id},
        {"$set": {"integrity": integ, "updated_at": _utc_now_iso()}},
    )
    return integ


async def _prewarm_pinned_bundles() -> None:
    """
    Prefetch pinned npm widgets into caches and calculate integrity (best
    effort).
    """
    try:
        docs: list[dict] = []
        cur = _coll.find(
            {},
            {
                "_id": 1,
                "type": 1,
                "pkg": 1,
                "version": 1,
                "origin": 1,
                "updated_at": 1,
            },
        )
        async for d in cur:
            if d.get("type") in {"npm", "artefact"} and d.get("version"):
                docs.append(d)

        sem = asyncio.Semaphore(_PREWARM_CONC)

        async def _one(d: dict) -> None:
            async with sem:
                doc_id = str(d["_id"])
                ver = d.get("version")
                key = _mem_key(doc_id, ver, d.get("updated_at"))
                if _mem_get(key) is not None:
                    return
                cache_name = f"{doc_id}@{ver or 'latest'}.js"
                cache_path = WIDGET_CACHE_DIR / cache_name
                if cache_path.exists():
                    try:
                        code = cache_path.read_text(encoding="utf-8")
                        _mem_put(
                            key,
                            _MemEntry(
                                version=ver or "latest",
                                code=code,
                                integrity=d.get("integrity"),
                            ),
                        )
                        return
                    except Exception:
                        pass
                try:
                    _, data = await _fetch_bundle_bytes(d, ver)
                    code = data.decode("utf-8", errors="replace")
                    cache_path.write_text(code, encoding="utf-8")
                    integ = _sha256(data)
                    _mem_put(
                        key,
                        _MemEntry(
                            version=ver or "latest", code=code, integrity=integ
                        ),
                    )
                    await _coll.update_one(
                        {"_id": d["_id"]},
                        {
                            "$set": {
                                "integrity": integ,
                                "updated_at": _utc_now_iso(),
                            }
                        },
                    )
                except Exception:
                    pass  # best-effort

        await asyncio.gather(*[asyncio.create_task(_one(d)) for d in docs])
    except Exception:
        pass


# ───────────── public API ─────────────


async def list_widgets() -> (
    list[dict[str, Any]]
):  # pragma: no cover - DB iteration
    cur = _coll.find({}).sort("label", 1)
    items: list[dict[str, Any]] = []
    async for d in cur:
        items.append(
            WidgetDoc(
                _id=d["_id"],
                type=d.get("type", "npm"),
                origin=d.get("origin", ""),
                pkg=d.get("pkg"),
                version=d.get("version"),
                label=d.get("label"),
                integrity=d.get("integrity"),
                updated_at=d.get("updated_at"),
            ).to_api()
        )
    return items





async def get_widget_bundle(
    widget_id: str, version: str | None
) -> dict[str, Any]:  # pragma: no cover - DB + filesystem I/O
    d = await _coll.find_one({"_id": ObjectId(widget_id)})
    if not d:
        raise ValueError("Widget not found")
    doc = WidgetDoc(
        _id=d["_id"],
        type=d.get("type", "npm"),
        origin=d.get("origin", ""),
        pkg=d.get("pkg"),
        version=d.get("version"),
        label=d.get("label"),
        integrity=d.get("integrity"),
        updated_at=d.get("updated_at"),
    )

    ver = version or doc.version
    if doc.type in {"npm", "artefact"} and not ver:
        raise ValueError("Widget version is not set")

    # 1) In-memory cache
    key = _mem_key(doc.id, ver, doc.updated_at)
    entry = _mem_get(key)
    if entry is not None:
        _mem_put(key, entry)
        return {
            "id": doc.id,
            "version": entry.version,
            "code": entry.code,
            "contentType": "text/javascript",
            "integrity": entry.integrity,
        }

    # 2) File cache
    cache_name = f"{doc.id}@{ver or 'latest'}.js"
    cache_path = WIDGET_CACHE_DIR / cache_name
    if cache_path.exists():
        code = cache_path.read_text(encoding="utf-8")
        entry = _MemEntry(
            version=ver or "latest", code=code, integrity=doc.integrity
        )
        _mem_put(key, entry)
        return {
            "id": doc.id,
            "version": entry.version,
            "code": entry.code,
            "contentType": "text/javascript",
            "integrity": entry.integrity,
        }

    # 3) Cold fetch (coalesced if concurrent)
    async def _do_fetch() -> tuple[str, str | None]:
        _url, data = await _fetch_bundle_bytes(d, ver)
        code_local = data.decode("utf-8", errors="replace")
        cache_path.write_text(code_local, encoding="utf-8")
        integ = await _write_cache_and_update(d["_id"], code_local, data)
        _mem_put(
            key,
            _MemEntry(
                version=ver or "latest", code=code_local, integrity=integ
            ),
        )
        return code_local, integ

    inflight_key = f"{doc.id}:{ver or 'latest'}"
    task = _INFLIGHT.get(inflight_key)
    if task is None:
        task = asyncio.create_task(_do_fetch())
        _INFLIGHT[inflight_key] = task
        try:
            code, integ = await task
        finally:
            _INFLIGHT.pop(inflight_key, None)
    else:
        code, integ = await task

    return {
        "id": doc.id,
        "version": ver or "latest",
        "code": code,
        "contentType": "text/javascript",
        "integrity": integ,
    }





async def widget_latest(
    ids: list[str],
) -> list[dict[str, Any]]:  # pragma: no cover - network/API calls
    out: list[dict[str, Any]] = []
    session = await _get_session()
    for sid in ids:
        d = await _coll.find_one({"_id": ObjectId(sid)})
        if not d:
            continue
        cur = str(d.get("version") or "")
        latest = None
        if d.get("pkg"):
            if d.get("type") == "npm":
                latest = await _npm_latest(session, d["pkg"])
            elif d.get("type") == "artefact":
                latest = await _artefact_latest(session, d["pkg"])
        out.append(
            {
                "id": str(d["_id"]),
                "current": cur or None,
                "latest": latest or None,
                "newer": _is_newer(latest, cur),
            }
        )
    return out





async def add_widget(  # pylint: disable=too-many-locals
    source: str,
) -> dict[str, Any]:  # pragma: no cover - DB + network I/O
    spec = _parse_source(source)

    # Special case: GitLab tarball download → peek headers for filename
    if (
        spec["type"] == "url"
        and "gitlab.com" in spec["origin"]
        and "/package_files/" in spec["origin"]
    ):
        url = spec["origin"][4:]  # strip "url:"
        session = await _get_session()
        async with session.head(url, allow_redirects=True) as resp:
            cd = resp.headers.get("Content-Disposition", "")
            m = re.search(r'filename="?([^"]+)"?', cd)
            if m:
                filename = urllib.parse.unquote(
                    m.group(1)
                )  # <--- decode %40 and %2F
                base = Path(filename).stem
                if "-" in base:
                    raw_name, ver = base.rsplit("-", 1)
                    if raw_name.startswith("@") and "_" in raw_name:
                        scope, name = raw_name.split("_", 1)
                        pkg = f"{scope}/{name}"
                    else:
                        pkg = raw_name
                    spec["pkg"] = pkg
                    spec["version"] = ver

    if spec["type"] in {"npm", "artefact"}:
        session = await _get_session()
        pkg = str(spec.get("pkg") or "")
        requested = str(spec.get("version") or "")
        resolved = await _resolve_source_version(
            session, spec["type"], pkg, requested
        )
        spec["version"] = resolved
        if spec["type"] == "npm":
            spec["origin"] = f"npm:{pkg}@{resolved}"
        else:
            spec["origin"] = f"artefact:{pkg}@{resolved}"

    # Always derive a clean label from pkg if present
    label = _derive_key(spec["pkg"]) if spec.get("pkg") else None
    doc = {
        **spec,
        "label": label,
        "integrity": None,
        "updated_at": _utc_now_iso(),
    }

    existing = await _coll.find_one({"origin": spec["origin"]})
    if existing:
        await _coll.update_one({"_id": existing["_id"]}, {"$set": doc})
        d = await _coll.find_one({"_id": existing["_id"]})
        _invalidate_mem_for(str(d["_id"]))
        return WidgetDoc(
            _id=d["_id"],
            type=d.get("type", "npm"),
            origin=d.get("origin", ""),
            pkg=d.get("pkg"),
            version=d.get("version"),
            label=d.get("label"),
            integrity=d.get("integrity"),
            updated_at=d.get("updated_at"),
        ).to_api()

    res = await _coll.insert_one(doc)
    doc["_id"] = res.inserted_id
    _invalidate_mem_for(str(doc["_id"]))
    return WidgetDoc(
        _id=doc["_id"],
        type=doc.get("type", "npm"),
        origin=doc.get("origin", ""),
        pkg=doc.get("pkg"),
        version=doc.get("version"),
        label=doc.get("label"),
        integrity=doc.get("integrity"),
        updated_at=doc.get("updated_at"),
    ).to_api()





async def remove_widget(widget_id: str) -> bool:  # pragma: no cover - DB
    _invalidate_mem_for(widget_id)
    res = await _coll.delete_one({"_id": ObjectId(widget_id)})
    return res.deleted_count > 0





async def upgrade_widget(
    widget_id: str, to_version: str | None
) -> dict[str, Any]:  # pragma: no cover - DB + network I/O
    d = await _coll.find_one({"_id": ObjectId(widget_id)})
    if not d:
        raise ValueError("Widget not found")

    if d.get("type") not in {"npm", "artefact"} or not d.get("pkg"):
        raise ValueError("Upgrade supported only for npm or artefact widgets")

    session = await _get_session()

    new_ver = to_version
    if new_ver:
        new_ver = await _resolve_source_version(
            session, d["type"], d["pkg"], new_ver
        )
    else:
        if d.get("type") == "npm":
            new_ver = await _npm_latest(session, d["pkg"])
        else:
            new_ver = await _artefact_latest(session, d["pkg"])
    if not new_ver:
        raise ValueError("Could not resolve latest version")

    await _coll.update_one(
        {"_id": d["_id"]},
        {
            "$set": {
                "version": new_ver,
                "updated_at": _utc_now_iso(),
                "integrity": None,
            }
        },
    )
    _invalidate_mem_for(widget_id)

    d2 = await _coll.find_one({"_id": d["_id"]})
    return WidgetDoc(
        _id=d2["_id"],
        type=d2.get("type", "npm"),
        origin=d2.get("origin", ""),
        pkg=d2.get("pkg"),
        version=d2.get("version"),
        label=d2.get("label"),
        integrity=d2.get("integrity"),
        updated_at=d2.get("updated_at"),
    ).to_api()





async def list_sources() -> (
    dict[str, Any]
):  # pragma: no cover - env formatting
    """Expose available upstream hosts and allowed npm scopes."""

    return build_sources_payload(
        WIDGET_ALLOWED_HOSTS,
        WIDGET_ALLOWED_NPM_SCOPES,
        ARTEFACT_REGISTRY_BASE,
    )