fix(payments): record-payment fails closed and shares the claim gate

Two fixes to POST /api/v1/record-payment:

- The Fava duplicate check caught every exception and proceeded to
  write, so a transient Fava blip produced double entries. It now
  fails closed: transport errors return 503 and the client retries.
  While here: the check queried {base_url}/api/journal, but base_url
  already ends in /api — the doubled path 404'd, meaning the
  duplicate check has silently never run.

- The endpoint now goes through the same processed_payments claim
  gate as the background invoice listener, so the webhook+poller pair
  can't both record the same payment_hash: a 'done' claim replays as
  "already recorded", an in-flight claim returns 409.

Addresses CODE-REVIEW-2026-06 finding #10.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

crud.py

@@ -1696,3 +1696,73 @@ async def check_user_has_role_permission(
                 return True
 
     return False
+
+
+# =============================================================================
+# PROCESSED PAYMENTS (Lightning payment idempotency gate)
+# =============================================================================
+# The Fava-side duplicate checks are read-then-write races; this table's
+# primary key on payment_hash makes exactly one claimant win. Shared by the
+# background invoice listener (tasks.on_invoice_paid) and the client-driven
+# /record-payment endpoint.
+
+
+async def claim_payment(payment_hash: str) -> bool:
+    """Atomically claim a Lightning payment for recording.
+
+    Returns True when this caller owns the claim; False when the payment
+    is already recorded or another coroutine is recording it right now.
+    """
+    result = await db.execute(
+        """
+        INSERT INTO processed_payments (payment_hash, status)
+        VALUES (:payment_hash, 'processing')
+        ON CONFLICT (payment_hash) DO NOTHING
+        """,
+        {"payment_hash": payment_hash},
+    )
+    return result.rowcount == 1
+
+
+async def get_processed_payment(payment_hash: str) -> Optional[dict]:
+    row = await db.fetchone(
+        "SELECT payment_hash, status, entry_id FROM processed_payments"
+        " WHERE payment_hash = :payment_hash",
+        {"payment_hash": payment_hash},
+    )
+    return dict(row) if row else None
+
+
+async def mark_payment_done(payment_hash: str, entry_id: Optional[str] = None) -> None:
+    await db.execute(
+        """
+        UPDATE processed_payments SET status = 'done', entry_id = :entry_id
+        WHERE payment_hash = :payment_hash
+        """,
+        {"payment_hash": payment_hash, "entry_id": entry_id},
+    )
+
+
+async def release_payment_claim(payment_hash: str) -> None:
+    """Compensating delete after a failed recording, so redelivery retries.
+
+    Only removes an in-flight claim — a 'done' row is permanent.
+    """
+    await db.execute(
+        "DELETE FROM processed_payments"
+        " WHERE payment_hash = :payment_hash AND status = 'processing'",
+        {"payment_hash": payment_hash},
+    )
+
+
+async def clear_stale_payment_claims() -> int:
+    """Drop 'processing' claims left behind by a previous process life.
+
+    A live claim only exists inside a running coroutine, so anything
+    still 'processing' at listener startup belongs to a crashed or
+    restarted process and would otherwise block that payment forever.
+    """
+    result = await db.execute(
+        "DELETE FROM processed_payments WHERE status = 'processing'"
+    )
+    return result.rowcount

migrations.py

@@ -34,9 +34,33 @@ Original migration sequence (Nov 2025):
 - m014: Removed legacy equity accounts (MemberEquity, RetainedEarnings)
 - m015: Converted entry_lines to single amount field
 - m016: Dropped journal_entries and entry_lines tables (Fava integration)
+
+IDEMPOTENCY CONTRACT:
+Every statement here must be a silent no-op on re-run. The migration
+version bump lands in the core LNbits DB (`dbversions`) while the DDL
+lands in `ext_libra` — the two writes are not atomic. If the version
+bump fails after the DDL commits, the whole migration re-runs on next
+boot; a bare CREATE/ALTER/INSERT then crashes the extension until
+manual dbversions surgery.
 """
 
 
+async def _alter_add_column_safe(db, sql: str) -> None:
+    """ALTER TABLE ADD COLUMN that swallows duplicate-column errors.
+
+    Neither SQLite nor Postgres supports ADD COLUMN IF NOT EXISTS
+    portably, so re-runs are made no-ops by swallowing the error both
+    backends raise for an existing column.
+    """
+    try:
+        await db.execute(sql)
+    except Exception as exc:
+        msg = str(exc).lower()
+        if "duplicate column" in msg or "already exists" in msg:
+            return
+        raise
+
+
 async def m001_initial(db):
     """
     Initial Libra database schema (squashed from m001-m016).
@@ -63,7 +87,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE accounts (
+        CREATE TABLE IF NOT EXISTS accounts (
             id TEXT PRIMARY KEY,
             name TEXT NOT NULL UNIQUE,
             account_type TEXT NOT NULL,
@@ -76,13 +100,13 @@ async def m001_initial(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_accounts_user_id ON accounts (user_id);
+        CREATE INDEX IF NOT EXISTS idx_accounts_user_id ON accounts (user_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_accounts_type ON accounts (account_type);
+        CREATE INDEX IF NOT EXISTS idx_accounts_type ON accounts (account_type);
         """
     )
 
@@ -93,7 +117,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE extension_settings (
+        CREATE TABLE IF NOT EXISTS extension_settings (
             id TEXT NOT NULL PRIMARY KEY,
             libra_wallet_id TEXT,
             fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333',
@@ -111,7 +135,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE user_wallet_settings (
+        CREATE TABLE IF NOT EXISTS user_wallet_settings (
             id TEXT NOT NULL PRIMARY KEY,
             user_wallet_id TEXT,
             updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
@@ -126,7 +150,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE manual_payment_requests (
+        CREATE TABLE IF NOT EXISTS manual_payment_requests (
             id TEXT PRIMARY KEY,
             user_id TEXT NOT NULL,
             amount INTEGER NOT NULL,
@@ -143,14 +167,14 @@ async def m001_initial(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_manual_payment_requests_user_id
+        CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_user_id
         ON manual_payment_requests (user_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_manual_payment_requests_status
+        CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_status
         ON manual_payment_requests (status);
         """
     )
@@ -163,7 +187,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE balance_assertions (
+        CREATE TABLE IF NOT EXISTS balance_assertions (
             id TEXT PRIMARY KEY,
             date TIMESTAMP NOT NULL,
             account_id TEXT NOT NULL,
@@ -188,21 +212,21 @@ async def m001_initial(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_balance_assertions_account_id
+        CREATE INDEX IF NOT EXISTS idx_balance_assertions_account_id
         ON balance_assertions (account_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_balance_assertions_status
+        CREATE INDEX IF NOT EXISTS idx_balance_assertions_status
         ON balance_assertions (status);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_balance_assertions_date
+        CREATE INDEX IF NOT EXISTS idx_balance_assertions_date
         ON balance_assertions (date);
         """
     )
@@ -216,7 +240,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE user_equity_status (
+        CREATE TABLE IF NOT EXISTS user_equity_status (
             user_id TEXT PRIMARY KEY,
             is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE,
             equity_account_name TEXT,
@@ -230,7 +254,7 @@ async def m001_initial(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_user_equity_status_eligible
+        CREATE INDEX IF NOT EXISTS idx_user_equity_status_eligible
         ON user_equity_status (is_equity_eligible)
         WHERE is_equity_eligible = TRUE;
         """
@@ -245,7 +269,7 @@ async def m001_initial(db):
 
     await db.execute(
         f"""
-        CREATE TABLE account_permissions (
+        CREATE TABLE IF NOT EXISTS account_permissions (
             id TEXT PRIMARY KEY,
             user_id TEXT NOT NULL,
             account_id TEXT NOT NULL,
@@ -262,7 +286,7 @@ async def m001_initial(db):
     # Index for looking up permissions by user
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_user_id
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_user_id
         ON account_permissions (user_id);
         """
     )
@@ -270,7 +294,7 @@ async def m001_initial(db):
     # Index for looking up permissions by account
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_account_id
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_account_id
         ON account_permissions (account_id);
         """
     )
@@ -278,7 +302,7 @@ async def m001_initial(db):
     # Composite index for checking specific user+account permissions
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_user_account
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_user_account
         ON account_permissions (user_id, account_id);
         """
     )
@@ -286,7 +310,7 @@ async def m001_initial(db):
     # Index for finding permissions by type
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_type
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_type
         ON account_permissions (permission_type);
         """
     )
@@ -294,7 +318,7 @@ async def m001_initial(db):
     # Index for finding expired permissions
     await db.execute(
         """
-        CREATE INDEX idx_account_permissions_expires
+        CREATE INDEX IF NOT EXISTS idx_account_permissions_expires
         ON account_permissions (expires_at)
         WHERE expires_at IS NOT NULL;
         """
@@ -320,6 +344,7 @@ async def m001_initial(db):
             f"""
             INSERT INTO accounts (id, name, account_type, description, created_at)
             VALUES (:id, :name, :type, :description, {db.timestamp_now})
+            ON CONFLICT (name) DO NOTHING
             """,
             {
                 "id": str(uuid.uuid4()),
@@ -342,17 +367,18 @@ async def m002_add_account_is_active(db):
 
     Default: All existing accounts are marked as active (TRUE).
     """
-    await db.execute(
+    await _alter_add_column_safe(
+        db,
         """
         ALTER TABLE accounts
         ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT TRUE
-        """
+        """,
     )
 
     # Create index for faster queries filtering by is_active
     await db.execute(
         """
-        CREATE INDEX idx_accounts_is_active ON accounts (is_active)
+        CREATE INDEX IF NOT EXISTS idx_accounts_is_active ON accounts (is_active)
         """
     )
 
@@ -372,17 +398,18 @@ async def m003_add_account_is_virtual(db):
 
     Default: All existing accounts are real (is_virtual = FALSE).
     """
-    await db.execute(
+    await _alter_add_column_safe(
+        db,
         """
         ALTER TABLE accounts
         ADD COLUMN is_virtual BOOLEAN NOT NULL DEFAULT FALSE
-        """
+        """,
     )
 
     # Create index for faster queries filtering by is_virtual
     await db.execute(
         """
-        CREATE INDEX idx_accounts_is_virtual ON accounts (is_virtual)
+        CREATE INDEX IF NOT EXISTS idx_accounts_is_virtual ON accounts (is_virtual)
         """
     )
 
@@ -402,6 +429,7 @@ async def m003_add_account_is_virtual(db):
             f"""
             INSERT INTO accounts (id, name, account_type, description, is_active, is_virtual, created_at)
             VALUES (:id, :name, :type, :description, TRUE, TRUE, {db.timestamp_now})
+            ON CONFLICT (name) DO NOTHING
             """,
             {
                 "id": str(uuid.uuid4()),
@@ -438,7 +466,7 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         f"""
-        CREATE TABLE roles (
+        CREATE TABLE IF NOT EXISTS roles (
             id TEXT PRIMARY KEY,
             name TEXT NOT NULL UNIQUE,
             description TEXT,
@@ -451,13 +479,13 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_roles_name ON roles (name);
+        CREATE INDEX IF NOT EXISTS idx_roles_name ON roles (name);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_roles_is_default ON roles (is_default)
+        CREATE INDEX IF NOT EXISTS idx_roles_is_default ON roles (is_default)
         WHERE is_default = TRUE;
         """
     )
@@ -469,7 +497,7 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         f"""
-        CREATE TABLE role_permissions (
+        CREATE TABLE IF NOT EXISTS role_permissions (
             id TEXT PRIMARY KEY,
             role_id TEXT NOT NULL,
             account_id TEXT NOT NULL,
@@ -484,19 +512,19 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_role_permissions_role_id ON role_permissions (role_id);
+        CREATE INDEX IF NOT EXISTS idx_role_permissions_role_id ON role_permissions (role_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_role_permissions_account_id ON role_permissions (account_id);
+        CREATE INDEX IF NOT EXISTS idx_role_permissions_account_id ON role_permissions (account_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_role_permissions_type ON role_permissions (permission_type);
+        CREATE INDEX IF NOT EXISTS idx_role_permissions_type ON role_permissions (permission_type);
         """
     )
 
@@ -507,7 +535,7 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         f"""
-        CREATE TABLE user_roles (
+        CREATE TABLE IF NOT EXISTS user_roles (
             id TEXT PRIMARY KEY,
             user_id TEXT NOT NULL,
             role_id TEXT NOT NULL,
@@ -522,19 +550,19 @@ async def m004_add_rbac_tables(db):
 
     await db.execute(
         """
-        CREATE INDEX idx_user_roles_user_id ON user_roles (user_id);
+        CREATE INDEX IF NOT EXISTS idx_user_roles_user_id ON user_roles (user_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_user_roles_role_id ON user_roles (role_id);
+        CREATE INDEX IF NOT EXISTS idx_user_roles_role_id ON user_roles (role_id);
         """
     )
 
     await db.execute(
         """
-        CREATE INDEX idx_user_roles_expires ON user_roles (expires_at)
+        CREATE INDEX IF NOT EXISTS idx_user_roles_expires ON user_roles (expires_at)
         WHERE expires_at IS NOT NULL;
         """
     )
@@ -542,7 +570,7 @@ async def m004_add_rbac_tables(db):
     # Composite index for checking specific user+role assignments
     await db.execute(
         """
-        CREATE INDEX idx_user_roles_user_role ON user_roles (user_id, role_id);
+        CREATE INDEX IF NOT EXISTS idx_user_roles_user_role ON user_roles (user_id, role_id);
         """
     )
 
@@ -586,6 +614,7 @@ async def m004_add_rbac_tables(db):
             f"""
             INSERT INTO roles (id, name, description, is_default, created_by, created_at)
             VALUES (:id, :name, :description, :is_default, :created_by, {db.timestamp_now})
+            ON CONFLICT (name) DO NOTHING
             """,
             {
                 "id": str(uuid.uuid4()),
@@ -595,3 +624,30 @@ async def m004_add_rbac_tables(db):
                 "created_by": "system",  # System-created default roles
             },
         )
+
+
+async def m005_add_processed_payments(db):
+    """
+    Local idempotency gate for Lightning payment recording.
+
+    The Fava-side duplicate check (`add_entry_idempotent`, journal-link
+    scan) is a read-then-write race: the background invoice listener and
+    the client-driven /record-payment endpoint can both pass the "not
+    present" check for the same payment_hash and both insert. The
+    primary key on payment_hash makes exactly one claimant win.
+
+    status lifecycle: 'processing' (claimed, write in flight) → 'done'
+    (entry recorded). Failed claims are deleted so redelivery retries;
+    'processing' rows from a crashed process are cleared at listener
+    startup.
+    """
+    await db.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS processed_payments (
+            payment_hash TEXT PRIMARY KEY,
+            status TEXT NOT NULL DEFAULT 'processing',
+            entry_id TEXT,
+            created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
+        );
+        """
+    )

tasks.py

@@ -179,12 +179,31 @@ async def wait_for_paid_invoices():
     This ensures payments are recorded even if the user closes their browser
     before the payment is detected by client-side polling.
     """
+    from .crud import clear_stale_payment_claims
+
     invoice_queue = Queue()
     register_invoice_listener(invoice_queue, "ext_libra")
 
+    # Claims from a previous process life can't be live anymore — clear
+    # them so those payments aren't blocked forever.
+    cleared = await clear_stale_payment_claims()
+    if cleared:
+        logger.warning(
+            f"[LIBRA] Cleared {cleared} stale in-flight payment claim(s) "
+            "from a previous run"
+        )
+
     while True:
         payment = await invoice_queue.get()
-        await on_invoice_paid(payment)
+        try:
+            await on_invoice_paid(payment)
+        except Exception:
+            # One bad payment must not kill the listener for the rest of
+            # the process lifetime; its claim was released, so redelivery
+            # can retry it.
+            logger.exception(
+                f"[LIBRA] Failed to record payment {payment.payment_hash}"
+            )
 
 
 async def on_invoice_paid(payment: Payment) -> None:
@@ -210,8 +229,20 @@ async def on_invoice_paid(payment: Payment) -> None:
         logger.warning(f"Libra invoice {payment.payment_hash} missing user_id in metadata")
         return
 
+    from .crud import claim_payment, mark_payment_done, release_payment_claim
     from .fava_client import get_fava_client
 
+    # Local idempotency gate: exactly one claimant (this listener or the
+    # /record-payment endpoint) gets to record a given payment_hash. The
+    # Fava-side idempotent write below stays as a second layer for
+    # entries recorded before this table existed.
+    if not await claim_payment(payment.payment_hash):
+        logger.info(
+            f"Payment {payment.payment_hash} already recorded or being "
+            "recorded; skipping"
+        )
+        return
+
     fava = get_fava_client()
 
     # Use idempotency key based on payment hash - this ensures duplicate
@@ -245,6 +276,7 @@ async def on_invoice_paid(payment: Payment) -> None:
 
             if not fiat_currency or not fiat_amount:
                 logger.error(f"Payment {payment.payment_hash} missing fiat currency/amount metadata")
+                await release_payment_claim(payment.payment_hash)
                 return
 
             # Get user's current balance to determine receivables and payables
@@ -276,6 +308,7 @@ async def on_invoice_paid(payment: Payment) -> None:
             lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
             if not lightning_account:
                 logger.error("Lightning account 'Assets:Bitcoin:Lightning' not found")
+                await release_payment_claim(payment.payment_hash)
                 return
 
             # Query for unsettled entries to link this settlement back to them
@@ -324,6 +357,11 @@ async def on_invoice_paid(payment: Payment) -> None:
                     f"{result.get('data', 'Unknown')}"
                 )
 
+            await mark_payment_done(payment.payment_hash, idempotency_key)
+
         except Exception as e:
             logger.error(f"Error recording Libra payment {payment.payment_hash}: {e}")
+            # Release the claim so redelivery (or the /record-payment
+            # endpoint) can retry this payment.
+            await release_payment_claim(payment.payment_hash)
             raise

tests/test_migrations.py

@@ -0,0 +1,110 @@
+"""Migration idempotency tests.
+
+The migration version bump lands in the core LNbits DB (`dbversions`)
+while the DDL lands in `ext_libra` — the two writes are not atomic. If
+the bump fails after the DDL commits, the whole migration re-runs on
+next boot, so every statement must be a silent no-op on re-run instead
+of crashing with `duplicate column` / `table exists` / UNIQUE
+violations (which bricks the extension until manual dbversions
+surgery).
+
+These tests run the full migration chain twice against a fresh SQLite
+database — the second pass simulates the lost-version-bump re-run.
+"""
+import importlib
+import re
+from uuid import uuid4
+
+import pytest
+
+from lnbits.db import Database
+
+pytestmark = pytest.mark.anyio
+
+
+def _module(name: str):
+    """Import a libra submodule under whichever path the active LNbits layout
+    uses (default `lnbits.extensions.libra` or bare `libra`)."""
+    for prefix in ("lnbits.extensions.libra", "libra"):
+        try:
+            return importlib.import_module(f"{prefix}.{name}")
+        except ModuleNotFoundError:
+            continue
+    raise ModuleNotFoundError(f"libra.{name}: tried both import paths")
+
+
+migrations = _module("migrations")
+
+# Same discovery as lnbits.core.helpers.run_migration: m### prefix,
+# module definition order.
+_MIGRATION_RE = re.compile(r"^m(\d\d\d)_")
+MIGRATION_FUNCTIONS = [
+    fn for name, fn in vars(migrations).items() if _MIGRATION_RE.match(name)
+]
+
+# Tables the chain must leave behind — one probe row read per table
+# proves both existence and queryability after a double run.
+EXPECTED_TABLES = [
+    "accounts",
+    "extension_settings",
+    "user_wallet_settings",
+    "manual_payment_requests",
+    "balance_assertions",
+    "user_equity_status",
+    "account_permissions",
+    "roles",
+    "role_permissions",
+    "user_roles",
+    "processed_payments",
+]
+
+
+async def _run_all_migrations(db: Database) -> None:
+    async with db.connect() as conn:
+        for migrate in MIGRATION_FUNCTIONS:
+            await migrate(conn)
+
+
+async def _seed_counts(db: Database) -> dict:
+    async with db.connect() as conn:
+        accounts = await conn.fetchall("SELECT id, name FROM accounts")
+        roles = await conn.fetchall("SELECT id, name FROM roles")
+    return {
+        "account_names": sorted(r["name"] for r in accounts),
+        "account_ids": sorted(r["id"] for r in accounts),
+        "role_names": sorted(r["name"] for r in roles),
+        "role_ids": sorted(r["id"] for r in roles),
+    }
+
+
+async def test_migrations_rerun_is_noop():
+    """Full chain twice: second run must not raise and not re-seed."""
+    db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}")
+
+    await _run_all_migrations(db)
+    first = await _seed_counts(db)
+
+    # Simulate the lost dbversions bump: everything runs again.
+    await _run_all_migrations(db)
+    second = await _seed_counts(db)
+
+    # Seeds must not duplicate (names) and must not be replaced (ids).
+    assert second == first
+    assert first["account_names"], "seed accounts missing after migration"
+    assert "Employee" in first["role_names"]
+
+    # Every table exists and is queryable after the double run.
+    async with db.connect() as conn:
+        for table in EXPECTED_TABLES:
+            await conn.fetchall(f"SELECT * FROM {table} LIMIT 1")  # noqa: S608
+
+
+async def test_single_migration_rerun_is_noop():
+    """Each migration individually survives an immediate re-run (the
+    version bump fails right after that one migration committed)."""
+    db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}")
+
+    async with db.connect() as conn:
+        for migrate in MIGRATION_FUNCTIONS:
+            await migrate(conn)
+            await migrate(conn)  # re-run before "bumping" to the next

tests/test_payment_idempotency.py

@@ -0,0 +1,281 @@
+"""Lightning payment idempotency — the `processed_payments` claim gate.
+
+The background invoice listener (`tasks.on_invoice_paid`) and the
+client-driven `POST /record-payment` endpoint can both fire for the
+same `payment_hash` (queue redelivery after restart, webhook + poller).
+The Fava-side duplicate checks are read-then-write races; the local
+`processed_payments` primary key makes exactly one claimant win.
+
+These tests bypass invoice generation (blocked by libra/issues/40) by
+delivering synthetic paid `Payment` objects straight to
+`on_invoice_paid` and by inserting paid payment rows via the LNbits
+core crud for the endpoint tests.
+"""
+import asyncio
+import importlib
+from uuid import uuid4
+
+import pytest
+
+from lnbits.core.crud.payments import create_payment
+from lnbits.core.models.payments import CreatePayment, Payment, PaymentState
+
+from .helpers import list_user_entries, post_receivable
+
+pytestmark = pytest.mark.anyio
+
+
+def _module(name: str):
+    """Import a libra submodule under whichever path the active LNbits layout
+    uses (default `lnbits.extensions.libra` or bare `libra`)."""
+    for prefix in ("lnbits.extensions.libra", "libra"):
+        try:
+            return importlib.import_module(f"{prefix}.{name}")
+        except ModuleNotFoundError:
+            continue
+    raise ModuleNotFoundError(f"libra.{name}: tried both import paths")
+
+
+tasks = _module("tasks")
+libra_crud = _module("crud")
+
+
+def _paid_payment(
+    wallet_id: str,
+    user_id: str,
+    *,
+    fiat_amount: str = "100.00",
+    fiat_currency: str = "EUR",
+    sats: int = 100_000,
+) -> Payment:
+    payment_hash = uuid4().hex + uuid4().hex[:32]
+    return Payment(
+        checking_id=payment_hash,
+        payment_hash=payment_hash,
+        wallet_id=wallet_id,
+        amount=sats * 1000,
+        fee=0,
+        bolt11="lnbcfake",
+        status=PaymentState.SUCCESS,
+        extra={
+            "tag": "libra",
+            "user_id": user_id,
+            "fiat_currency": fiat_currency,
+            "fiat_amount": fiat_amount,
+        },
+    )
+
+
+async def _setup_receivable(
+    client, super_user_headers, configured_user, standard_accounts,
+    amount: str = "100.00",
+):
+    user, wallet = configured_user
+    await post_receivable(
+        client,
+        super_user_headers=super_user_headers,
+        user_id=user.id,
+        amount=amount,
+        description=f"Idempotency setup {uuid4().hex[:6]}",
+        revenue_account=standard_accounts["revenue_rent"]["name"],
+    )
+    # Force a Fava reload before downstream balance reads (see #37).
+    await list_user_entries(client, wallet_inkey=wallet.inkey)
+    return user, wallet
+
+
+async def _entries_with_link(client, wallet_inkey: str, link: str) -> list:
+    payload = await list_user_entries(client, wallet_inkey=wallet_inkey)
+    return [
+        e for e in payload["entries"] if link in (e.get("links") or [])
+    ]
+
+
+# ---------------------------------------------------------------------------
+# on_invoice_paid — the background listener path
+# ---------------------------------------------------------------------------
+
+
+async def test_double_delivery_records_exactly_once(
+    client, super_user_headers, configured_user, standard_accounts
+):
+    """Same payment delivered twice (queue redelivery) → one ledger entry."""
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+    payment = _paid_payment(wallet.id, user.id)
+
+    await tasks.on_invoice_paid(payment)
+    await tasks.on_invoice_paid(payment)
+
+    link = f"ln-{payment.payment_hash[:16]}"
+    assert len(await _entries_with_link(client, wallet.inkey, link)) == 1
+
+    row = await libra_crud.get_processed_payment(payment.payment_hash)
+    assert row is not None and row["status"] == "done"
+
+
+async def test_failed_recording_releases_claim_and_retry_succeeds(
+    client, super_user_headers, configured_user, standard_accounts, monkeypatch
+):
+    """A Fava failure mid-write must not permanently block the payment."""
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+    payment = _paid_payment(wallet.id, user.id)
+
+    fava_client = _module("fava_client")
+    fava = fava_client.get_fava_client()
+
+    async def _boom(*args, **kwargs):
+        raise RuntimeError("fava down")
+
+    monkeypatch.setattr(fava, "add_entry_idempotent", _boom)
+    with pytest.raises(RuntimeError):
+        await tasks.on_invoice_paid(payment)
+    monkeypatch.undo()
+
+    # Claim released → nothing recorded, retry allowed.
+    assert await libra_crud.get_processed_payment(payment.payment_hash) is None
+
+    await tasks.on_invoice_paid(payment)
+    row = await libra_crud.get_processed_payment(payment.payment_hash)
+    assert row is not None and row["status"] == "done"
+    link = f"ln-{payment.payment_hash[:16]}"
+    assert len(await _entries_with_link(client, wallet.inkey, link)) == 1
+
+
+async def test_listener_survives_poison_payment_and_clears_stale_claims(
+    client, super_user_headers, configured_user, standard_accounts, monkeypatch
+):
+    """One bad payment must not kill the listener; stale 'processing'
+    claims from a previous process life are cleared at startup."""
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+
+    # A claim left behind by a "crashed" previous run.
+    stale_hash = uuid4().hex + uuid4().hex[:32]
+    assert await libra_crud.claim_payment(stale_hash)
+
+    captured: dict = {}
+    monkeypatch.setattr(
+        tasks,
+        "register_invoice_listener",
+        lambda queue, name: captured.update(queue=queue),
+    )
+
+    listener = asyncio.create_task(tasks.wait_for_paid_invoices())
+    try:
+        for _ in range(50):
+            if "queue" in captured:
+                break
+            await asyncio.sleep(0.05)
+        assert "queue" in captured, "listener never registered its queue"
+
+        poison = _paid_payment(wallet.id, user.id, fiat_amount="not-a-number")
+        good = _paid_payment(wallet.id, user.id)
+        captured["queue"].put_nowait(poison)
+        captured["queue"].put_nowait(good)
+
+        row = None
+        for _ in range(100):
+            row = await libra_crud.get_processed_payment(good.payment_hash)
+            if row and row["status"] == "done":
+                break
+            await asyncio.sleep(0.1)
+        assert row is not None and row["status"] == "done", (
+            "good payment was not recorded after the poison payment"
+        )
+    finally:
+        listener.cancel()
+
+    # Startup cleared the stale claim; the poison payment's claim was
+    # released on failure so redelivery could retry it.
+    assert await libra_crud.get_processed_payment(stale_hash) is None
+    assert await libra_crud.get_processed_payment(poison.payment_hash) is None
+
+
+# ---------------------------------------------------------------------------
+# POST /record-payment — the client-driven path
+# ---------------------------------------------------------------------------
+
+
+async def _insert_paid_payment_row(wallet_id: str, user_id: str) -> str:
+    payment_hash = uuid4().hex + uuid4().hex[:32]
+    await create_payment(
+        checking_id=payment_hash,
+        data=CreatePayment(
+            wallet_id=wallet_id,
+            payment_hash=payment_hash,
+            bolt11="lnbcfake",
+            amount_msat=100_000_000,
+            memo="idempotency test",
+            extra={
+                "tag": "libra",
+                "user_id": user_id,
+                "fiat_currency": "EUR",
+                "fiat_amount": "100.00",
+            },
+        ),
+        status=PaymentState.SUCCESS,
+    )
+    return payment_hash
+
+
+async def test_record_payment_conflicts_while_in_flight(
+    client, super_user_headers, configured_user, standard_accounts
+):
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+    payment_hash = await _insert_paid_payment_row(wallet.id, user.id)
+
+    # Another claimant (e.g. the background listener) is mid-recording.
+    assert await libra_crud.claim_payment(payment_hash)
+
+    r = await client.post(
+        "/libra/api/v1/record-payment",
+        headers={"X-Api-Key": wallet.inkey},
+        json={"payment_hash": payment_hash},
+    )
+    assert r.status_code == 409, r.text
+
+    # Once that claimant finishes, a replay reports "already recorded"
+    # instead of writing a second entry.
+    await libra_crud.mark_payment_done(payment_hash, f"ln-{payment_hash[:16]}")
+    r = await client.post(
+        "/libra/api/v1/record-payment",
+        headers={"X-Api-Key": wallet.inkey},
+        json={"payment_hash": payment_hash},
+    )
+    assert r.status_code == 200, r.text
+    assert "already recorded" in r.json()["message"].lower()
+
+
+async def test_record_payment_records_once_then_replays_safely(
+    client, super_user_headers, configured_user, standard_accounts
+):
+    user, wallet = await _setup_receivable(
+        client, super_user_headers, configured_user, standard_accounts
+    )
+    payment_hash = await _insert_paid_payment_row(wallet.id, user.id)
+
+    r = await client.post(
+        "/libra/api/v1/record-payment",
+        headers={"X-Api-Key": wallet.inkey},
+        json={"payment_hash": payment_hash},
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["message"] == "Payment recorded successfully"
+
+    r = await client.post(
+        "/libra/api/v1/record-payment",
+        headers={"X-Api-Key": wallet.inkey},
+        json={"payment_hash": payment_hash},
+    )
+    assert r.status_code == 200, r.text
+    assert "already recorded" in r.json()["message"].lower()
+
+    link = f"ln-{payment_hash[:16]}"
+    assert len(await _entries_with_link(client, wallet.inkey, link)) == 1

views_api.py

@@ -1850,90 +1850,130 @@ async def api_record_payment(
 
     try:
         async with httpx.AsyncClient(timeout=5.0) as client:
-            # Get recent entries from Fava's journal endpoint
+            # Get recent entries from Fava's journal endpoint. base_url
+            # already ends in /api — the previous "/api/journal" path
+            # 404'd, so this duplicate check silently never ran.
             response = await client.get(
-                f"{fava.base_url}/api/journal",
+                f"{fava.base_url}/journal",
                 params={"time": ""}  # Get all entries
             )
+            response.raise_for_status()
+            response_data = response.json()
+            entries = response_data.get('entries', [])
 
-            if response.status_code == 200:
-                response_data = response.json()
-                entries = response_data.get('entries', [])
-
-                # Check if any entry has our payment link
-                for entry in entries:
-                    entry_links = entry.get('links', [])
-                    if link_to_find in entry_links:
-                        # Payment already recorded, return existing entry
-                        balance_data = await fava.get_user_balance_bql(target_user_id)
-                        return {
-                            "journal_entry_id": f"fava-exists-{data.payment_hash[:16]}",
-                            "new_balance": balance_data["balance"],
-                            "message": "Payment already recorded",
-                        }
-    except Exception as e:
+            # Check if any entry has our payment link
+            for entry in entries:
+                entry_links = entry.get('links', [])
+                if link_to_find in entry_links:
+                    # Payment already recorded, return existing entry
+                    balance_data = await fava.get_user_balance_bql(target_user_id)
+                    return {
+                        "journal_entry_id": f"fava-exists-{data.payment_hash[:16]}",
+                        "new_balance": balance_data["balance"],
+                        "message": "Payment already recorded",
+                    }
+    except httpx.HTTPError as e:
+        # Fail CLOSED: if Fava can't confirm the payment isn't already
+        # recorded, refuse to write — proceeding on a transient blip is
+        # how double entries happen. The client can simply retry.
         logger.warning(f"Could not check Fava for duplicate payment: {e}")
-        # Continue anyway - Fava/Beancount will catch duplicate if it exists
-
-    # Convert amount from millisatoshis to satoshis
-    amount_sats = payment.amount // 1000
-
-    # Extract fiat metadata from invoice (if present)
-    fiat_currency = None
-    fiat_amount = None
-    if payment.extra and isinstance(payment.extra, dict):
-        logger.info(f"Payment.extra contents: {payment.extra}")
-        fiat_currency = payment.extra.get("fiat_currency")
-        fiat_amount_str = payment.extra.get("fiat_amount")
-        if fiat_amount_str:
-            from decimal import Decimal
-            fiat_amount = Decimal(str(fiat_amount_str))
-
-    logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}")
-
-    # Get user's receivable account (what user owes)
-    user_receivable = await get_or_create_user_account(
-        target_user_id, AccountType.ASSET, "Accounts Receivable"
-    )
-
-    # Get lightning account
-    lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
-    if not lightning_account:
         raise HTTPException(
-            status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found"
+            status_code=HTTPStatus.SERVICE_UNAVAILABLE,
+            detail="Cannot verify payment duplicate status; try again shortly",
         )
 
-    # Get unsettled receivable entries to link to this settlement
-    unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable")
-    settled_links = [e["link"] for e in unsettled if e.get("link")]
-
-    # Format payment entry and submit to Fava
-    entry = format_payment_entry(
-        user_id=target_user_id,
-        payment_account=lightning_account.name,
-        payable_or_receivable_account=user_receivable.name,
-        amount_sats=amount_sats,
-        description=f"Lightning payment from user {target_user_id[:8]}",
-        entry_date=datetime.now().date(),
-        is_payable=False,  # User paying libra (receivable settlement)
-        fiat_currency=fiat_currency,
-        fiat_amount=fiat_amount,
-        payment_hash=data.payment_hash,
-        reference=data.payment_hash,
-        settled_entry_links=settled_links
+    # Local idempotency gate shared with the background invoice listener
+    # (tasks.on_invoice_paid): exactly one claimant records a payment_hash.
+    from .crud import (
+        claim_payment,
+        get_processed_payment,
+        mark_payment_done,
+        release_payment_claim,
     )
 
-    logger.info(f"Formatted payment entry: {entry}")
+    if not await claim_payment(data.payment_hash):
+        existing = await get_processed_payment(data.payment_hash)
+        if existing and existing["status"] == "done":
+            balance_data = await fava.get_user_balance_bql(target_user_id)
+            return {
+                "journal_entry_id": existing.get("entry_id")
+                or f"fava-exists-{data.payment_hash[:16]}",
+                "new_balance": balance_data["balance"],
+                "message": "Payment already recorded",
+            }
+        raise HTTPException(
+            status_code=HTTPStatus.CONFLICT,
+            detail="Payment is being recorded; check balance shortly",
+        )
 
-    # Submit to Fava
-    result = await fava.add_entry(entry)
-    logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}")
+    # Convert amount from millisatoshis to satoshis
+    try:
+        amount_sats = payment.amount // 1000
+
+        # Extract fiat metadata from invoice (if present)
+        fiat_currency = None
+        fiat_amount = None
+        if payment.extra and isinstance(payment.extra, dict):
+            logger.info(f"Payment.extra contents: {payment.extra}")
+            fiat_currency = payment.extra.get("fiat_currency")
+            fiat_amount_str = payment.extra.get("fiat_amount")
+            if fiat_amount_str:
+                from decimal import Decimal
+                fiat_amount = Decimal(str(fiat_amount_str))
+
+        logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}")
+
+        # Get user's receivable account (what user owes)
+        user_receivable = await get_or_create_user_account(
+            target_user_id, AccountType.ASSET, "Accounts Receivable"
+        )
+
+        # Get lightning account
+        lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
+        if not lightning_account:
+            raise HTTPException(
+                status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found"
+            )
+
+        # Get unsettled receivable entries to link to this settlement
+        unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable")
+        settled_links = [e["link"] for e in unsettled if e.get("link")]
+
+        # Format payment entry and submit to Fava
+        entry = format_payment_entry(
+            user_id=target_user_id,
+            payment_account=lightning_account.name,
+            payable_or_receivable_account=user_receivable.name,
+            amount_sats=amount_sats,
+            description=f"Lightning payment from user {target_user_id[:8]}",
+            entry_date=datetime.now().date(),
+            is_payable=False,  # User paying libra (receivable settlement)
+            fiat_currency=fiat_currency,
+            fiat_amount=fiat_amount,
+            payment_hash=data.payment_hash,
+            reference=data.payment_hash,
+            settled_entry_links=settled_links
+        )
+
+        logger.info(f"Formatted payment entry: {entry}")
+
+        # Submit to Fava
+        result = await fava.add_entry(entry)
+        logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}")
+
+        entry_id = f"ln-{data.payment_hash[:16]}"
+        await mark_payment_done(data.payment_hash, entry_id)
+    except BaseException:
+        # Release the claim so a retry (client or background listener)
+        # can record this payment.
+        await release_payment_claim(data.payment_hash)
+        raise
 
     # Get updated balance from Fava
     balance_data = await fava.get_user_balance_bql(target_user_id)
 
     return {
-        "journal_entry_id": f"fava-{datetime.now().timestamp()}",
+        "journal_entry_id": entry_id,
         "new_balance": balance_data["balance"],
         "message": "Payment recorded successfully",
     }