feat: dumb

lnbits/core/extensions/__init__.py

@@ -0,0 +1 @@
+"""Extension runtime contracts."""

lnbits/core/extensions/capabilities/__init__.py

@@ -0,0 +1,8 @@
+from .catalog import get_extension_capability_registry
+from .registry import CapabilityDefinition, CapabilityRegistry
+
+__all__ = [
+    "CapabilityDefinition",
+    "CapabilityRegistry",
+    "get_extension_capability_registry",
+]

lnbits/core/extensions/capabilities/catalog.py

@@ -0,0 +1,120 @@
+from .models import (
+    CreateInvoiceRequest,
+    CreateInvoiceResponse,
+    EmptyRequest,
+    KvGetRequest,
+    KvGetResponse,
+    KvListRequest,
+    KvListResponse,
+    KvSetRequest,
+    KvSetResponse,
+    LogRequest,
+    LogResponse,
+    NowResponse,
+    RandomIdRequest,
+    RandomIdResponse,
+    WatchPaymentRequest,
+    WatchPaymentResponse,
+)
+from .registry import CapabilityDefinition, CapabilityRegistry
+
+
+def get_extension_capability_registry() -> CapabilityRegistry:
+    registry = CapabilityRegistry()
+
+    for capability in [
+        CapabilityDefinition(
+            id="storage.get",
+            namespace="storage",
+            name="Get storage value",
+            host_name="kv_get",
+            sdk_name="get",
+            description="Read one value from the extension storage namespace.",
+            request_model=KvGetRequest,
+            response_model=KvGetResponse,
+            required_permission="ext.storage.read_write",
+            public_context=True,
+        ),
+        CapabilityDefinition(
+            id="storage.set",
+            namespace="storage",
+            name="Set storage value",
+            host_name="kv_set",
+            sdk_name="set",
+            description="Write one value to the extension storage namespace.",
+            request_model=KvSetRequest,
+            response_model=KvSetResponse,
+            required_permission="ext.storage.read_write",
+        ),
+        CapabilityDefinition(
+            id="storage.list",
+            namespace="storage",
+            name="List storage keys",
+            host_name="kv_list",
+            sdk_name="list",
+            description="List keys under a prefix in the extension storage namespace.",
+            request_model=KvListRequest,
+            response_model=KvListResponse,
+            required_permission="ext.storage.read_write",
+            public_context=True,
+        ),
+        CapabilityDefinition(
+            id="wallet.create_invoice",
+            namespace="wallet",
+            name="Create invoice",
+            host_name="create_invoice",
+            sdk_name="createInvoice",
+            description="Create an incoming Lightning invoice for an allowed wallet.",
+            request_model=CreateInvoiceRequest,
+            response_model=CreateInvoiceResponse,
+            required_permission="wallet.create_invoice",
+            public_context=True,
+        ),
+        CapabilityDefinition(
+            id="payments.watch",
+            namespace="payments",
+            name="Watch payment",
+            host_name="watch_payment",
+            sdk_name="watch",
+            description="Subscribe the extension to a payment state callback.",
+            request_model=WatchPaymentRequest,
+            response_model=WatchPaymentResponse,
+            required_permission="payments.watch",
+        ),
+        CapabilityDefinition(
+            id="system.random_id",
+            namespace="system",
+            name="Random ID",
+            host_name="random_id",
+            sdk_name="id",
+            description="Create a random extension-local identifier.",
+            request_model=RandomIdRequest,
+            response_model=RandomIdResponse,
+            public_context=True,
+        ),
+        CapabilityDefinition(
+            id="system.now",
+            namespace="system",
+            name="Current timestamp",
+            host_name="now",
+            sdk_name="now",
+            description="Return the current Unix timestamp.",
+            request_model=EmptyRequest,
+            response_model=NowResponse,
+            public_context=True,
+        ),
+        CapabilityDefinition(
+            id="system.log",
+            namespace="system",
+            name="Log message",
+            host_name="log",
+            sdk_name="log",
+            description="Write a bounded message to the extension log.",
+            request_model=LogRequest,
+            response_model=LogResponse,
+            public_context=True,
+        ),
+    ]:
+        registry.register(capability)
+
+    return registry

lnbits/core/extensions/capabilities/models.py

@@ -0,0 +1,76 @@
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class EmptyRequest(BaseModel):
+    pass
+
+
+class KvGetRequest(BaseModel):
+    key: str = Field(..., min_length=1, max_length=512)
+
+
+class KvGetResponse(BaseModel):
+    value: str | None = None
+
+
+class KvSetRequest(BaseModel):
+    key: str = Field(..., min_length=1, max_length=512)
+    value: str = Field(..., max_length=65536)
+
+
+class KvSetResponse(BaseModel):
+    ok: bool = True
+
+
+class KvListRequest(BaseModel):
+    prefix: str = Field(..., min_length=1, max_length=512)
+
+
+class KvListResponse(BaseModel):
+    keys: list[str] = []
+
+
+class CreateInvoiceRequest(BaseModel):
+    wallet_id: str = Field(..., min_length=1, max_length=128)
+    amount_sat: int = Field(..., gt=0)
+    memo: str = Field(..., max_length=512)
+    tag: str = Field(..., min_length=1, max_length=64)
+    extra: dict[str, str] = Field(default_factory=dict)
+
+
+class CreateInvoiceResponse(BaseModel):
+    payment_hash: str
+    payment_request: str
+    checking_id: str
+
+
+class WatchPaymentRequest(BaseModel):
+    payment_hash: str = Field(..., min_length=1, max_length=128)
+    callback_export: str = Field(..., min_length=1, max_length=128)
+
+
+class WatchPaymentResponse(BaseModel):
+    ok: bool = True
+
+
+class RandomIdRequest(BaseModel):
+    prefix: str = Field(..., min_length=1, max_length=32)
+
+
+class RandomIdResponse(BaseModel):
+    id: str
+
+
+class NowResponse(BaseModel):
+    timestamp: int
+
+
+class LogRequest(BaseModel):
+    level: Literal["debug", "info", "warning", "error"] = "info"
+    message: str = Field(..., min_length=1, max_length=2048)
+
+
+class LogResponse(BaseModel):
+    ok: bool = True

lnbits/core/extensions/capabilities/registry.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from pydantic import BaseModel
+
+
+@dataclass(frozen=True)
+class CapabilityDefinition:
+    id: str
+    namespace: str
+    name: str
+    host_name: str
+    sdk_name: str
+    description: str
+    request_model: type[BaseModel]
+    response_model: type[BaseModel]
+    required_permission: str | None = None
+    public_context: bool = False
+
+    @property
+    def sdk_qualified_name(self) -> str:
+        return f"{self.namespace}.{self.sdk_name}"
+
+    def request_schema(self) -> dict[str, Any]:
+        return self.request_model.schema(ref_template="#/definitions/{model}")
+
+    def response_schema(self) -> dict[str, Any]:
+        return self.response_model.schema(ref_template="#/definitions/{model}")
+
+    def to_codegen_contract(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "namespace": self.namespace,
+            "name": self.name,
+            "host_name": self.host_name,
+            "sdk_name": self.sdk_name,
+            "sdk_qualified_name": self.sdk_qualified_name,
+            "description": self.description,
+            "required_permission": self.required_permission,
+            "public_context": self.public_context,
+            "request_schema": self.request_schema(),
+            "response_schema": self.response_schema(),
+        }
+
+
+class CapabilityRegistry:
+    def __init__(self) -> None:
+        self._capabilities: dict[str, CapabilityDefinition] = {}
+
+    def register(self, capability: CapabilityDefinition) -> None:
+        if capability.id in self._capabilities:
+            raise ValueError(f"Capability '{capability.id}' already registered.")
+        self._capabilities[capability.id] = capability
+
+    def get(self, capability_id: str) -> CapabilityDefinition | None:
+        return self._capabilities.get(capability_id)
+
+    def require(self, capability_id: str) -> CapabilityDefinition:
+        capability = self.get(capability_id)
+        if not capability:
+            raise KeyError(f"Unknown capability '{capability_id}'.")
+        return capability
+
+    def all(self) -> list[CapabilityDefinition]:
+        return sorted(self._capabilities.values(), key=lambda c: c.id)
+
+    def by_permission(self, permission_id: str) -> list[CapabilityDefinition]:
+        return [c for c in self.all() if c.required_permission == permission_id]
+
+    def to_codegen_contract(self) -> dict[str, Any]:
+        return {
+            "version": 1,
+            "capabilities": [c.to_codegen_contract() for c in self.all()],
+        }