fix(payments): add local idempotency gate for Lightning recording

The Fava-side duplicate checks (add_entry_idempotent, journal-link
scan) are read-then-write races: on restart with a persisted invoice
queue, or webhook + poller firing together, both callers pass the
"not present" check and both insert.

New processed_payments table (m005) keyed on payment_hash; exactly
one claimant wins the INSERT ... ON CONFLICT DO NOTHING. Lifecycle:
'processing' while the write is in flight, 'done' after; failed
recordings release the claim so redelivery retries, and 'processing'
rows from a crashed process are cleared at listener startup.

Also wraps the invoice-listener loop body in try/except so one poison
payment can't kill payment recording for the process lifetime.

Addresses CODE-REVIEW-2026-06 findings #4 and #9.

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