feat: pull wrapper fingerprint

__init__.py

@@ -23,6 +23,13 @@
     }
 ]
 
+tpos_redirect_paths = [
+    {
+        "from_path": "/.well-known/assetlinks.json",
+        "redirect_to_path": "/api/v1/well-known/assetlinks.json",
+    }
+]
+
 scheduled_tasks: list[asyncio.Task] = []
 
 
@@ -47,6 +54,7 @@ def tpos_start():
 __all__ = [
     "db",
     "tpos_ext",
+    "tpos_redirect_paths",
     "tpos_start",
     "tpos_static_files",
     "tpos_stop",

crud.py

@@ -1,3 +1,6 @@
+import json
+from typing import Any
+
 from lnbits.db import Database
 from lnbits.helpers import urlsafe_short_hash
 
@@ -6,6 +9,54 @@
 
 db = Database("ext_tpos")
 
+WRAPPER_ASSETLINKS_CACHE_KEY = "wrapper_assetlinks"
+
+
+async def get_wrapper_assetlinks_cache() -> tuple[dict | list, int] | None:
+    row: Any = await db.fetchone(
+        "SELECT value, updated_at FROM tpos.cache WHERE key = :key",
+        {"key": WRAPPER_ASSETLINKS_CACHE_KEY},
+    )
+    if not row or not row.get("value"):
+        return None
+    return json.loads(row["value"]), int(row["updated_at"] or 0)
+
+
+async def set_wrapper_assetlinks_cache(
+    assetlinks: dict | list, updated_at: int
+) -> None:
+    payload = json.dumps(assetlinks)
+    existing: Any = await db.fetchone(
+        "SELECT key FROM tpos.cache WHERE key = :key",
+        {"key": WRAPPER_ASSETLINKS_CACHE_KEY},
+    )
+    if existing:
+        await db.execute(
+            """
+            UPDATE tpos.cache
+            SET value = :value, updated_at = :updated_at
+            WHERE key = :key
+            """,
+            {
+                "key": WRAPPER_ASSETLINKS_CACHE_KEY,
+                "value": payload,
+                "updated_at": updated_at,
+            },
+        )
+        return
+
+    await db.execute(
+        """
+        INSERT INTO tpos.cache (key, value, updated_at)
+        VALUES (:key, :value, :updated_at)
+        """,
+        {
+            "key": WRAPPER_ASSETLINKS_CACHE_KEY,
+            "value": payload,
+            "updated_at": updated_at,
+        },
+    )
+
 
 async def create_tpos(data: CreateTposData) -> Tpos:
     tpos_id = urlsafe_short_hash()

migrations.py

@@ -299,3 +299,16 @@ async def m023_add_allow_price_adjustment(db: Database):
     await db.execute("""
         ALTER TABLE tpos.pos ADD allow_price_adjustment BOOLEAN DEFAULT true;
     """)
+
+
+async def m024_add_assetlinks_cache(db: Database):
+    """
+    Add cache table for TPoS wrapper Android asset links.
+    """
+    await db.execute("""
+        CREATE TABLE tpos.cache (
+            key TEXT PRIMARY KEY,
+            value TEXT NOT NULL,
+            updated_at INTEGER NOT NULL DEFAULT 0
+        );
+    """)

services.py

@@ -1,3 +1,4 @@
+import time
 from typing import Any
 
 import httpx
@@ -11,8 +12,47 @@
 from lnbits.settings import settings
 from loguru import logger
 
+from .crud import get_wrapper_assetlinks_cache, set_wrapper_assetlinks_cache
 from .helpers import from_csv, inventory_tags_to_list
 
+WRAPPER_ASSETLINKS_URL = (
+    "https://github.com/lnbits/TPoS-Stripe-Tap-to-Pay-Wrapper-Stripev5"
+    "/releases/latest/download/assetlinks.json"
+)
+WRAPPER_ASSETLINKS_CACHE_SECONDS = 60 * 60
+
+
+async def fetch_wrapper_assetlinks() -> dict | list:
+    now = int(time.time())
+    cached = await get_wrapper_assetlinks_cache()
+    if cached:
+        cached_assetlinks, cached_at = cached
+        cache_fresh = now - cached_at < WRAPPER_ASSETLINKS_CACHE_SECONDS
+        if cache_fresh:
+            return cached_assetlinks
+
+    try:
+        async with httpx.AsyncClient(
+            follow_redirects=True, headers={"User-Agent": settings.user_agent}
+        ) as client:
+            resp = await client.get(WRAPPER_ASSETLINKS_URL, timeout=10)
+            resp.raise_for_status()
+            assetlinks = resp.json()
+    except Exception as exc:
+        if cached:
+            logger.warning(f"Using cached TPoS wrapper assetlinks.json: {exc!s}")
+            return cached[0]
+        raise RuntimeError("Unable to fetch TPoS wrapper assetlinks.json.") from exc
+
+    if not isinstance(assetlinks, (dict, list)):
+        if cached:
+            logger.warning("Using cached TPoS wrapper assetlinks.json: invalid JSON")
+            return cached[0]
+        raise RuntimeError("TPoS wrapper assetlinks.json is not valid JSON.")
+
+    await set_wrapper_assetlinks_cache(assetlinks, now)
+    return assetlinks
+
 
 async def deduct_inventory_stock(wallet_id: str, inventory_payload: dict) -> None:
     wallet = await get_wallet(wallet_id)

static/js/index.js

@@ -777,6 +777,22 @@ window.app = Vue.createApp({
         tpos.loadingWrapperToken = false
       }
     },
+    async warmWrapperAssetlinks(enabled) {
+      if (!enabled) {
+        return
+      }
+      try {
+        await LNbits.api.request(
+          'GET',
+          '/tpos/api/v1/well-known/assetlinks.json'
+        )
+      } catch (error) {
+        Quasar.Notify.create({
+          type: 'warning',
+          message: 'Unable to cache TPoS wrapper assetlinks.json.'
+        })
+      }
+    },
     openUrlDialog(id) {
       if (this.tposs.stripe_card_payments) {
         this.urlDialog.data = _.findWhere(this.tposs, {

templates/tpos/index.html

@@ -832,6 +832,7 @@ <h6 class="text-subtitle1 q-my-none">{{SITE_TITLE}} TPoS extension</h6>
           <q-toggle
             v-model="urlDialog.data.useWrapper"
             label="Use with TPoS-Wrapper Android app"
+            @update:model-value="warmWrapperAssetlinks($event)"
           ></q-toggle>
         </div>
       </div>

tests/test_init.py

@@ -1,11 +1,18 @@
 import pytest
 from fastapi import APIRouter
 
-from .. import tpos_ext
+from .. import tpos_ext, tpos_redirect_paths
 
 
 # just import router and add it to a test router
 @pytest.mark.asyncio
 async def test_router():
     router = APIRouter()
     router.include_router(tpos_ext)
+
+
+def test_assetlinks_redirect_path():
+    assert {
+        "from_path": "/.well-known/assetlinks.json",
+        "redirect_to_path": "/api/v1/well-known/assetlinks.json",
+    } in tpos_redirect_paths

views_api.py

@@ -7,6 +7,7 @@
 
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi.responses import JSONResponse
 from lnbits.core.crud import (
     get_account,
     get_standalone_payment,
@@ -75,6 +76,7 @@
     fetch_watchonly_config,
     fetch_watchonly_wallet,
     fetch_watchonly_wallets,
+    fetch_wrapper_assetlinks,
     get_default_inventory,
     get_inventory_items_for_tpos,
     inventory_available_for_user,
@@ -94,6 +96,18 @@ def _two_year_token_expiry_minutes() -> int:
     return max(1, int((expires_at - now).total_seconds() // 60))
 
 
+@tpos_api_router.get("/api/v1/well-known/assetlinks.json")
+async def api_tpos_assetlinks() -> JSONResponse:
+    try:
+        assetlinks = await fetch_wrapper_assetlinks()
+    except RuntimeError as exc:
+        raise HTTPException(
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE,
+            detail=str(exc),
+        ) from exc
+    return JSONResponse(content=assetlinks, media_type="application/json")
+
+
 def _build_receipt_data(
     tpos: Tpos, payment: Payment, tpos_payment: TposPayment | None = None
 ) -> ReceiptData: