Compare commits

...
Sign in to create a new pull request.

4 commits

Author SHA1 Message Date
cf1a0967bf fix(settlement): per-currency netting, balance guards, Decimal rates
Settlement correctness cluster from CODE-REVIEW-2026-06 (#2, #8, #13)
plus libra-#38:

- format_net_settlement_entry now enforces the same inline balance
  constraint as the fiat formatter (payment = receivable - payable
  + credit) and grows an optional credit leg. An unbalanced
  settlement raises instead of reaching the ledger.
- on_invoice_paid settles only what the payment covers: a partial
  payment clears that much receivable; excess (or a payment with
  nothing owed) becomes user credit. Previously the full prior
  balance was cleared against a smaller payment, shipping unbalanced
  postings. Settlement links are attached only when the payment
  clears the full open balance, and only for same-currency entries.
- get_unsettled_entries_bql returns each entry's real posting
  currency (was hardcoded "EUR") and exact Decimal amount strings
  (was float). /receivables/settle nets only entries denominated in
  the settlement currency.
- fiat_rate/btc_rate metadata computed via Decimal (new
  fiat_rate_metadata helper) instead of float division — cost-basis
  records no longer carry float drift.
- format_posting_at_average_cost omits the cost braces when
  cost_currency is unset ("SATS {}" is invalid Beancount).
- Underpay error payload serializes amounts as exact Decimal strings.
- validate_metadata catches decimal.InvalidOperation so bad
  fiat_amount input becomes ValidationError (libra-#38); flipped the
  tracking xfail.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-12 12:54:46 +02:00
44e10caac7 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>
2026-07-12 12:41:42 +02:00
4fdb358bb0 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>
2026-07-12 12:33:42 +02:00
c50455d5f6 fix(migrations): make all migrations idempotent
The migration version bump lands in the core dbversions table while
the DDL lands in ext_libra — the two writes are not atomic. A failed
bump re-runs the whole migration on next boot; bare
CREATE/ALTER/INSERT then crashes the extension until manual
dbversions surgery.

- CREATE TABLE / CREATE INDEX -> IF NOT EXISTS
- ALTER TABLE ADD COLUMN -> _alter_add_column_safe (same swallow
  pattern as the events/withdraw fork migrations)
- seed INSERTs (default accounts, virtual parents, default roles) ->
  ON CONFLICT (name) DO NOTHING

Tests run the chain twice against a fresh SQLite DB (full-chain rerun
and per-migration rerun); both fail against the previous migrations.

Addresses CODE-REVIEW-2026-06 findings #3 and #12.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-12 12:23:04 +02:00
11 changed files with 975 additions and 172 deletions

View file

@ -252,9 +252,10 @@ def format_posting_at_average_cost(
amount_str = f"{amount_sats} SATS {{{cost_currency}}}" amount_str = f"{amount_sats} SATS {{{cost_currency}}}"
logger.info(f"format_posting_at_average_cost: Generated amount_str='{amount_str}' with cost_currency='{cost_currency}'") logger.info(f"format_posting_at_average_cost: Generated amount_str='{amount_str}' with cost_currency='{cost_currency}'")
else: else:
# No cost # No cost basis — omit the braces entirely. Empty "{}" is not
amount_str = f"{amount_sats} SATS {{}}" # valid Beancount syntax and fails to parse on ledger load.
logger.warning(f"format_posting_at_average_cost: cost_currency is None, using empty cost basis") amount_str = f"{amount_sats} SATS"
logger.warning(f"format_posting_at_average_cost: cost_currency is None, omitting cost basis")
posting_meta = metadata or {} posting_meta = metadata or {}
@ -304,6 +305,25 @@ def format_posting_simple(
} }
def fiat_rate_metadata(amount_sats: int, fiat_amount: Decimal) -> Dict[str, str]:
"""Exchange-rate metadata (sats per fiat unit, fiat per BTC) as exact
Decimal strings.
These values become the cost-basis record for the entry, so they must
not carry float drift CLAUDE.md mandates Decimal for all fiat math.
Returns:
{"fiat_rate": "<sats per fiat unit>", "btc_rate": "<fiat per BTC>"}
"""
if amount_sats <= 0 or fiat_amount <= 0:
return {"fiat_rate": "0", "btc_rate": "0"}
fiat_rate = (Decimal(amount_sats) / fiat_amount).quantize(Decimal("0.000001"))
btc_rate = (
fiat_amount / Decimal(amount_sats) * Decimal(100_000_000)
).quantize(Decimal("0.01"))
return {"fiat_rate": str(fiat_rate), "btc_rate": str(btc_rate)}
def format_expense_entry( def format_expense_entry(
user_id: str, user_id: str,
expense_account: str, expense_account: str,
@ -718,15 +738,18 @@ def format_net_settlement_entry(
entry_date: date, entry_date: date,
payment_hash: Optional[str] = None, payment_hash: Optional[str] = None,
reference: Optional[str] = None, reference: Optional[str] = None,
settled_entry_links: Optional[List[str]] = None settled_entry_links: Optional[List[str]] = None,
credit_account: Optional[str] = None,
credit_overflow_fiat: Decimal = Decimal(0),
) -> Dict[str, Any]: ) -> Dict[str, Any]:
""" """
Format a net settlement payment entry (user paying net balance). Format a net settlement payment entry (user paying net balance).
Creates a three-posting transaction: Creates a three- to four-posting transaction:
1. Lightning payment in SATS with @@ total price notation 1. Lightning payment in SATS with @@ total price notation
2. Clear receivables in EUR 2. Clear receivables in EUR
3. Clear payables in EUR 3. Clear payables in EUR
4. Credit overflow when the payment exceeds what it clears
Example: Example:
Assets:Bitcoin:Lightning 565251 SATS @@ 517.00 EUR Assets:Bitcoin:Lightning 565251 SATS @@ 517.00 EUR
@ -734,25 +757,61 @@ def format_net_settlement_entry(
Liabilities:Payable:User 38.00 EUR Liabilities:Payable:User 38.00 EUR
= 517 - 555 + 38 = 0 = 517 - 555 + 38 = 0
Constraint enforced inline (same contract as
`format_fiat_net_settlement_entry`):
net_fiat_amount = total_receivable_fiat - total_payable_fiat
+ credit_overflow_fiat
Args: Args:
user_id: User ID user_id: User ID
payment_account: Payment account (e.g., "Assets:Bitcoin:Lightning") payment_account: Payment account (e.g., "Assets:Bitcoin:Lightning")
receivable_account: User's receivable account receivable_account: User's receivable account
payable_account: User's payable account payable_account: User's payable account
amount_sats: SATS amount paid amount_sats: SATS amount paid
net_fiat_amount: Net fiat amount (receivable - payable) net_fiat_amount: Fiat value of the payment being recorded
total_receivable_fiat: Total receivables to clear total_receivable_fiat: Receivables cleared by this payment
total_payable_fiat: Total payables to clear total_payable_fiat: Payables cleared by this payment
fiat_currency: Currency (EUR, USD) fiat_currency: Currency (EUR, USD)
description: Payment description description: Payment description
entry_date: Date of payment entry_date: Date of payment
payment_hash: Lightning payment hash payment_hash: Lightning payment hash
reference: Optional reference reference: Optional reference
settled_entry_links: List of expense/receivable links being settled (e.g., ["exp-abc123", "rcv-def456"]) settled_entry_links: List of expense/receivable links being settled (e.g., ["exp-abc123", "rcv-def456"])
credit_account: User's credit account receiving overflow (required
when credit_overflow_fiat > 0)
credit_overflow_fiat: Payment excess beyond what it clears, absorbed
as a liability libra owes the user going forward
Returns: Returns:
Fava API entry dict Fava API entry dict
Raises:
ValueError: if any amount is negative, or the payment doesn't
balance against what it clears an unbalanced settlement
must never reach the ledger.
""" """
for label, value in (
("net_fiat_amount", net_fiat_amount),
("total_receivable_fiat", total_receivable_fiat),
("total_payable_fiat", total_payable_fiat),
("credit_overflow_fiat", credit_overflow_fiat),
):
if value < 0:
raise ValueError(f"{label} must be non-negative; got {value}")
expected_payment = (
total_receivable_fiat - total_payable_fiat + credit_overflow_fiat
)
if abs(net_fiat_amount - expected_payment) > Decimal("0.01"):
raise ValueError(
f"net_fiat_amount {net_fiat_amount} does not match expected "
f"{expected_payment} (= receivable {total_receivable_fiat} "
f"- payable {total_payable_fiat} + credit {credit_overflow_fiat}); "
f"refusing to write an unbalanced settlement"
)
if credit_overflow_fiat > 0 and not credit_account:
raise ValueError("credit_account required when credit_overflow_fiat > 0")
# Build postings for net settlement # Build postings for net settlement
# Note: We use @@ (total price) syntax for cleaner formatting, but Fava's API # Note: We use @@ (total price) syntax for cleaner formatting, but Fava's API
# will convert this to @ (per-unit price) with a long decimal when writing to file. # will convert this to @ (per-unit price) with a long decimal when writing to file.
@ -761,20 +820,26 @@ def format_net_settlement_entry(
postings = [ postings = [
{ {
"account": payment_account, "account": payment_account,
"amount": f"{abs(amount_sats)} SATS @@ {abs(net_fiat_amount):.2f} {fiat_currency}", "amount": f"{abs(amount_sats)} SATS @@ {net_fiat_amount:.2f} {fiat_currency}",
"meta": {"payment-hash": payment_hash} if payment_hash else {} "meta": {"payment-hash": payment_hash} if payment_hash else {}
}, },
{ {
"account": receivable_account, "account": receivable_account,
"amount": f"-{abs(total_receivable_fiat):.2f} {fiat_currency}", "amount": f"-{total_receivable_fiat:.2f} {fiat_currency}",
"meta": {"sats-equivalent": str(abs(amount_sats))} "meta": {"sats-equivalent": str(abs(amount_sats))}
}, },
{ {
"account": payable_account, "account": payable_account,
"amount": f"{abs(total_payable_fiat):.2f} {fiat_currency}", "amount": f"{total_payable_fiat:.2f} {fiat_currency}",
"meta": {} "meta": {}
} }
] ]
if credit_overflow_fiat > 0:
postings.append({
"account": credit_account,
"amount": f"-{credit_overflow_fiat:.2f} {fiat_currency}",
"meta": {}
})
entry_meta = { entry_meta = {
"user-id": user_id, "user-id": user_id,

View file

@ -5,7 +5,7 @@ Comprehensive validation following Beancount's plugin system approach,
but implemented as simple functions that can be called directly. but implemented as simple functions that can be called directly.
""" """
from decimal import Decimal from decimal import Decimal, InvalidOperation
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
@ -278,11 +278,14 @@ def validate_metadata(
} }
) )
# Validate fiat amount is valid Decimal # Validate fiat amount is valid Decimal. InvalidOperation is what
# Decimal actually raises on garbage input ("abc") — it is not a
# ValueError subclass, so without it the raw exception leaked to
# callers (libra-#38).
if has_fiat_amount: if has_fiat_amount:
try: try:
Decimal(str(metadata["fiat_amount"])) Decimal(str(metadata["fiat_amount"]))
except (ValueError, TypeError) as e: except (ValueError, TypeError, InvalidOperation) as e:
raise ValidationError( raise ValidationError(
f"Invalid fiat_amount: {metadata['fiat_amount']}", f"Invalid fiat_amount: {metadata['fiat_amount']}",
{"error": str(e)} {"error": str(e)}

70
crud.py
View file

@ -1696,3 +1696,73 @@ async def check_user_has_role_permission(
return True return True
return False 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

View file

@ -1817,7 +1817,7 @@ class FavaClient:
# Query 1: Get all original expense/receivable entries for this user # Query 1: Get all original expense/receivable entries for this user
# These are entries with the expense-entry or receivable-entry tag # These are entries with the expense-entry or receivable-entry tag
original_query = f""" original_query = f"""
SELECT date, narration, account, number, weight, links, SELECT date, narration, account, number, currency, weight, links,
any_meta('entry-id') as entry_id any_meta('entry-id') as entry_id
WHERE account ~ '{account_pattern}' WHERE account ~ '{account_pattern}'
AND '{entry_tag}' IN tags AND '{entry_tag}' IN tags
@ -1851,7 +1851,10 @@ class FavaClient:
entries_by_link: Dict[str, Dict[str, Any]] = {} entries_by_link: Dict[str, Dict[str, Any]] = {}
for row in original_result["rows"]: for row in original_result["rows"]:
date_val, narration, account, number, weight, links, entry_id = row (
date_val, narration, account, number, currency,
weight, links, entry_id,
) = row
# Skip if no links # Skip if no links
if not links or not isinstance(links, list): if not links or not isinstance(links, list):
@ -1875,9 +1878,11 @@ class FavaClient:
if entry_link in entries_by_link: if entry_link in entries_by_link:
continue continue
# Parse amounts # Parse amounts. The posting's real currency matters: callers
fiat_amount = abs(float(number)) if number else 0.0 # net these totals per currency, and the old hardcoded "EUR"
fiat_currency = "EUR" # Default, could be extracted from posting # let USD (or SATS-only) entries be summed as if they were EUR.
fiat_amount = str(abs(Decimal(str(number)))) if number else "0"
fiat_currency = currency or "EUR"
# Parse SATS from weight column # Parse SATS from weight column
sats_amount = 0 sats_amount = 0

View file

@ -34,9 +34,33 @@ Original migration sequence (Nov 2025):
- m014: Removed legacy equity accounts (MemberEquity, RetainedEarnings) - m014: Removed legacy equity accounts (MemberEquity, RetainedEarnings)
- m015: Converted entry_lines to single amount field - m015: Converted entry_lines to single amount field
- m016: Dropped journal_entries and entry_lines tables (Fava integration) - 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): async def m001_initial(db):
""" """
Initial Libra database schema (squashed from m001-m016). Initial Libra database schema (squashed from m001-m016).
@ -63,7 +87,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE accounts ( CREATE TABLE IF NOT EXISTS accounts (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE, name TEXT NOT NULL UNIQUE,
account_type TEXT NOT NULL, account_type TEXT NOT NULL,
@ -76,13 +100,13 @@ async def m001_initial(db):
await db.execute( 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( 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( await db.execute(
f""" f"""
CREATE TABLE extension_settings ( CREATE TABLE IF NOT EXISTS extension_settings (
id TEXT NOT NULL PRIMARY KEY, id TEXT NOT NULL PRIMARY KEY,
libra_wallet_id TEXT, libra_wallet_id TEXT,
fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333', fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333',
@ -111,7 +135,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE user_wallet_settings ( CREATE TABLE IF NOT EXISTS user_wallet_settings (
id TEXT NOT NULL PRIMARY KEY, id TEXT NOT NULL PRIMARY KEY,
user_wallet_id TEXT, user_wallet_id TEXT,
updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now} updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now}
@ -126,7 +150,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE manual_payment_requests ( CREATE TABLE IF NOT EXISTS manual_payment_requests (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
amount INTEGER NOT NULL, amount INTEGER NOT NULL,
@ -143,14 +167,14 @@ async def m001_initial(db):
await db.execute( 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); ON manual_payment_requests (user_id);
""" """
) )
await db.execute( 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); ON manual_payment_requests (status);
""" """
) )
@ -163,7 +187,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE balance_assertions ( CREATE TABLE IF NOT EXISTS balance_assertions (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
date TIMESTAMP NOT NULL, date TIMESTAMP NOT NULL,
account_id TEXT NOT NULL, account_id TEXT NOT NULL,
@ -188,21 +212,21 @@ async def m001_initial(db):
await db.execute( 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); ON balance_assertions (account_id);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX idx_balance_assertions_status CREATE INDEX IF NOT EXISTS idx_balance_assertions_status
ON balance_assertions (status); ON balance_assertions (status);
""" """
) )
await db.execute( await db.execute(
""" """
CREATE INDEX idx_balance_assertions_date CREATE INDEX IF NOT EXISTS idx_balance_assertions_date
ON balance_assertions (date); ON balance_assertions (date);
""" """
) )
@ -216,7 +240,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE user_equity_status ( CREATE TABLE IF NOT EXISTS user_equity_status (
user_id TEXT PRIMARY KEY, user_id TEXT PRIMARY KEY,
is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE, is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE,
equity_account_name TEXT, equity_account_name TEXT,
@ -230,7 +254,7 @@ async def m001_initial(db):
await db.execute( 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) ON user_equity_status (is_equity_eligible)
WHERE is_equity_eligible = TRUE; WHERE is_equity_eligible = TRUE;
""" """
@ -245,7 +269,7 @@ async def m001_initial(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE account_permissions ( CREATE TABLE IF NOT EXISTS account_permissions (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
account_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 # Index for looking up permissions by user
await db.execute( 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); ON account_permissions (user_id);
""" """
) )
@ -270,7 +294,7 @@ async def m001_initial(db):
# Index for looking up permissions by account # Index for looking up permissions by account
await db.execute( 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); ON account_permissions (account_id);
""" """
) )
@ -278,7 +302,7 @@ async def m001_initial(db):
# Composite index for checking specific user+account permissions # Composite index for checking specific user+account permissions
await db.execute( 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); ON account_permissions (user_id, account_id);
""" """
) )
@ -286,7 +310,7 @@ async def m001_initial(db):
# Index for finding permissions by type # Index for finding permissions by type
await db.execute( await db.execute(
""" """
CREATE INDEX idx_account_permissions_type CREATE INDEX IF NOT EXISTS idx_account_permissions_type
ON account_permissions (permission_type); ON account_permissions (permission_type);
""" """
) )
@ -294,7 +318,7 @@ async def m001_initial(db):
# Index for finding expired permissions # Index for finding expired permissions
await db.execute( await db.execute(
""" """
CREATE INDEX idx_account_permissions_expires CREATE INDEX IF NOT EXISTS idx_account_permissions_expires
ON account_permissions (expires_at) ON account_permissions (expires_at)
WHERE expires_at IS NOT NULL; WHERE expires_at IS NOT NULL;
""" """
@ -320,6 +344,7 @@ async def m001_initial(db):
f""" f"""
INSERT INTO accounts (id, name, account_type, description, created_at) INSERT INTO accounts (id, name, account_type, description, created_at)
VALUES (:id, :name, :type, :description, {db.timestamp_now}) VALUES (:id, :name, :type, :description, {db.timestamp_now})
ON CONFLICT (name) DO NOTHING
""", """,
{ {
"id": str(uuid.uuid4()), "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). Default: All existing accounts are marked as active (TRUE).
""" """
await db.execute( await _alter_add_column_safe(
db,
""" """
ALTER TABLE accounts ALTER TABLE accounts
ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT TRUE ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT TRUE
""" """,
) )
# Create index for faster queries filtering by is_active # Create index for faster queries filtering by is_active
await db.execute( 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). Default: All existing accounts are real (is_virtual = FALSE).
""" """
await db.execute( await _alter_add_column_safe(
db,
""" """
ALTER TABLE accounts ALTER TABLE accounts
ADD COLUMN is_virtual BOOLEAN NOT NULL DEFAULT FALSE ADD COLUMN is_virtual BOOLEAN NOT NULL DEFAULT FALSE
""" """,
) )
# Create index for faster queries filtering by is_virtual # Create index for faster queries filtering by is_virtual
await db.execute( 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""" f"""
INSERT INTO accounts (id, name, account_type, description, is_active, is_virtual, created_at) INSERT INTO accounts (id, name, account_type, description, is_active, is_virtual, created_at)
VALUES (:id, :name, :type, :description, TRUE, TRUE, {db.timestamp_now}) VALUES (:id, :name, :type, :description, TRUE, TRUE, {db.timestamp_now})
ON CONFLICT (name) DO NOTHING
""", """,
{ {
"id": str(uuid.uuid4()), "id": str(uuid.uuid4()),
@ -438,7 +466,7 @@ async def m004_add_rbac_tables(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE roles ( CREATE TABLE IF NOT EXISTS roles (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE, name TEXT NOT NULL UNIQUE,
description TEXT, description TEXT,
@ -451,13 +479,13 @@ async def m004_add_rbac_tables(db):
await db.execute( 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( 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; WHERE is_default = TRUE;
""" """
) )
@ -469,7 +497,7 @@ async def m004_add_rbac_tables(db):
await db.execute( await db.execute(
f""" f"""
CREATE TABLE role_permissions ( CREATE TABLE IF NOT EXISTS role_permissions (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
role_id TEXT NOT NULL, role_id TEXT NOT NULL,
account_id TEXT NOT NULL, account_id TEXT NOT NULL,
@ -484,19 +512,19 @@ async def m004_add_rbac_tables(db):
await db.execute( 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( 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( 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( await db.execute(
f""" f"""
CREATE TABLE user_roles ( CREATE TABLE IF NOT EXISTS user_roles (
id TEXT PRIMARY KEY, id TEXT PRIMARY KEY,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
role_id TEXT NOT NULL, role_id TEXT NOT NULL,
@ -522,19 +550,19 @@ async def m004_add_rbac_tables(db):
await db.execute( 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( 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( 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; 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 # Composite index for checking specific user+role assignments
await db.execute( 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""" f"""
INSERT INTO roles (id, name, description, is_default, created_by, created_at) INSERT INTO roles (id, name, description, is_default, created_by, created_at)
VALUES (:id, :name, :description, :is_default, :created_by, {db.timestamp_now}) VALUES (:id, :name, :description, :is_default, :created_by, {db.timestamp_now})
ON CONFLICT (name) DO NOTHING
""", """,
{ {
"id": str(uuid.uuid4()), "id": str(uuid.uuid4()),
@ -595,3 +624,30 @@ async def m004_add_rbac_tables(db):
"created_by": "system", # System-created default roles "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}
);
"""
)

View file

@ -179,12 +179,31 @@ async def wait_for_paid_invoices():
This ensures payments are recorded even if the user closes their browser This ensures payments are recorded even if the user closes their browser
before the payment is detected by client-side polling. before the payment is detected by client-side polling.
""" """
from .crud import clear_stale_payment_claims
invoice_queue = Queue() invoice_queue = Queue()
register_invoice_listener(invoice_queue, "ext_libra") 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: while True:
payment = await invoice_queue.get() payment = await invoice_queue.get()
try:
await on_invoice_paid(payment) 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: 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") logger.warning(f"Libra invoice {payment.payment_hash} missing user_id in metadata")
return return
from .crud import claim_payment, mark_payment_done, release_payment_claim
from .fava_client import get_fava_client 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() fava = get_fava_client()
# Use idempotency key based on payment hash - this ensures duplicate # Use idempotency key based on payment hash - this ensures duplicate
@ -245,26 +276,36 @@ async def on_invoice_paid(payment: Payment) -> None:
if not fiat_currency or not fiat_amount: if not fiat_currency or not fiat_amount:
logger.error(f"Payment {payment.payment_hash} missing fiat currency/amount metadata") logger.error(f"Payment {payment.payment_hash} missing fiat currency/amount metadata")
await release_payment_claim(payment.payment_hash)
return return
# Get user's current balance to determine receivables and payables # Get user's current balance to determine what this payment clears
balance = await fava.get_user_balance(user_id) balance = await fava.get_user_balance(user_id)
fiat_balances = balance.get("fiat_balances", {}) fiat_balances = balance.get("fiat_balances", {})
total_fiat_balance = fiat_balances.get(fiat_currency, Decimal(0)) total_fiat_balance = fiat_balances.get(fiat_currency, Decimal(0))
# Determine receivables and payables based on balance # Settle only what this payment covers. The balance is already
# Positive balance = user owes libra (receivable) # net (positive = user owes libra); a partial payment clears
# Negative balance = libra owes user (payable) # that much receivable, and any excess — or the whole payment
if total_fiat_balance > 0: # when nothing is owed — becomes credit libra owes the user.
# User owes libra # (Previously partial payments cleared the FULL balance against
total_receivable = total_fiat_balance # a smaller payment, shipping unbalanced postings.)
tolerance = Decimal("0.01")
open_receivable = (
total_fiat_balance if total_fiat_balance > 0 else Decimal(0)
)
total_receivable = min(open_receivable, fiat_amount)
total_payable = Decimal(0) total_payable = Decimal(0)
else: credit_overflow = fiat_amount - total_receivable
# Libra owes user if credit_overflow < tolerance:
total_receivable = Decimal(0) # Absorb sub-cent rounding into the receivable leg.
total_payable = abs(total_fiat_balance) credit_overflow = Decimal(0)
total_receivable = fiat_amount
logger.info(f"Settlement: {fiat_amount} {fiat_currency} (Receivable: {total_receivable}, Payable: {total_payable})") logger.info(
f"Settlement: {fiat_amount} {fiat_currency} "
f"(clears receivable: {total_receivable}, credit: {credit_overflow})"
)
# Get account names # Get account names
user_receivable = await get_or_create_user_account( user_receivable = await get_or_create_user_account(
@ -273,19 +314,32 @@ async def on_invoice_paid(payment: Payment) -> None:
user_payable = await get_or_create_user_account( user_payable = await get_or_create_user_account(
user_id, AccountType.LIABILITY, "Accounts Payable" user_id, AccountType.LIABILITY, "Accounts Payable"
) )
user_credit = None
if credit_overflow > 0:
user_credit = await get_or_create_user_account(
user_id, AccountType.LIABILITY, "Credit"
)
lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning") lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning")
if not lightning_account: if not lightning_account:
logger.error("Lightning account 'Assets:Bitcoin:Lightning' not found") logger.error("Lightning account 'Assets:Bitcoin:Lightning' not found")
await release_payment_claim(payment.payment_hash)
return return
# Query for unsettled entries to link this settlement back to them # Link the source entries this settlement reconciles — but only
# Net settlement can settle both expenses and receivables # when the payment clears the full open balance. On a partial
# payment we can't know which entries are covered, and linking
# them would make get_unsettled_entries_bql treat them as
# settled. Only same-currency entries qualify either way.
settled_links = [] settled_links = []
if open_receivable > 0 and fiat_amount + tolerance >= open_receivable:
try: try:
unsettled_expenses = await fava.get_unsettled_entries_bql(user_id, "expense") unsettled_expenses = await fava.get_unsettled_entries_bql(user_id, "expense")
settled_links.extend([e["link"] for e in unsettled_expenses if e.get("link")])
unsettled_receivables = await fava.get_unsettled_entries_bql(user_id, "receivable") unsettled_receivables = await fava.get_unsettled_entries_bql(user_id, "receivable")
settled_links.extend([e["link"] for e in unsettled_receivables if e.get("link")]) settled_links.extend(
e["link"]
for e in unsettled_expenses + unsettled_receivables
if e.get("link") and e.get("fiat_currency") == fiat_currency
)
except Exception as e: except Exception as e:
logger.warning(f"Could not query unsettled entries for settlement links: {e}") logger.warning(f"Could not query unsettled entries for settlement links: {e}")
# Continue without links - settlement will still be recorded # Continue without links - settlement will still be recorded
@ -305,7 +359,9 @@ async def on_invoice_paid(payment: Payment) -> None:
entry_date=datetime.now().date(), entry_date=datetime.now().date(),
payment_hash=payment.payment_hash, payment_hash=payment.payment_hash,
reference=payment.payment_hash, reference=payment.payment_hash,
settled_entry_links=settled_links if settled_links else None settled_entry_links=settled_links if settled_links else None,
credit_account=user_credit.name if user_credit else None,
credit_overflow_fiat=credit_overflow,
) )
# Submit to Fava using idempotent method to prevent duplicates # Submit to Fava using idempotent method to prevent duplicates
@ -324,6 +380,11 @@ async def on_invoice_paid(payment: Payment) -> None:
f"{result.get('data', 'Unknown')}" f"{result.get('data', 'Unknown')}"
) )
await mark_payment_done(payment.payment_hash, idempotency_key)
except Exception as e: except Exception as e:
logger.error(f"Error recording Libra payment {payment.payment_hash}: {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 raise

110
tests/test_migrations.py Normal file
View file

@ -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

View file

@ -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

View file

@ -10,6 +10,7 @@ Underpay without explicit entry-picks returns 400 with diff details so
the operator can either pay the exact net or specify `settled_entry_links`. the operator can either pay the exact net or specify `settled_entry_links`.
""" """
import importlib import importlib
from decimal import Decimal
from uuid import uuid4 from uuid import uuid4
import pytest import pytest
@ -268,10 +269,12 @@ async def test_underpay_without_explicit_links_returns_400(
assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}" assert r.status_code == 400, f"expected 400, got {r.status_code}: {r.text}"
payload = r.json().get("detail") payload = r.json().get("detail")
assert isinstance(payload, dict), f"expected structured detail, got {payload!r}" assert isinstance(payload, dict), f"expected structured detail, got {payload!r}"
assert payload.get("cash_paid") == 30.0 # Amounts are exact Decimal strings (not floats) so the operator can
assert payload.get("net_obligation") == 100.0 # act on them without precision loss.
assert payload.get("receivable_total") == 100.0 assert Decimal(payload.get("cash_paid")) == Decimal("30.00")
assert payload.get("payable_total") == 0.0 assert Decimal(payload.get("net_obligation")) == Decimal("100.00")
assert Decimal(payload.get("receivable_total")) == Decimal("100.00")
assert Decimal(payload.get("payable_total")) == Decimal("0")
@pytest.mark.anyio @pytest.mark.anyio

View file

@ -553,11 +553,6 @@ def test_validate_metadata_fiat_amount_without_currency_raises():
val.validate_metadata({"fiat_amount": "10.00"}) val.validate_metadata({"fiat_amount": "10.00"})
@pytest.mark.xfail(
reason="libra/issues/38 — except clause doesn't catch decimal.InvalidOperation, "
"so the raw exception leaks instead of becoming ValidationError. Flip when fixed.",
strict=True,
)
def test_validate_metadata_fiat_amount_invalid_decimal_raises(): def test_validate_metadata_fiat_amount_invalid_decimal_raises():
with pytest.raises(val.ValidationError) as exc: with pytest.raises(val.ValidationError) as exc:
val.validate_metadata({"fiat_amount": "not-a-number", "fiat_currency": "EUR"}) val.validate_metadata({"fiat_amount": "not-a-number", "fiat_currency": "EUR"})
@ -570,3 +565,111 @@ def test_validate_metadata_both_present_passes():
def test_validate_metadata_neither_present_passes(): def test_validate_metadata_neither_present_passes():
val.validate_metadata({"source": "api"}) val.validate_metadata({"source": "api"})
# ---------------------------------------------------------------------------
# format_net_settlement_entry — balance guard + credit overflow
# ---------------------------------------------------------------------------
def _net_settlement(**overrides):
kwargs = dict(
user_id="abc12345",
payment_account="Assets:Bitcoin:Lightning",
receivable_account="Assets:Receivable:User-abc12345",
payable_account="Liabilities:Payable:User-abc12345",
amount_sats=565251,
net_fiat_amount=Decimal("517.00"),
total_receivable_fiat=Decimal("555.00"),
total_payable_fiat=Decimal("38.00"),
fiat_currency="EUR",
description="test settlement",
entry_date=date(2026, 7, 12),
payment_hash="ff" * 32,
)
kwargs.update(overrides)
return bf.format_net_settlement_entry(**kwargs)
def test_net_settlement_balanced_passes():
entry = _net_settlement()
amounts = [p["amount"] for p in entry["postings"]]
assert any("@@ 517.00 EUR" in a for a in amounts)
assert "-555.00 EUR" in amounts
assert "38.00 EUR" in amounts
def test_net_settlement_unbalanced_partial_payment_raises():
# Payment of 300 can't clear a 555 receivable net of 38 payable —
# this is the pre-fix partial-payment shape that shipped unbalanced
# postings to the ledger.
with pytest.raises(ValueError, match="unbalanced"):
_net_settlement(net_fiat_amount=Decimal("300.00"))
def test_net_settlement_negative_amount_raises():
with pytest.raises(ValueError, match="non-negative"):
_net_settlement(total_receivable_fiat=Decimal("-1.00"))
def test_net_settlement_credit_overflow_adds_leg():
entry = _net_settlement(
net_fiat_amount=Decimal("600.00"),
credit_account="Liabilities:Credit:User-abc12345",
credit_overflow_fiat=Decimal("83.00"),
)
amounts = [p["amount"] for p in entry["postings"]]
assert "-83.00 EUR" in amounts
def test_net_settlement_credit_overflow_without_account_raises():
with pytest.raises(ValueError, match="credit_account"):
_net_settlement(
net_fiat_amount=Decimal("600.00"),
credit_overflow_fiat=Decimal("83.00"),
)
# ---------------------------------------------------------------------------
# format_posting_at_average_cost — no empty cost braces
# ---------------------------------------------------------------------------
def test_average_cost_posting_without_currency_omits_braces():
posting = bf.format_posting_at_average_cost(
account="Assets:Receivable:User-abc", amount_sats=-996896,
)
assert posting["amount"] == "-996896 SATS"
assert "{" not in posting["amount"]
def test_average_cost_posting_with_currency_keeps_braces():
posting = bf.format_posting_at_average_cost(
account="Assets:Receivable:User-abc",
amount_sats=-996896,
cost_currency="EUR",
)
assert posting["amount"] == "-996896 SATS {EUR}"
# ---------------------------------------------------------------------------
# fiat_rate_metadata — Decimal-exact rate strings
# ---------------------------------------------------------------------------
def test_fiat_rate_metadata_is_exact():
meta = bf.fiat_rate_metadata(107419, Decimal("100.00"))
assert meta["fiat_rate"] == "1074.190000"
assert meta["btc_rate"] == "93093.40"
# No float artifacts like 1074.1899999999998
Decimal(meta["fiat_rate"])
Decimal(meta["btc_rate"])
def test_fiat_rate_metadata_zero_amounts():
assert bf.fiat_rate_metadata(0, Decimal("100.00")) == {
"fiat_rate": "0", "btc_rate": "0",
}
assert bf.fiat_rate_metadata(1000, Decimal("0")) == {
"fiat_rate": "0", "btc_rate": "0",
}

View file

@ -13,6 +13,7 @@ from lnbits.decorators import (
) )
from lnbits.utils.exchange_rates import allowed_currencies, fiat_amount_as_satoshis from lnbits.utils.exchange_rates import allowed_currencies, fiat_amount_as_satoshis
from .beancount_format import fiat_rate_metadata
from .crud import ( from .crud import (
approve_manual_payment_request, approve_manual_payment_request,
check_balance_assertion, check_balance_assertion,
@ -1074,8 +1075,7 @@ async def api_create_expense_entry(
metadata = { metadata = {
"fiat_currency": data.currency.upper(), "fiat_currency": data.currency.upper(),
"fiat_amount": str(data.amount.quantize(Decimal("0.001"))), # Store as string with 3 decimal places "fiat_amount": str(data.amount.quantize(Decimal("0.001"))), # Store as string with 3 decimal places
"fiat_rate": float(amount_sats) / float(data.amount) if data.amount > 0 else 0, **fiat_rate_metadata(amount_sats, data.amount),
"btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0,
} }
# Get or create expense account # Get or create expense account
@ -1272,8 +1272,7 @@ async def api_create_income_entry(
metadata = { metadata = {
"fiat_currency": fiat_currency, "fiat_currency": fiat_currency,
"fiat_amount": str(data.amount.quantize(Decimal("0.001"))), "fiat_amount": str(data.amount.quantize(Decimal("0.001"))),
"fiat_rate": float(amount_sats) / float(data.amount) if data.amount > 0 else 0, **fiat_rate_metadata(amount_sats, data.amount),
"btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0,
} }
# Submit to Fava # Submit to Fava
@ -1371,8 +1370,7 @@ async def api_create_receivable_entry(
metadata = { metadata = {
"fiat_currency": data.currency.upper(), "fiat_currency": data.currency.upper(),
"fiat_amount": str(data.amount.quantize(Decimal("0.001"))), # Store as string with 3 decimal places "fiat_amount": str(data.amount.quantize(Decimal("0.001"))), # Store as string with 3 decimal places
"fiat_rate": float(amount_sats) / float(data.amount) if data.amount > 0 else 0, **fiat_rate_metadata(amount_sats, data.amount),
"btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0,
} }
# Get or create revenue account # Get or create revenue account
@ -1760,15 +1758,10 @@ async def api_generate_payment_invoice(
proportion = Decimal(data.amount) / Decimal(total_sat_balance) proportion = Decimal(data.amount) / Decimal(total_sat_balance)
invoice_fiat_amount = abs(total_fiat_balance) * proportion invoice_fiat_amount = abs(total_fiat_balance) * proportion
# Calculate fiat rate (sats per fiat unit)
fiat_rate = float(data.amount) / float(invoice_fiat_amount) if invoice_fiat_amount > 0 else 0
btc_rate = float(invoice_fiat_amount) / float(data.amount) * 100_000_000 if data.amount > 0 else 0
invoice_extra.update({ invoice_extra.update({
"fiat_currency": fiat_currency, "fiat_currency": fiat_currency,
"fiat_amount": str(invoice_fiat_amount.quantize(Decimal("0.001"))), "fiat_amount": str(invoice_fiat_amount.quantize(Decimal("0.001"))),
"fiat_rate": fiat_rate, **fiat_rate_metadata(data.amount, invoice_fiat_amount),
"btc_rate": btc_rate,
}) })
logger.info(f"Invoice extra metadata: {invoice_extra}") logger.info(f"Invoice extra metadata: {invoice_extra}")
@ -1850,13 +1843,14 @@ async def api_record_payment(
try: try:
async with httpx.AsyncClient(timeout=5.0) as client: 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( response = await client.get(
f"{fava.base_url}/api/journal", f"{fava.base_url}/journal",
params={"time": ""} # Get all entries params={"time": ""} # Get all entries
) )
response.raise_for_status()
if response.status_code == 200:
response_data = response.json() response_data = response.json()
entries = response_data.get('entries', []) entries = response_data.get('entries', [])
@ -1871,11 +1865,42 @@ async def api_record_payment(
"new_balance": balance_data["balance"], "new_balance": balance_data["balance"],
"message": "Payment already recorded", "message": "Payment already recorded",
} }
except Exception as e: 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}") logger.warning(f"Could not check Fava for duplicate payment: {e}")
# Continue anyway - Fava/Beancount will catch duplicate if it exists 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",
)
# Convert amount from millisatoshis to satoshis # Convert amount from millisatoshis to satoshis
try:
amount_sats = payment.amount // 1000 amount_sats = payment.amount // 1000
# Extract fiat metadata from invoice (if present) # Extract fiat metadata from invoice (if present)
@ -1929,11 +1954,19 @@ async def api_record_payment(
result = await fava.add_entry(entry) result = await fava.add_entry(entry)
logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}") 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 # Get updated balance from Fava
balance_data = await fava.get_user_balance_bql(target_user_id) balance_data = await fava.get_user_balance_bql(target_user_id)
return { return {
"journal_entry_id": f"fava-{datetime.now().timestamp()}", "journal_entry_id": entry_id,
"new_balance": balance_data["balance"], "new_balance": balance_data["balance"],
"message": "Payment recorded successfully", "message": "Payment recorded successfully",
} }
@ -2024,6 +2057,19 @@ async def api_settle_receivable(
unsettled_payables = await fava.get_unsettled_entries_bql(data.user_id, "expense") unsettled_payables = await fava.get_unsettled_entries_bql(data.user_id, "expense")
unsettled_receivables = await fava.get_unsettled_entries_bql(data.user_id, "receivable") unsettled_receivables = await fava.get_unsettled_entries_bql(data.user_id, "receivable")
# Net only entries denominated in the settlement currency — summing
# mixed currencies as if they were one silently mis-states the net
# obligation and links entries this settlement doesn't actually clear.
settle_currency = data.currency.upper()
unsettled_payables = [
e for e in unsettled_payables
if e.get("fiat_currency") == settle_currency
]
unsettled_receivables = [
e for e in unsettled_receivables
if e.get("fiat_currency") == settle_currency
]
payable_total = sum( payable_total = sum(
(Decimal(str(e["fiat_amount"])) for e in unsettled_payables), (Decimal(str(e["fiat_amount"])) for e in unsettled_payables),
Decimal(0), Decimal(0),
@ -2067,10 +2113,10 @@ async def api_settle_receivable(
"net to clear all open entries, or pass " "net to clear all open entries, or pass "
"`settled_entry_links` to settle a specific subset." "`settled_entry_links` to settle a specific subset."
), ),
"cash_paid": float(cash_paid), "cash_paid": str(cash_paid),
"net_obligation": float(net_obligation), "net_obligation": str(net_obligation),
"receivable_total": float(receivable_total), "receivable_total": str(receivable_total),
"payable_total": float(payable_total), "payable_total": str(payable_total),
"currency": data.currency.upper(), "currency": data.currency.upper(),
}, },
) )