Compare commits
No commits in common. "fix/payment-idempotency" and "main" have entirely different histories.
fix/paymen
...
main
6 changed files with 104 additions and 699 deletions
70
crud.py
70
crud.py
|
|
@ -1696,73 +1696,3 @@ 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
|
||||
|
|
|
|||
132
migrations.py
132
migrations.py
|
|
@ -34,33 +34,9 @@ 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).
|
||||
|
|
@ -87,7 +63,7 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS accounts (
|
||||
CREATE TABLE accounts (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL UNIQUE,
|
||||
account_type TEXT NOT NULL,
|
||||
|
|
@ -100,13 +76,13 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_accounts_user_id ON accounts (user_id);
|
||||
CREATE INDEX idx_accounts_user_id ON accounts (user_id);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_accounts_type ON accounts (account_type);
|
||||
CREATE INDEX idx_accounts_type ON accounts (account_type);
|
||||
"""
|
||||
)
|
||||
|
||||
|
|
@ -117,7 +93,7 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS extension_settings (
|
||||
CREATE TABLE extension_settings (
|
||||
id TEXT NOT NULL PRIMARY KEY,
|
||||
libra_wallet_id TEXT,
|
||||
fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333',
|
||||
|
|
@ -135,7 +111,7 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS user_wallet_settings (
|
||||
CREATE TABLE user_wallet_settings (
|
||||
id TEXT NOT NULL PRIMARY KEY,
|
||||
user_wallet_id TEXT,
|
||||
updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
|
||||
|
|
@ -150,7 +126,7 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS manual_payment_requests (
|
||||
CREATE TABLE manual_payment_requests (
|
||||
id TEXT PRIMARY KEY,
|
||||
user_id TEXT NOT NULL,
|
||||
amount INTEGER NOT NULL,
|
||||
|
|
@ -167,14 +143,14 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_user_id
|
||||
CREATE INDEX idx_manual_payment_requests_user_id
|
||||
ON manual_payment_requests (user_id);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_status
|
||||
CREATE INDEX idx_manual_payment_requests_status
|
||||
ON manual_payment_requests (status);
|
||||
"""
|
||||
)
|
||||
|
|
@ -187,7 +163,7 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS balance_assertions (
|
||||
CREATE TABLE balance_assertions (
|
||||
id TEXT PRIMARY KEY,
|
||||
date TIMESTAMP NOT NULL,
|
||||
account_id TEXT NOT NULL,
|
||||
|
|
@ -212,21 +188,21 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_balance_assertions_account_id
|
||||
CREATE INDEX idx_balance_assertions_account_id
|
||||
ON balance_assertions (account_id);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_balance_assertions_status
|
||||
CREATE INDEX idx_balance_assertions_status
|
||||
ON balance_assertions (status);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_balance_assertions_date
|
||||
CREATE INDEX idx_balance_assertions_date
|
||||
ON balance_assertions (date);
|
||||
"""
|
||||
)
|
||||
|
|
@ -240,7 +216,7 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS user_equity_status (
|
||||
CREATE TABLE user_equity_status (
|
||||
user_id TEXT PRIMARY KEY,
|
||||
is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE,
|
||||
equity_account_name TEXT,
|
||||
|
|
@ -254,7 +230,7 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_user_equity_status_eligible
|
||||
CREATE INDEX idx_user_equity_status_eligible
|
||||
ON user_equity_status (is_equity_eligible)
|
||||
WHERE is_equity_eligible = TRUE;
|
||||
"""
|
||||
|
|
@ -269,7 +245,7 @@ async def m001_initial(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS account_permissions (
|
||||
CREATE TABLE account_permissions (
|
||||
id TEXT PRIMARY KEY,
|
||||
user_id TEXT NOT NULL,
|
||||
account_id TEXT NOT NULL,
|
||||
|
|
@ -286,7 +262,7 @@ async def m001_initial(db):
|
|||
# Index for looking up permissions by user
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_account_permissions_user_id
|
||||
CREATE INDEX idx_account_permissions_user_id
|
||||
ON account_permissions (user_id);
|
||||
"""
|
||||
)
|
||||
|
|
@ -294,7 +270,7 @@ async def m001_initial(db):
|
|||
# Index for looking up permissions by account
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_account_permissions_account_id
|
||||
CREATE INDEX idx_account_permissions_account_id
|
||||
ON account_permissions (account_id);
|
||||
"""
|
||||
)
|
||||
|
|
@ -302,7 +278,7 @@ async def m001_initial(db):
|
|||
# Composite index for checking specific user+account permissions
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_account_permissions_user_account
|
||||
CREATE INDEX idx_account_permissions_user_account
|
||||
ON account_permissions (user_id, account_id);
|
||||
"""
|
||||
)
|
||||
|
|
@ -310,7 +286,7 @@ async def m001_initial(db):
|
|||
# Index for finding permissions by type
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_account_permissions_type
|
||||
CREATE INDEX idx_account_permissions_type
|
||||
ON account_permissions (permission_type);
|
||||
"""
|
||||
)
|
||||
|
|
@ -318,7 +294,7 @@ async def m001_initial(db):
|
|||
# Index for finding expired permissions
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_account_permissions_expires
|
||||
CREATE INDEX idx_account_permissions_expires
|
||||
ON account_permissions (expires_at)
|
||||
WHERE expires_at IS NOT NULL;
|
||||
"""
|
||||
|
|
@ -344,7 +320,6 @@ 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()),
|
||||
|
|
@ -367,18 +342,17 @@ async def m002_add_account_is_active(db):
|
|||
|
||||
Default: All existing accounts are marked as active (TRUE).
|
||||
"""
|
||||
await _alter_add_column_safe(
|
||||
db,
|
||||
await db.execute(
|
||||
"""
|
||||
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 IF NOT EXISTS idx_accounts_is_active ON accounts (is_active)
|
||||
CREATE INDEX idx_accounts_is_active ON accounts (is_active)
|
||||
"""
|
||||
)
|
||||
|
||||
|
|
@ -398,18 +372,17 @@ async def m003_add_account_is_virtual(db):
|
|||
|
||||
Default: All existing accounts are real (is_virtual = FALSE).
|
||||
"""
|
||||
await _alter_add_column_safe(
|
||||
db,
|
||||
await db.execute(
|
||||
"""
|
||||
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 IF NOT EXISTS idx_accounts_is_virtual ON accounts (is_virtual)
|
||||
CREATE INDEX idx_accounts_is_virtual ON accounts (is_virtual)
|
||||
"""
|
||||
)
|
||||
|
||||
|
|
@ -429,7 +402,6 @@ 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()),
|
||||
|
|
@ -466,7 +438,7 @@ async def m004_add_rbac_tables(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS roles (
|
||||
CREATE TABLE roles (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL UNIQUE,
|
||||
description TEXT,
|
||||
|
|
@ -479,13 +451,13 @@ async def m004_add_rbac_tables(db):
|
|||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_roles_name ON roles (name);
|
||||
CREATE INDEX idx_roles_name ON roles (name);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_roles_is_default ON roles (is_default)
|
||||
CREATE INDEX idx_roles_is_default ON roles (is_default)
|
||||
WHERE is_default = TRUE;
|
||||
"""
|
||||
)
|
||||
|
|
@ -497,7 +469,7 @@ async def m004_add_rbac_tables(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS role_permissions (
|
||||
CREATE TABLE role_permissions (
|
||||
id TEXT PRIMARY KEY,
|
||||
role_id TEXT NOT NULL,
|
||||
account_id TEXT NOT NULL,
|
||||
|
|
@ -512,19 +484,19 @@ async def m004_add_rbac_tables(db):
|
|||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_role_permissions_role_id ON role_permissions (role_id);
|
||||
CREATE INDEX idx_role_permissions_role_id ON role_permissions (role_id);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_role_permissions_account_id ON role_permissions (account_id);
|
||||
CREATE INDEX idx_role_permissions_account_id ON role_permissions (account_id);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_role_permissions_type ON role_permissions (permission_type);
|
||||
CREATE INDEX idx_role_permissions_type ON role_permissions (permission_type);
|
||||
"""
|
||||
)
|
||||
|
||||
|
|
@ -535,7 +507,7 @@ async def m004_add_rbac_tables(db):
|
|||
|
||||
await db.execute(
|
||||
f"""
|
||||
CREATE TABLE IF NOT EXISTS user_roles (
|
||||
CREATE TABLE user_roles (
|
||||
id TEXT PRIMARY KEY,
|
||||
user_id TEXT NOT NULL,
|
||||
role_id TEXT NOT NULL,
|
||||
|
|
@ -550,19 +522,19 @@ async def m004_add_rbac_tables(db):
|
|||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_user_roles_user_id ON user_roles (user_id);
|
||||
CREATE INDEX idx_user_roles_user_id ON user_roles (user_id);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_user_roles_role_id ON user_roles (role_id);
|
||||
CREATE INDEX idx_user_roles_role_id ON user_roles (role_id);
|
||||
"""
|
||||
)
|
||||
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_user_roles_expires ON user_roles (expires_at)
|
||||
CREATE INDEX idx_user_roles_expires ON user_roles (expires_at)
|
||||
WHERE expires_at IS NOT NULL;
|
||||
"""
|
||||
)
|
||||
|
|
@ -570,7 +542,7 @@ async def m004_add_rbac_tables(db):
|
|||
# Composite index for checking specific user+role assignments
|
||||
await db.execute(
|
||||
"""
|
||||
CREATE INDEX IF NOT EXISTS idx_user_roles_user_role ON user_roles (user_id, role_id);
|
||||
CREATE INDEX idx_user_roles_user_role ON user_roles (user_id, role_id);
|
||||
"""
|
||||
)
|
||||
|
||||
|
|
@ -614,7 +586,6 @@ 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()),
|
||||
|
|
@ -624,30 +595,3 @@ 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}
|
||||
);
|
||||
"""
|
||||
)
|
||||
|
|
|
|||
40
tasks.py
40
tasks.py
|
|
@ -179,31 +179,12 @@ 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()
|
||||
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}"
|
||||
)
|
||||
await on_invoice_paid(payment)
|
||||
|
||||
|
||||
async def on_invoice_paid(payment: Payment) -> None:
|
||||
|
|
@ -229,20 +210,8 @@ 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
|
||||
|
|
@ -276,7 +245,6 @@ 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
|
||||
|
|
@ -308,7 +276,6 @@ 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
|
||||
|
|
@ -357,11 +324,6 @@ 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
|
||||
|
|
|
|||
|
|
@ -1,110 +0,0 @@
|
|||
"""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
|
||||
|
|
@ -1,281 +0,0 @@
|
|||
"""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
|
||||
168
views_api.py
168
views_api.py
|
|
@ -1850,130 +1850,90 @@ async def api_record_payment(
|
|||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||||
# 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.
|
||||
# Get recent entries from Fava's journal endpoint
|
||||
response = await client.get(
|
||||
f"{fava.base_url}/journal",
|
||||
f"{fava.base_url}/api/journal",
|
||||
params={"time": ""} # Get all entries
|
||||
)
|
||||
response.raise_for_status()
|
||||
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 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.
|
||||
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:
|
||||
logger.warning(f"Could not check Fava for duplicate payment: {e}")
|
||||
raise HTTPException(
|
||||
status_code=HTTPStatus.SERVICE_UNAVAILABLE,
|
||||
detail="Cannot verify payment duplicate status; try again shortly",
|
||||
)
|
||||
|
||||
# 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,
|
||||
)
|
||||
|
||||
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",
|
||||
)
|
||||
# Continue anyway - Fava/Beancount will catch duplicate if it exists
|
||||
|
||||
# Convert amount from millisatoshis to satoshis
|
||||
try:
|
||||
amount_sats = payment.amount // 1000
|
||||
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))
|
||||
# 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}")
|
||||
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 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 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")]
|
||||
|
||||
# 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
|
||||
)
|
||||
|
||||
# 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}")
|
||||
|
||||
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
|
||||
# Submit to Fava
|
||||
result = await fava.add_entry(entry)
|
||||
logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}")
|
||||
|
||||
# Get updated balance from Fava
|
||||
balance_data = await fava.get_user_balance_bql(target_user_id)
|
||||
|
||||
return {
|
||||
"journal_entry_id": entry_id,
|
||||
"journal_entry_id": f"fava-{datetime.now().timestamp()}",
|
||||
"new_balance": balance_data["balance"],
|
||||
"message": "Payment recorded successfully",
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue