diff --git a/__init__.py b/__init__.py index 614a8ba..42a2648 100644 --- a/__init__.py +++ b/__init__.py @@ -30,6 +30,17 @@ def libra_stop(): except Exception as ex: logger.warning(ex) + # Close the Fava client's shared HTTP connection pool. libra_stop is + # synchronous, so schedule the close; if the loop is already gone the + # sockets die with the process anyway. + from .fava_client import _fava_client + + if _fava_client is not None: + try: + asyncio.get_event_loop().create_task(_fava_client.aclose()) + except Exception as ex: + logger.warning(f"Could not close Fava HTTP client: {ex}") + def libra_start(): """Initialize Libra extension background tasks""" diff --git a/account_sync.py b/account_sync.py index 3d82381..870d1d7 100644 --- a/account_sync.py +++ b/account_sync.py @@ -285,7 +285,11 @@ async def sync_accounts_from_beancount(force_full_sync: bool = False) -> dict: return stats -async def sync_single_account_from_beancount(account_name: str) -> bool: +async def sync_single_account_from_beancount( + account_name: str, + description: Optional[str] = None, + assume_exists: bool = False, +) -> bool: """ Sync a single account from Beancount to Libra DB. @@ -294,6 +298,13 @@ async def sync_single_account_from_beancount(account_name: str) -> bool: Args: account_name: Hierarchical account name (e.g., "Expenses:Food") + description: Description for the Libra DB row (only used with + assume_exists — otherwise read from Beancount metadata) + assume_exists: Skip the Fava existence lookup. Pass when the + caller just wrote the Open directive itself — verifying via + a second serialized get_all_accounts round-trip doubles the + latency of every account create for no information gain + (libra-#53). Returns: True if account was created/updated, False if it already existed or failed @@ -306,6 +317,22 @@ async def sync_single_account_from_beancount(account_name: str) -> bool: logger.debug(f"Account already exists: {account_name}") return False + if assume_exists: + try: + await create_account( + CreateAccount( + name=account_name, + account_type=infer_account_type_from_name(account_name), + description=description, + user_id=extract_user_id_from_account_name(account_name), + ) + ) + logger.info(f"Created account (writer-asserted): {account_name}") + return True + except Exception as e: + logger.error(f"Failed to sync account {account_name}: {e}") + return False + # Get from Beancount fava = get_fava_client() try: diff --git a/beancount_format.py b/beancount_format.py index f233ee5..fbdfba9 100644 --- a/beancount_format.py +++ b/beancount_format.py @@ -115,13 +115,18 @@ def format_balance( account: str, amount: int, currency: str = "SATS" -) -> str: +) -> Dict[str, Any]: """ - Format a balance assertion directive for Beancount. + Format a balance assertion directive for Fava's JSON API. Balance assertions verify that an account has an expected balance on a specific date. They are checked automatically by Beancount when the file is loaded. + Fava's `deserialise` (fava/serialisation.py) expects + `{"t": "Balance", "amount": {"number", "currency"}, ...}` — the + previous source-string return 500'd on every assertion create + (libra-#39). + Args: date_val: Date of the balance assertion account: Account name (e.g., "Assets:Bitcoin:Lightning") @@ -129,15 +134,15 @@ def format_balance( currency: Currency code (default: "SATS") Returns: - Beancount balance directive as a string - - Example: - >>> format_balance(date(2025, 11, 10), "Assets:Bitcoin:Lightning", 1500000, "SATS") - '2025-11-10 balance Assets:Bitcoin:Lightning 1500000 SATS' + Fava API Balance entry dict ready for `fava.add_entry`. """ - date_str = date_val.strftime('%Y-%m-%d') - # Two spaces between account and amount (Beancount convention) - return f"{date_str} balance {account} {amount} {currency}" + return { + "t": "Balance", + "date": date_val.strftime('%Y-%m-%d'), + "account": account, + "amount": {"number": str(amount), "currency": currency}, + "meta": {}, + } def format_posting_with_cost( @@ -252,9 +257,10 @@ def format_posting_at_average_cost( 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}'") else: - # No cost - amount_str = f"{amount_sats} SATS {{}}" - logger.warning(f"format_posting_at_average_cost: cost_currency is None, using empty cost basis") + # No cost basis — omit the braces entirely. Empty "{}" is not + # valid Beancount syntax and fails to parse on ledger load. + 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 {} @@ -304,6 +310,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": "", "btc_rate": ""} + """ + 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( user_id: str, expense_account: str, @@ -718,15 +743,18 @@ def format_net_settlement_entry( entry_date: date, payment_hash: 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]: """ 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 2. Clear receivables in EUR 3. Clear payables in EUR + 4. Credit overflow when the payment exceeds what it clears Example: Assets:Bitcoin:Lightning 565251 SATS @@ 517.00 EUR @@ -734,25 +762,61 @@ def format_net_settlement_entry( Liabilities:Payable:User 38.00 EUR = 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: user_id: User ID payment_account: Payment account (e.g., "Assets:Bitcoin:Lightning") receivable_account: User's receivable account payable_account: User's payable account amount_sats: SATS amount paid - net_fiat_amount: Net fiat amount (receivable - payable) - total_receivable_fiat: Total receivables to clear - total_payable_fiat: Total payables to clear + net_fiat_amount: Fiat value of the payment being recorded + total_receivable_fiat: Receivables cleared by this payment + total_payable_fiat: Payables cleared by this payment fiat_currency: Currency (EUR, USD) description: Payment description entry_date: Date of payment payment_hash: Lightning payment hash reference: Optional reference 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: 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 # 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. @@ -761,20 +825,26 @@ def format_net_settlement_entry( postings = [ { "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 {} }, { "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))} }, { "account": payable_account, - "amount": f"{abs(total_payable_fiat):.2f} {fiat_currency}", + "amount": f"{total_payable_fiat:.2f} {fiat_currency}", "meta": {} } ] + if credit_overflow_fiat > 0: + postings.append({ + "account": credit_account, + "amount": f"-{credit_overflow_fiat:.2f} {fiat_currency}", + "meta": {} + }) entry_meta = { "user-id": user_id, diff --git a/core/validation.py b/core/validation.py index c29f069..913bb8a 100644 --- a/core/validation.py +++ b/core/validation.py @@ -5,7 +5,7 @@ Comprehensive validation following Beancount's plugin system approach, 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 @@ -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: try: Decimal(str(metadata["fiat_amount"])) - except (ValueError, TypeError) as e: + except (ValueError, TypeError, InvalidOperation) as e: raise ValidationError( f"Invalid fiat_amount: {metadata['fiat_amount']}", {"error": str(e)} diff --git a/crud.py b/crud.py index 0692806..b49bb9b 100644 --- a/crud.py +++ b/crud.py @@ -1696,3 +1696,73 @@ async def check_user_has_role_permission( return True return False + + +# ============================================================================= +# PROCESSED PAYMENTS (Lightning payment idempotency gate) +# ============================================================================= +# The Fava-side duplicate checks are read-then-write races; this table's +# primary key on payment_hash makes exactly one claimant win. Shared by the +# background invoice listener (tasks.on_invoice_paid) and the client-driven +# /record-payment endpoint. + + +async def claim_payment(payment_hash: str) -> bool: + """Atomically claim a Lightning payment for recording. + + Returns True when this caller owns the claim; False when the payment + is already recorded or another coroutine is recording it right now. + """ + result = await db.execute( + """ + INSERT INTO processed_payments (payment_hash, status) + VALUES (:payment_hash, 'processing') + ON CONFLICT (payment_hash) DO NOTHING + """, + {"payment_hash": payment_hash}, + ) + return result.rowcount == 1 + + +async def get_processed_payment(payment_hash: str) -> Optional[dict]: + row = await db.fetchone( + "SELECT payment_hash, status, entry_id FROM processed_payments" + " WHERE payment_hash = :payment_hash", + {"payment_hash": payment_hash}, + ) + return dict(row) if row else None + + +async def mark_payment_done(payment_hash: str, entry_id: Optional[str] = None) -> None: + await db.execute( + """ + UPDATE processed_payments SET status = 'done', entry_id = :entry_id + WHERE payment_hash = :payment_hash + """, + {"payment_hash": payment_hash, "entry_id": entry_id}, + ) + + +async def release_payment_claim(payment_hash: str) -> None: + """Compensating delete after a failed recording, so redelivery retries. + + Only removes an in-flight claim — a 'done' row is permanent. + """ + await db.execute( + "DELETE FROM processed_payments" + " WHERE payment_hash = :payment_hash AND status = 'processing'", + {"payment_hash": payment_hash}, + ) + + +async def clear_stale_payment_claims() -> int: + """Drop 'processing' claims left behind by a previous process life. + + A live claim only exists inside a running coroutine, so anything + still 'processing' at listener startup belongs to a crashed or + restarted process and would otherwise block that payment forever. + """ + result = await db.execute( + "DELETE FROM processed_payments WHERE status = 'processing'" + ) + return result.rowcount diff --git a/fava_client.py b/fava_client.py index eaed06b..faad8f7 100644 --- a/fava_client.py +++ b/fava_client.py @@ -20,7 +20,8 @@ See: https://github.com/beancount/fava/blob/main/src/fava/json_api.py import asyncio import re import httpx -from typing import Any, Dict, List, Optional +from contextlib import asynccontextmanager +from typing import Any, AsyncIterator, Callable, Dict, List, Optional from decimal import Decimal from datetime import date, datetime from loguru import logger @@ -44,6 +45,34 @@ def _infer_target_file(account_name: str) -> str: return "accounts/chart.beancount" +# Posting amount-string patterns shared by the balance parsers. Fava's +# @@ → @ normalisation can emit decimal SATS values, so every SATS group +# must tolerate decimals (the old integer-only pattern silently dropped +# those postings from balances). +_TOTAL_PRICE_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?[\d.]+)\s+SATS$') +_UNIT_PRICE_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$') +_FIAT_AMOUNT_RE = re.compile(r'^(-?[\d.]+)\s+([A-Z]{3})$') +_SATS_AMOUNT_RE = re.compile(r'^(-?[\d.]+)\s+SATS') + + +def _sats_to_int(value: str) -> int: + """Parse a (possibly decimal) SATS amount string to whole sats.""" + return int(Decimal(value)) + + +# Account names/patterns are interpolated into BQL string literals; restrict +# them to Beancount account characters so caller-supplied input can't break +# out of the quoted literal. +_BQL_ACCOUNT_RE = re.compile(r'^[A-Za-z0-9:_-]+$') + + +def _validate_bql_account(value: str) -> str: + """Validate a value bound for interpolation into a BQL string literal.""" + if not _BQL_ACCOUNT_RE.match(value): + raise ValueError(f"Invalid account name for BQL query: {value!r}") + return value + + def _escape_beancount_string(value: str) -> str: """Escape a value for safe inclusion in a Beancount string literal. @@ -136,6 +165,27 @@ class FavaClient: self._main_dir_cache: Optional[str] = None self._main_dir_lock = asyncio.Lock() + # Shared HTTP client, created lazily on first use. One client + # means one connection pool instead of a TCP handshake per call. + self._http: Optional[httpx.AsyncClient] = None + + @asynccontextmanager + async def _client(self) -> AsyncIterator[httpx.AsyncClient]: + """Yield the shared HTTP client (lazily created). + + Kept as a context manager so call sites read the same as the + per-call clients they replace; the client itself is NOT closed on + exit — call `aclose()` at extension shutdown. + """ + if self._http is None or self._http.is_closed: + self._http = httpx.AsyncClient(timeout=self.timeout) + yield self._http + + async def aclose(self) -> None: + """Close the shared HTTP client (extension shutdown).""" + if self._http is not None and not self._http.is_closed: + await self._http.aclose() + async def _resolve_target_file(self, target_file: str) -> str: """ Turn a relative include path into the absolute path fava expects. @@ -160,7 +210,7 @@ class FavaClient: if self._main_dir_cache is None: async with self._main_dir_lock: if self._main_dir_cache is None: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: resp = await client.get(f"{self.base_url}/options") resp.raise_for_status() main_file = resp.json()["data"]["beancount_options"]["filename"] @@ -236,7 +286,7 @@ class FavaClient: # Acquire global write lock to serialize ledger modifications async with self._write_lock: try: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: response = await client.put( f"{self.base_url}/add_entries", json={"entries": [entry]}, @@ -350,10 +400,11 @@ class FavaClient: # Use sum(weight) for SATS and sum(number) for fiat # Note: BQL doesn't support != operator, so use flag = '*' to exclude pending + _validate_bql_account(account_name) query = f"SELECT sum(number), sum(weight) WHERE account = '{account_name}' AND flag = '*'" try: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: response = await client.get( f"{self.base_url}/query", params={"query_string": query} @@ -446,14 +497,14 @@ class FavaClient: import re # Try total price notation: "50.00 EUR @@ 50000 SATS" - total_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?\d+)\s+SATS$', amount_str) + total_price_match = _TOTAL_PRICE_RE.match(amount_str) # Try per-unit price notation: "50.00 EUR @ 1000.5 SATS" - unit_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$', amount_str) + unit_price_match = _UNIT_PRICE_RE.match(amount_str) if total_price_match: fiat_amount = Decimal(total_price_match.group(1)) fiat_currency = total_price_match.group(2) - sats_amount = int(total_price_match.group(3)) + sats_amount = _sats_to_int(total_price_match.group(3)) if fiat_currency not in fiat_balances: fiat_balances[fiat_currency] = Decimal(0) @@ -480,8 +531,8 @@ class FavaClient: accounts_dict[account_name]["sats"] += sats_amount # Try simple fiat format: "50.00 EUR" (check metadata for sats) - elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str): - fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str) + elif _FIAT_AMOUNT_RE.match(amount_str): + fiat_match = _FIAT_AMOUNT_RE.match(amount_str) if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'): fiat_amount = Decimal(fiat_match.group(1)) fiat_currency = fiat_match.group(2) @@ -502,9 +553,9 @@ class FavaClient: else: # Old format: SATS with cost/price notation - extract SATS amount - sats_match = re.match(r'^(-?\d+)\s+SATS', amount_str) + sats_match = _SATS_AMOUNT_RE.match(amount_str) if sats_match: - sats_amount = int(sats_match.group(1)) + sats_amount = _sats_to_int(sats_match.group(1)) total_sats += sats_amount # Track per account @@ -603,14 +654,14 @@ class FavaClient: import re # Try total price notation: "50.00 EUR @@ 50000 SATS" - total_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?\d+)\s+SATS$', amount_str) + total_price_match = _TOTAL_PRICE_RE.match(amount_str) # Try per-unit price notation: "50.00 EUR @ 1000.5 SATS" - unit_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$', amount_str) + unit_price_match = _UNIT_PRICE_RE.match(amount_str) if total_price_match: fiat_amount = Decimal(total_price_match.group(1)) fiat_currency = total_price_match.group(2) - sats_amount = int(total_price_match.group(3)) + sats_amount = _sats_to_int(total_price_match.group(3)) if fiat_currency not in user_data[user_id]["fiat_balances"]: user_data[user_id]["fiat_balances"][fiat_currency] = Decimal(0) @@ -629,8 +680,8 @@ class FavaClient: user_data[user_id]["balance"] += sats_amount # Try simple fiat format: "50.00 EUR" (check metadata for sats) - elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str): - fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str) + elif _FIAT_AMOUNT_RE.match(amount_str): + fiat_match = _FIAT_AMOUNT_RE.match(amount_str) if fiat_match and fiat_match.group(2) in ('EUR', 'USD', 'GBP'): fiat_amount = Decimal(fiat_match.group(1)) fiat_currency = fiat_match.group(2) @@ -648,9 +699,9 @@ class FavaClient: else: # Old format: SATS with cost/price notation - sats_match = re.match(r'^(-?\d+)\s+SATS', amount_str) + sats_match = _SATS_AMOUNT_RE.match(amount_str) if sats_match: - sats_amount = int(sats_match.group(1)) + sats_amount = _sats_to_int(sats_match.group(1)) user_data[user_id]["balance"] += sats_amount # Extract fiat from cost syntax or metadata (backward compatibility) @@ -683,9 +734,12 @@ class FavaClient: True if Fava responds, False otherwise """ try: - async with httpx.AsyncClient(timeout=2.0) as client: + async with self._client() as client: + # Health probes stay fast regardless of the configured + # request timeout. response = await client.get( - f"{self.base_url}/changed" + f"{self.base_url}/changed", + timeout=2.0 ) return response.status_code == 200 except Exception as e: @@ -721,12 +775,13 @@ class FavaClient: """ # Build Beancount query if account_pattern: + _validate_bql_account(account_pattern) query = f"SELECT * WHERE account ~ '{account_pattern}' ORDER BY date DESC LIMIT {limit}" else: query = f"SELECT * ORDER BY date DESC LIMIT {limit}" try: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: response = await client.get( f"{self.base_url}/query", params={"query_string": query} @@ -807,7 +862,7 @@ class FavaClient: https://beancount.github.io/docs/beancount_query_language.html """ try: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: response = await client.get( f"{self.base_url}/query", params={"query_string": query_string} @@ -1341,7 +1396,7 @@ class FavaClient: # (BQL's SELECT DISTINCT account only returns accounts with postings) account_names: set[str] = set() - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: for endpoint in ("balance_sheet", "income_statement"): try: response = await client.get(f"{self.base_url}/{endpoint}") @@ -1440,7 +1495,7 @@ class FavaClient: params["time"] = f"{cutoff_date.isoformat()} - {today.isoformat()}" logger.info(f"Querying journal for last {days} days (from {cutoff_date})") - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: response = await client.get(f"{self.base_url}/journal", params=params) response.raise_for_status() result = response.json() @@ -1482,7 +1537,7 @@ class FavaClient: sha256sum = context["sha256sum"] """ try: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: response = await client.get( f"{self.base_url}/context", params={"entry_hash": entry_hash} @@ -1529,7 +1584,7 @@ class FavaClient: # Acquire global write lock to serialize ledger modifications async with self._write_lock: try: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: response = await client.put( f"{self.base_url}/source_slice", json={ @@ -1544,11 +1599,78 @@ class FavaClient: except httpx.HTTPStatusError as e: logger.error(f"Fava update error: {e.response.status_code} - {e.response.text}") + if e.response.status_code in (409, 412): + raise ChecksumConflictError( + f"Entry {entry_hash} changed concurrently" + ) from e raise except httpx.RequestError as e: logger.error(f"Fava connection error: {e}") raise + async def transform_source_line( + self, + filename: str, + lineno: int, + transform: Callable[[str], str], + ) -> bool: + """Atomically read-modify-write one line of a ledger source file. + + Holds the global write lock across the whole read-modify-write, so + another writer can't slip in between the checksum read and the + write (libra-#23: the approve/reject endpoints used to do this + read-then-write with raw httpx and no lock). + + The transform receives the current line and returns the new one; + returning it unchanged skips the write. + + Returns: + True when the line was changed and written, False on a no-op. + + Raises: + ValueError: lineno is outside the file. + ChecksumConflictError: an out-of-process writer changed the + file between read and write (409/412 from Fava). + """ + async with self._write_lock: + async with self._client() as client: + response = await client.get( + f"{self.base_url}/source", + params={"filename": filename}, + ) + response.raise_for_status() + source_data = response.json()["data"] + sha256sum = source_data["sha256sum"] + lines = source_data["source"].split("\n") + + idx = lineno - 1 + if idx < 0 or idx >= len(lines): + raise ValueError(f"Line {lineno} not found in {filename}") + + new_line = transform(lines[idx]) + if new_line == lines[idx]: + return False + lines[idx] = new_line + + try: + update = await client.put( + f"{self.base_url}/source", + json={ + "file_path": filename, + "source": "\n".join(lines), + "sha256sum": sha256sum, + }, + headers={"Content-Type": "application/json"}, + ) + update.raise_for_status() + except httpx.HTTPStatusError as e: + if e.response.status_code in (409, 412): + raise ChecksumConflictError( + f"{filename} changed concurrently" + ) from e + raise + return True + async def delete_entry(self, entry_hash: str, sha256sum: str) -> str: """ Delete an entry from the Beancount file. @@ -1571,7 +1693,7 @@ class FavaClient: # Acquire global write lock to serialize ledger modifications async with self._write_lock: try: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: response = await client.delete( f"{self.base_url}/source_slice", params={ @@ -1585,6 +1707,10 @@ class FavaClient: except httpx.HTTPStatusError as e: logger.error(f"Fava delete error: {e.response.status_code} - {e.response.text}") + if e.response.status_code in (409, 412): + raise ChecksumConflictError( + f"Entry {entry_hash} changed concurrently" + ) from e raise except httpx.RequestError as e: logger.error(f"Fava connection error: {e}") @@ -1664,7 +1790,7 @@ class FavaClient: # Acquire global write lock to serialize ledger modifications async with self._write_lock: try: - async with httpx.AsyncClient(timeout=self.timeout) as client: + async with self._client() as client: # Step 1: Get current source file (fresh read on each attempt) response = await client.get( f"{self.base_url}/source", @@ -1817,7 +1943,7 @@ class FavaClient: # Query 1: Get all original expense/receivable entries for this user # These are entries with the expense-entry or receivable-entry tag 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 WHERE account ~ '{account_pattern}' AND '{entry_tag}' IN tags @@ -1851,7 +1977,10 @@ class FavaClient: entries_by_link: Dict[str, Dict[str, Any]] = {} 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 if not links or not isinstance(links, list): @@ -1875,9 +2004,11 @@ class FavaClient: if entry_link in entries_by_link: continue - # Parse amounts - fiat_amount = abs(float(number)) if number else 0.0 - fiat_currency = "EUR" # Default, could be extracted from posting + # Parse amounts. The posting's real currency matters: callers + # net these totals per currency, and the old hardcoded "EUR" + # 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 sats_amount = 0 diff --git a/migrations.py b/migrations.py index 9c38c55..d8a4bba 100644 --- a/migrations.py +++ b/migrations.py @@ -34,9 +34,33 @@ Original migration sequence (Nov 2025): - m014: Removed legacy equity accounts (MemberEquity, RetainedEarnings) - m015: Converted entry_lines to single amount field - m016: Dropped journal_entries and entry_lines tables (Fava integration) + +IDEMPOTENCY CONTRACT: +Every statement here must be a silent no-op on re-run. The migration +version bump lands in the core LNbits DB (`dbversions`) while the DDL +lands in `ext_libra` — the two writes are not atomic. If the version +bump fails after the DDL commits, the whole migration re-runs on next +boot; a bare CREATE/ALTER/INSERT then crashes the extension until +manual dbversions surgery. """ +async def _alter_add_column_safe(db, sql: str) -> None: + """ALTER TABLE ADD COLUMN that swallows duplicate-column errors. + + Neither SQLite nor Postgres supports ADD COLUMN IF NOT EXISTS + portably, so re-runs are made no-ops by swallowing the error both + backends raise for an existing column. + """ + try: + await db.execute(sql) + except Exception as exc: + msg = str(exc).lower() + if "duplicate column" in msg or "already exists" in msg: + return + raise + + async def m001_initial(db): """ Initial Libra database schema (squashed from m001-m016). @@ -63,7 +87,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE accounts ( + CREATE TABLE IF NOT EXISTS accounts ( id TEXT PRIMARY KEY, name TEXT NOT NULL UNIQUE, account_type TEXT NOT NULL, @@ -76,13 +100,13 @@ async def m001_initial(db): await db.execute( """ - CREATE INDEX idx_accounts_user_id ON accounts (user_id); + CREATE INDEX IF NOT EXISTS idx_accounts_user_id ON accounts (user_id); """ ) await db.execute( """ - CREATE INDEX idx_accounts_type ON accounts (account_type); + CREATE INDEX IF NOT EXISTS idx_accounts_type ON accounts (account_type); """ ) @@ -93,7 +117,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE extension_settings ( + CREATE TABLE IF NOT EXISTS extension_settings ( id TEXT NOT NULL PRIMARY KEY, libra_wallet_id TEXT, fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333', @@ -111,7 +135,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE user_wallet_settings ( + CREATE TABLE IF NOT EXISTS user_wallet_settings ( id TEXT NOT NULL PRIMARY KEY, user_wallet_id TEXT, updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now} @@ -126,7 +150,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE manual_payment_requests ( + CREATE TABLE IF NOT EXISTS manual_payment_requests ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, amount INTEGER NOT NULL, @@ -143,14 +167,14 @@ async def m001_initial(db): await db.execute( """ - CREATE INDEX idx_manual_payment_requests_user_id + CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_user_id ON manual_payment_requests (user_id); """ ) await db.execute( """ - CREATE INDEX idx_manual_payment_requests_status + CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_status ON manual_payment_requests (status); """ ) @@ -163,7 +187,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE balance_assertions ( + CREATE TABLE IF NOT EXISTS balance_assertions ( id TEXT PRIMARY KEY, date TIMESTAMP NOT NULL, account_id TEXT NOT NULL, @@ -188,21 +212,21 @@ async def m001_initial(db): await db.execute( """ - CREATE INDEX idx_balance_assertions_account_id + CREATE INDEX IF NOT EXISTS idx_balance_assertions_account_id ON balance_assertions (account_id); """ ) await db.execute( """ - CREATE INDEX idx_balance_assertions_status + CREATE INDEX IF NOT EXISTS idx_balance_assertions_status ON balance_assertions (status); """ ) await db.execute( """ - CREATE INDEX idx_balance_assertions_date + CREATE INDEX IF NOT EXISTS idx_balance_assertions_date ON balance_assertions (date); """ ) @@ -216,7 +240,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE user_equity_status ( + CREATE TABLE IF NOT EXISTS user_equity_status ( user_id TEXT PRIMARY KEY, is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE, equity_account_name TEXT, @@ -230,7 +254,7 @@ async def m001_initial(db): await db.execute( """ - CREATE INDEX idx_user_equity_status_eligible + CREATE INDEX IF NOT EXISTS idx_user_equity_status_eligible ON user_equity_status (is_equity_eligible) WHERE is_equity_eligible = TRUE; """ @@ -245,7 +269,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE account_permissions ( + CREATE TABLE IF NOT EXISTS account_permissions ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, account_id TEXT NOT NULL, @@ -262,7 +286,7 @@ async def m001_initial(db): # Index for looking up permissions by user await db.execute( """ - CREATE INDEX idx_account_permissions_user_id + CREATE INDEX IF NOT EXISTS idx_account_permissions_user_id ON account_permissions (user_id); """ ) @@ -270,7 +294,7 @@ async def m001_initial(db): # Index for looking up permissions by account await db.execute( """ - CREATE INDEX idx_account_permissions_account_id + CREATE INDEX IF NOT EXISTS idx_account_permissions_account_id ON account_permissions (account_id); """ ) @@ -278,7 +302,7 @@ async def m001_initial(db): # Composite index for checking specific user+account permissions await db.execute( """ - CREATE INDEX idx_account_permissions_user_account + CREATE INDEX IF NOT EXISTS idx_account_permissions_user_account ON account_permissions (user_id, account_id); """ ) @@ -286,7 +310,7 @@ async def m001_initial(db): # Index for finding permissions by type await db.execute( """ - CREATE INDEX idx_account_permissions_type + CREATE INDEX IF NOT EXISTS idx_account_permissions_type ON account_permissions (permission_type); """ ) @@ -294,7 +318,7 @@ async def m001_initial(db): # Index for finding expired permissions await db.execute( """ - CREATE INDEX idx_account_permissions_expires + CREATE INDEX IF NOT EXISTS idx_account_permissions_expires ON account_permissions (expires_at) WHERE expires_at IS NOT NULL; """ @@ -320,6 +344,7 @@ async def m001_initial(db): f""" INSERT INTO accounts (id, name, account_type, description, created_at) VALUES (:id, :name, :type, :description, {db.timestamp_now}) + ON CONFLICT (name) DO NOTHING """, { "id": str(uuid.uuid4()), @@ -342,17 +367,18 @@ async def m002_add_account_is_active(db): Default: All existing accounts are marked as active (TRUE). """ - await db.execute( + await _alter_add_column_safe( + db, """ ALTER TABLE accounts ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT TRUE - """ + """, ) # Create index for faster queries filtering by is_active await db.execute( """ - CREATE INDEX idx_accounts_is_active ON accounts (is_active) + CREATE INDEX IF NOT EXISTS idx_accounts_is_active ON accounts (is_active) """ ) @@ -372,17 +398,18 @@ async def m003_add_account_is_virtual(db): Default: All existing accounts are real (is_virtual = FALSE). """ - await db.execute( + await _alter_add_column_safe( + db, """ ALTER TABLE accounts ADD COLUMN is_virtual BOOLEAN NOT NULL DEFAULT FALSE - """ + """, ) # Create index for faster queries filtering by is_virtual await db.execute( """ - CREATE INDEX idx_accounts_is_virtual ON accounts (is_virtual) + CREATE INDEX IF NOT EXISTS idx_accounts_is_virtual ON accounts (is_virtual) """ ) @@ -402,6 +429,7 @@ async def m003_add_account_is_virtual(db): f""" INSERT INTO accounts (id, name, account_type, description, is_active, is_virtual, created_at) VALUES (:id, :name, :type, :description, TRUE, TRUE, {db.timestamp_now}) + ON CONFLICT (name) DO NOTHING """, { "id": str(uuid.uuid4()), @@ -438,7 +466,7 @@ async def m004_add_rbac_tables(db): await db.execute( f""" - CREATE TABLE roles ( + CREATE TABLE IF NOT EXISTS roles ( id TEXT PRIMARY KEY, name TEXT NOT NULL UNIQUE, description TEXT, @@ -451,13 +479,13 @@ async def m004_add_rbac_tables(db): await db.execute( """ - CREATE INDEX idx_roles_name ON roles (name); + CREATE INDEX IF NOT EXISTS idx_roles_name ON roles (name); """ ) await db.execute( """ - CREATE INDEX idx_roles_is_default ON roles (is_default) + CREATE INDEX IF NOT EXISTS idx_roles_is_default ON roles (is_default) WHERE is_default = TRUE; """ ) @@ -469,7 +497,7 @@ async def m004_add_rbac_tables(db): await db.execute( f""" - CREATE TABLE role_permissions ( + CREATE TABLE IF NOT EXISTS role_permissions ( id TEXT PRIMARY KEY, role_id TEXT NOT NULL, account_id TEXT NOT NULL, @@ -484,19 +512,19 @@ async def m004_add_rbac_tables(db): await db.execute( """ - CREATE INDEX idx_role_permissions_role_id ON role_permissions (role_id); + CREATE INDEX IF NOT EXISTS idx_role_permissions_role_id ON role_permissions (role_id); """ ) await db.execute( """ - CREATE INDEX idx_role_permissions_account_id ON role_permissions (account_id); + CREATE INDEX IF NOT EXISTS idx_role_permissions_account_id ON role_permissions (account_id); """ ) await db.execute( """ - CREATE INDEX idx_role_permissions_type ON role_permissions (permission_type); + CREATE INDEX IF NOT EXISTS idx_role_permissions_type ON role_permissions (permission_type); """ ) @@ -507,7 +535,7 @@ async def m004_add_rbac_tables(db): await db.execute( f""" - CREATE TABLE user_roles ( + CREATE TABLE IF NOT EXISTS user_roles ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, role_id TEXT NOT NULL, @@ -522,19 +550,19 @@ async def m004_add_rbac_tables(db): await db.execute( """ - CREATE INDEX idx_user_roles_user_id ON user_roles (user_id); + CREATE INDEX IF NOT EXISTS idx_user_roles_user_id ON user_roles (user_id); """ ) await db.execute( """ - CREATE INDEX idx_user_roles_role_id ON user_roles (role_id); + CREATE INDEX IF NOT EXISTS idx_user_roles_role_id ON user_roles (role_id); """ ) await db.execute( """ - CREATE INDEX idx_user_roles_expires ON user_roles (expires_at) + CREATE INDEX IF NOT EXISTS idx_user_roles_expires ON user_roles (expires_at) WHERE expires_at IS NOT NULL; """ ) @@ -542,7 +570,7 @@ async def m004_add_rbac_tables(db): # Composite index for checking specific user+role assignments await db.execute( """ - CREATE INDEX idx_user_roles_user_role ON user_roles (user_id, role_id); + CREATE INDEX IF NOT EXISTS idx_user_roles_user_role ON user_roles (user_id, role_id); """ ) @@ -586,6 +614,7 @@ async def m004_add_rbac_tables(db): f""" INSERT INTO roles (id, name, description, is_default, created_by, created_at) VALUES (:id, :name, :description, :is_default, :created_by, {db.timestamp_now}) + ON CONFLICT (name) DO NOTHING """, { "id": str(uuid.uuid4()), @@ -595,3 +624,30 @@ async def m004_add_rbac_tables(db): "created_by": "system", # System-created default roles }, ) + + +async def m005_add_processed_payments(db): + """ + Local idempotency gate for Lightning payment recording. + + The Fava-side duplicate check (`add_entry_idempotent`, journal-link + scan) is a read-then-write race: the background invoice listener and + the client-driven /record-payment endpoint can both pass the "not + present" check for the same payment_hash and both insert. The + primary key on payment_hash makes exactly one claimant win. + + status lifecycle: 'processing' (claimed, write in flight) → 'done' + (entry recorded). Failed claims are deleted so redelivery retries; + 'processing' rows from a crashed process are cleared at listener + startup. + """ + await db.execute( + f""" + CREATE TABLE IF NOT EXISTS processed_payments ( + payment_hash TEXT PRIMARY KEY, + status TEXT NOT NULL DEFAULT 'processing', + entry_id TEXT, + created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now} + ); + """ + ) diff --git a/tasks.py b/tasks.py index 8ed5a33..23dd0b5 100644 --- a/tasks.py +++ b/tasks.py @@ -179,12 +179,31 @@ async def wait_for_paid_invoices(): This ensures payments are recorded even if the user closes their browser before the payment is detected by client-side polling. """ + from .crud import clear_stale_payment_claims + invoice_queue = Queue() register_invoice_listener(invoice_queue, "ext_libra") + # Claims from a previous process life can't be live anymore — clear + # them so those payments aren't blocked forever. + cleared = await clear_stale_payment_claims() + if cleared: + logger.warning( + f"[LIBRA] Cleared {cleared} stale in-flight payment claim(s) " + "from a previous run" + ) + while True: payment = await invoice_queue.get() - await on_invoice_paid(payment) + try: + await on_invoice_paid(payment) + except Exception: + # One bad payment must not kill the listener for the rest of + # the process lifetime; its claim was released, so redelivery + # can retry it. + logger.exception( + f"[LIBRA] Failed to record payment {payment.payment_hash}" + ) async def on_invoice_paid(payment: Payment) -> None: @@ -210,8 +229,20 @@ async def on_invoice_paid(payment: Payment) -> None: logger.warning(f"Libra invoice {payment.payment_hash} missing user_id in metadata") return + from .crud import claim_payment, mark_payment_done, release_payment_claim from .fava_client import get_fava_client + # Local idempotency gate: exactly one claimant (this listener or the + # /record-payment endpoint) gets to record a given payment_hash. The + # Fava-side idempotent write below stays as a second layer for + # entries recorded before this table existed. + if not await claim_payment(payment.payment_hash): + logger.info( + f"Payment {payment.payment_hash} already recorded or being " + "recorded; skipping" + ) + return + fava = get_fava_client() # Use idempotency key based on payment hash - this ensures duplicate @@ -245,26 +276,36 @@ 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 + # Get user's current balance to determine what this payment clears balance = await fava.get_user_balance(user_id) fiat_balances = balance.get("fiat_balances", {}) total_fiat_balance = fiat_balances.get(fiat_currency, Decimal(0)) - # Determine receivables and payables based on balance - # Positive balance = user owes libra (receivable) - # Negative balance = libra owes user (payable) - if total_fiat_balance > 0: - # User owes libra - total_receivable = total_fiat_balance - total_payable = Decimal(0) - else: - # Libra owes user - total_receivable = Decimal(0) - total_payable = abs(total_fiat_balance) + # Settle only what this payment covers. The balance is already + # net (positive = user owes libra); a partial payment clears + # that much receivable, and any excess — or the whole payment + # when nothing is owed — becomes credit libra owes the user. + # (Previously partial payments cleared the FULL balance against + # 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) + credit_overflow = fiat_amount - total_receivable + if credit_overflow < tolerance: + # Absorb sub-cent rounding into the receivable leg. + 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 user_receivable = await get_or_create_user_account( @@ -273,22 +314,35 @@ async def on_invoice_paid(payment: Payment) -> None: user_payable = await get_or_create_user_account( 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") 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 - # Net settlement can settle both expenses and receivables + # Link the source entries this settlement reconciles — but only + # 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 = [] - try: - 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") - settled_links.extend([e["link"] for e in unsettled_receivables if e.get("link")]) - except Exception as e: - logger.warning(f"Could not query unsettled entries for settlement links: {e}") - # Continue without links - settlement will still be recorded + if open_receivable > 0 and fiat_amount + tolerance >= open_receivable: + try: + unsettled_expenses = await fava.get_unsettled_entries_bql(user_id, "expense") + unsettled_receivables = await fava.get_unsettled_entries_bql(user_id, "receivable") + 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: + logger.warning(f"Could not query unsettled entries for settlement links: {e}") + # Continue without links - settlement will still be recorded # Format as net settlement transaction entry = format_net_settlement_entry( @@ -305,7 +359,9 @@ async def on_invoice_paid(payment: Payment) -> None: entry_date=datetime.now().date(), payment_hash=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 @@ -324,6 +380,11 @@ async def on_invoice_paid(payment: Payment) -> None: f"{result.get('data', 'Unknown')}" ) + await mark_payment_done(payment.payment_hash, idempotency_key) + except Exception as e: logger.error(f"Error recording Libra payment {payment.payment_hash}: {e}") + # Release the claim so redelivery (or the /record-payment + # endpoint) can retry this payment. + await release_payment_claim(payment.payment_hash) raise diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..1586473 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,110 @@ +"""Migration idempotency tests. + +The migration version bump lands in the core LNbits DB (`dbversions`) +while the DDL lands in `ext_libra` — the two writes are not atomic. If +the bump fails after the DDL commits, the whole migration re-runs on +next boot, so every statement must be a silent no-op on re-run instead +of crashing with `duplicate column` / `table exists` / UNIQUE +violations (which bricks the extension until manual dbversions +surgery). + +These tests run the full migration chain twice against a fresh SQLite +database — the second pass simulates the lost-version-bump re-run. +""" +import importlib +import re +from uuid import uuid4 + +import pytest + +from lnbits.db import Database + +pytestmark = pytest.mark.anyio + + +def _module(name: str): + """Import a libra submodule under whichever path the active LNbits layout + uses (default `lnbits.extensions.libra` or bare `libra`).""" + for prefix in ("lnbits.extensions.libra", "libra"): + try: + return importlib.import_module(f"{prefix}.{name}") + except ModuleNotFoundError: + continue + raise ModuleNotFoundError(f"libra.{name}: tried both import paths") + + +migrations = _module("migrations") + +# Same discovery as lnbits.core.helpers.run_migration: m### prefix, +# module definition order. +_MIGRATION_RE = re.compile(r"^m(\d\d\d)_") +MIGRATION_FUNCTIONS = [ + fn for name, fn in vars(migrations).items() if _MIGRATION_RE.match(name) +] + +# Tables the chain must leave behind — one probe row read per table +# proves both existence and queryability after a double run. +EXPECTED_TABLES = [ + "accounts", + "extension_settings", + "user_wallet_settings", + "manual_payment_requests", + "balance_assertions", + "user_equity_status", + "account_permissions", + "roles", + "role_permissions", + "user_roles", + "processed_payments", +] + + +async def _run_all_migrations(db: Database) -> None: + async with db.connect() as conn: + for migrate in MIGRATION_FUNCTIONS: + await migrate(conn) + + +async def _seed_counts(db: Database) -> dict: + async with db.connect() as conn: + accounts = await conn.fetchall("SELECT id, name FROM accounts") + roles = await conn.fetchall("SELECT id, name FROM roles") + return { + "account_names": sorted(r["name"] for r in accounts), + "account_ids": sorted(r["id"] for r in accounts), + "role_names": sorted(r["name"] for r in roles), + "role_ids": sorted(r["id"] for r in roles), + } + + +async def test_migrations_rerun_is_noop(): + """Full chain twice: second run must not raise and not re-seed.""" + db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}") + + await _run_all_migrations(db) + first = await _seed_counts(db) + + # Simulate the lost dbversions bump: everything runs again. + await _run_all_migrations(db) + second = await _seed_counts(db) + + # Seeds must not duplicate (names) and must not be replaced (ids). + assert second == first + assert first["account_names"], "seed accounts missing after migration" + assert "Employee" in first["role_names"] + + # Every table exists and is queryable after the double run. + async with db.connect() as conn: + for table in EXPECTED_TABLES: + await conn.fetchall(f"SELECT * FROM {table} LIMIT 1") # noqa: S608 + + +async def test_single_migration_rerun_is_noop(): + """Each migration individually survives an immediate re-run (the + version bump fails right after that one migration committed).""" + db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}") + + async with db.connect() as conn: + for migrate in MIGRATION_FUNCTIONS: + await migrate(conn) + await migrate(conn) # re-run before "bumping" to the next diff --git a/tests/test_payment_idempotency.py b/tests/test_payment_idempotency.py new file mode 100644 index 0000000..969107b --- /dev/null +++ b/tests/test_payment_idempotency.py @@ -0,0 +1,281 @@ +"""Lightning payment idempotency — the `processed_payments` claim gate. + +The background invoice listener (`tasks.on_invoice_paid`) and the +client-driven `POST /record-payment` endpoint can both fire for the +same `payment_hash` (queue redelivery after restart, webhook + poller). +The Fava-side duplicate checks are read-then-write races; the local +`processed_payments` primary key makes exactly one claimant win. + +These tests bypass invoice generation (blocked by libra/issues/40) by +delivering synthetic paid `Payment` objects straight to +`on_invoice_paid` and by inserting paid payment rows via the LNbits +core crud for the endpoint tests. +""" +import asyncio +import importlib +from uuid import uuid4 + +import pytest + +from lnbits.core.crud.payments import create_payment +from lnbits.core.models.payments import CreatePayment, Payment, PaymentState + +from .helpers import list_user_entries, post_receivable + +pytestmark = pytest.mark.anyio + + +def _module(name: str): + """Import a libra submodule under whichever path the active LNbits layout + uses (default `lnbits.extensions.libra` or bare `libra`).""" + for prefix in ("lnbits.extensions.libra", "libra"): + try: + return importlib.import_module(f"{prefix}.{name}") + except ModuleNotFoundError: + continue + raise ModuleNotFoundError(f"libra.{name}: tried both import paths") + + +tasks = _module("tasks") +libra_crud = _module("crud") + + +def _paid_payment( + wallet_id: str, + user_id: str, + *, + fiat_amount: str = "100.00", + fiat_currency: str = "EUR", + sats: int = 100_000, +) -> Payment: + payment_hash = uuid4().hex + uuid4().hex[:32] + return Payment( + checking_id=payment_hash, + payment_hash=payment_hash, + wallet_id=wallet_id, + amount=sats * 1000, + fee=0, + bolt11="lnbcfake", + status=PaymentState.SUCCESS, + extra={ + "tag": "libra", + "user_id": user_id, + "fiat_currency": fiat_currency, + "fiat_amount": fiat_amount, + }, + ) + + +async def _setup_receivable( + client, super_user_headers, configured_user, standard_accounts, + amount: str = "100.00", +): + user, wallet = configured_user + await post_receivable( + client, + super_user_headers=super_user_headers, + user_id=user.id, + amount=amount, + description=f"Idempotency setup {uuid4().hex[:6]}", + revenue_account=standard_accounts["revenue_rent"]["name"], + ) + # Force a Fava reload before downstream balance reads (see #37). + await list_user_entries(client, wallet_inkey=wallet.inkey) + return user, wallet + + +async def _entries_with_link(client, wallet_inkey: str, link: str) -> list: + payload = await list_user_entries(client, wallet_inkey=wallet_inkey) + return [ + e for e in payload["entries"] if link in (e.get("links") or []) + ] + + +# --------------------------------------------------------------------------- +# on_invoice_paid — the background listener path +# --------------------------------------------------------------------------- + + +async def test_double_delivery_records_exactly_once( + client, super_user_headers, configured_user, standard_accounts +): + """Same payment delivered twice (queue redelivery) → one ledger entry.""" + user, wallet = await _setup_receivable( + client, super_user_headers, configured_user, standard_accounts + ) + payment = _paid_payment(wallet.id, user.id) + + await tasks.on_invoice_paid(payment) + await tasks.on_invoice_paid(payment) + + link = f"ln-{payment.payment_hash[:16]}" + assert len(await _entries_with_link(client, wallet.inkey, link)) == 1 + + row = await libra_crud.get_processed_payment(payment.payment_hash) + assert row is not None and row["status"] == "done" + + +async def test_failed_recording_releases_claim_and_retry_succeeds( + client, super_user_headers, configured_user, standard_accounts, monkeypatch +): + """A Fava failure mid-write must not permanently block the payment.""" + user, wallet = await _setup_receivable( + client, super_user_headers, configured_user, standard_accounts + ) + payment = _paid_payment(wallet.id, user.id) + + fava_client = _module("fava_client") + fava = fava_client.get_fava_client() + + async def _boom(*args, **kwargs): + raise RuntimeError("fava down") + + monkeypatch.setattr(fava, "add_entry_idempotent", _boom) + with pytest.raises(RuntimeError): + await tasks.on_invoice_paid(payment) + monkeypatch.undo() + + # Claim released → nothing recorded, retry allowed. + assert await libra_crud.get_processed_payment(payment.payment_hash) is None + + await tasks.on_invoice_paid(payment) + row = await libra_crud.get_processed_payment(payment.payment_hash) + assert row is not None and row["status"] == "done" + link = f"ln-{payment.payment_hash[:16]}" + assert len(await _entries_with_link(client, wallet.inkey, link)) == 1 + + +async def test_listener_survives_poison_payment_and_clears_stale_claims( + client, super_user_headers, configured_user, standard_accounts, monkeypatch +): + """One bad payment must not kill the listener; stale 'processing' + claims from a previous process life are cleared at startup.""" + user, wallet = await _setup_receivable( + client, super_user_headers, configured_user, standard_accounts + ) + + # A claim left behind by a "crashed" previous run. + stale_hash = uuid4().hex + uuid4().hex[:32] + assert await libra_crud.claim_payment(stale_hash) + + captured: dict = {} + monkeypatch.setattr( + tasks, + "register_invoice_listener", + lambda queue, name: captured.update(queue=queue), + ) + + listener = asyncio.create_task(tasks.wait_for_paid_invoices()) + try: + for _ in range(50): + if "queue" in captured: + break + await asyncio.sleep(0.05) + assert "queue" in captured, "listener never registered its queue" + + poison = _paid_payment(wallet.id, user.id, fiat_amount="not-a-number") + good = _paid_payment(wallet.id, user.id) + captured["queue"].put_nowait(poison) + captured["queue"].put_nowait(good) + + row = None + for _ in range(100): + row = await libra_crud.get_processed_payment(good.payment_hash) + if row and row["status"] == "done": + break + await asyncio.sleep(0.1) + assert row is not None and row["status"] == "done", ( + "good payment was not recorded after the poison payment" + ) + finally: + listener.cancel() + + # Startup cleared the stale claim; the poison payment's claim was + # released on failure so redelivery could retry it. + assert await libra_crud.get_processed_payment(stale_hash) is None + assert await libra_crud.get_processed_payment(poison.payment_hash) is None + + +# --------------------------------------------------------------------------- +# POST /record-payment — the client-driven path +# --------------------------------------------------------------------------- + + +async def _insert_paid_payment_row(wallet_id: str, user_id: str) -> str: + payment_hash = uuid4().hex + uuid4().hex[:32] + await create_payment( + checking_id=payment_hash, + data=CreatePayment( + wallet_id=wallet_id, + payment_hash=payment_hash, + bolt11="lnbcfake", + amount_msat=100_000_000, + memo="idempotency test", + extra={ + "tag": "libra", + "user_id": user_id, + "fiat_currency": "EUR", + "fiat_amount": "100.00", + }, + ), + status=PaymentState.SUCCESS, + ) + return payment_hash + + +async def test_record_payment_conflicts_while_in_flight( + client, super_user_headers, configured_user, standard_accounts +): + user, wallet = await _setup_receivable( + client, super_user_headers, configured_user, standard_accounts + ) + payment_hash = await _insert_paid_payment_row(wallet.id, user.id) + + # Another claimant (e.g. the background listener) is mid-recording. + assert await libra_crud.claim_payment(payment_hash) + + r = await client.post( + "/libra/api/v1/record-payment", + headers={"X-Api-Key": wallet.inkey}, + json={"payment_hash": payment_hash}, + ) + assert r.status_code == 409, r.text + + # Once that claimant finishes, a replay reports "already recorded" + # instead of writing a second entry. + await libra_crud.mark_payment_done(payment_hash, f"ln-{payment_hash[:16]}") + r = await client.post( + "/libra/api/v1/record-payment", + headers={"X-Api-Key": wallet.inkey}, + json={"payment_hash": payment_hash}, + ) + assert r.status_code == 200, r.text + assert "already recorded" in r.json()["message"].lower() + + +async def test_record_payment_records_once_then_replays_safely( + client, super_user_headers, configured_user, standard_accounts +): + user, wallet = await _setup_receivable( + client, super_user_headers, configured_user, standard_accounts + ) + payment_hash = await _insert_paid_payment_row(wallet.id, user.id) + + r = await client.post( + "/libra/api/v1/record-payment", + headers={"X-Api-Key": wallet.inkey}, + json={"payment_hash": payment_hash}, + ) + assert r.status_code == 200, r.text + assert r.json()["message"] == "Payment recorded successfully" + + r = await client.post( + "/libra/api/v1/record-payment", + headers={"X-Api-Key": wallet.inkey}, + json={"payment_hash": payment_hash}, + ) + assert r.status_code == 200, r.text + assert "already recorded" in r.json()["message"].lower() + + link = f"ln-{payment_hash[:16]}" + assert len(await _entries_with_link(client, wallet.inkey, link)) == 1 diff --git a/tests/test_reconciliation_api.py b/tests/test_reconciliation_api.py index 66757be..a168ff8 100644 --- a/tests/test_reconciliation_api.py +++ b/tests/test_reconciliation_api.py @@ -18,19 +18,6 @@ from uuid import uuid4 import pytest -# Tests that try to actually create + check an assertion all hit issue #39: -# `format_balance` returns a Beancount source string but `fava.add_entry` -# expects a dict, so Fava 500s on every assertion-create call. The contract -# violation is on libra's side; mark these strict-xfail so they go green -# automatically once #39 lands and the format_balance return shape is fixed. -ASSERTION_CREATE_BROKEN = pytest.mark.xfail( - reason="libra/issues/39 — POST /assertions submits a Beancount source string " - "to Fava's JSON API and 500s. Drop this marker when the format_balance " - "return type is changed to a dict.", - strict=True, -) - - # --------------------------------------------------------------------------- # helpers (local — assertion endpoints don't have wrapper helpers yet) # --------------------------------------------------------------------------- @@ -58,7 +45,6 @@ async def _create_assertion( # --------------------------------------------------------------------------- -@ASSERTION_CREATE_BROKEN @pytest.mark.anyio async def test_assertion_against_empty_account_passes( client, super_user_headers, standard_accounts, @@ -79,7 +65,6 @@ async def test_assertion_against_empty_account_passes( assert body.get("difference_sats", 0) == 0 -@ASSERTION_CREATE_BROKEN @pytest.mark.anyio async def test_assertion_with_wrong_balance_returns_409( client, super_user_headers, standard_accounts, @@ -102,7 +87,6 @@ async def test_assertion_with_wrong_balance_returns_409( assert detail.get("difference_sats") == 999_999 or detail.get("difference_sats") == -999_999 -@ASSERTION_CREATE_BROKEN @pytest.mark.anyio async def test_assertion_with_tolerance_accepts_small_diff( client, super_user_headers, standard_accounts, @@ -119,7 +103,6 @@ async def test_assertion_with_tolerance_accepts_small_diff( assert r.json().get("status") == "passed" -@ASSERTION_CREATE_BROKEN @pytest.mark.anyio async def test_list_assertions_returns_created( client, super_user_headers, standard_accounts, @@ -145,7 +128,6 @@ async def test_list_assertions_returns_created( assert assertion_id in ids, f"created assertion {assertion_id} missing from list {ids}" -@ASSERTION_CREATE_BROKEN @pytest.mark.anyio async def test_get_assertion_by_id( client, super_user_headers, standard_accounts, @@ -167,7 +149,6 @@ async def test_get_assertion_by_id( assert r.json().get("id") == assertion_id -@ASSERTION_CREATE_BROKEN @pytest.mark.anyio async def test_recheck_assertion_via_check_endpoint( client, super_user_headers, standard_accounts, @@ -190,7 +171,6 @@ async def test_recheck_assertion_via_check_endpoint( assert r.json().get("status") == "passed" -@ASSERTION_CREATE_BROKEN @pytest.mark.anyio async def test_delete_assertion_removes_it( client, super_user_headers, standard_accounts, diff --git a/tests/test_settlement_api.py b/tests/test_settlement_api.py index 442a01e..f7de090 100644 --- a/tests/test_settlement_api.py +++ b/tests/test_settlement_api.py @@ -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`. """ import importlib +from decimal import Decimal from uuid import uuid4 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}" payload = r.json().get("detail") assert isinstance(payload, dict), f"expected structured detail, got {payload!r}" - assert payload.get("cash_paid") == 30.0 - assert payload.get("net_obligation") == 100.0 - assert payload.get("receivable_total") == 100.0 - assert payload.get("payable_total") == 0.0 + # Amounts are exact Decimal strings (not floats) so the operator can + # act on them without precision loss. + assert Decimal(payload.get("cash_paid")) == Decimal("30.00") + 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 diff --git a/tests/test_unit.py b/tests/test_unit.py index 1c7dabc..11779e9 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -553,11 +553,6 @@ def test_validate_metadata_fiat_amount_without_currency_raises(): 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(): with pytest.raises(val.ValidationError) as exc: 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(): 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", + } diff --git a/tests/test_void_reject_api.py b/tests/test_void_reject_api.py index 66e2180..0f2b817 100644 --- a/tests/test_void_reject_api.py +++ b/tests/test_void_reject_api.py @@ -210,3 +210,69 @@ async def test_double_reject_returns_404_on_second_call( assert r.status_code in (200, 404), ( f"second reject should be deterministic, got {r.status_code}: {r.text}" ) + + +@pytest.mark.anyio +async def test_concurrent_approve_and_reject_are_serialized( + client, super_user_headers, configured_user, standard_accounts, +): + """Two mutations of the same ledger source file fired concurrently must + BOTH land. Before libra-#23 each endpoint did its own read-modify-write + with raw httpx and no lock, so one writer overwrote the other's change + (or 412'd on the stale checksum). Now both route through + FavaClient.transform_source_line under the global write lock. + """ + import asyncio + + _, wallet = configured_user + approve_tag = f"conc-approve-{uuid4().hex[:6]}" + reject_tag = f"conc-reject-{uuid4().hex[:6]}" + + posted = {} + for tag in (approve_tag, reject_tag): + posted[tag] = await post_expense( + client, + wallet_inkey=wallet.inkey, + user_wallet_id=wallet.id, + amount="10.00", + currency="EUR", + description=tag, + expense_account=standard_accounts["expense_food"]["name"], + ) + + # Force a Fava reload so the approve/reject lookups see both fresh + # pending entries (see #37). + await list_user_entries(client, wallet_inkey=wallet.inkey) + + r_approve, r_reject = await asyncio.gather( + client.post( + f"/libra/api/v1/entries/{posted[approve_tag]['id']}/approve", + headers=super_user_headers, + ), + client.post( + f"/libra/api/v1/entries/{posted[reject_tag]['id']}/reject", + headers=super_user_headers, + ), + ) + assert r_approve.status_code == 200, f"approve: {r_approve.text}" + assert r_reject.status_code == 200, f"reject: {r_reject.text}" + + # Both mutations must be visible: one entry voided, the other cleared + # (a cleared entry no longer matches the pending-only reject lookup). + listing = await list_user_entries(client, wallet_inkey=wallet.inkey) + entries = listing.get("entries", []) + rejected = next( + (e for e in entries if reject_tag in (e.get("description") or "")), None, + ) + assert rejected is not None and "voided" in rejected.get("tags", []), ( + f"rejected entry lost its #voided tag: {rejected}" + ) + + second_reject = await client.post( + f"/libra/api/v1/entries/{posted[approve_tag]['id']}/reject", + headers=super_user_headers, + ) + assert second_reject.status_code == 404, ( + "approved entry should no longer match the pending-only reject " + f"lookup, got {second_reject.status_code}" + ) diff --git a/views_api.py b/views_api.py index 1b3149a..1c3269f 100644 --- a/views_api.py +++ b/views_api.py @@ -13,6 +13,7 @@ from lnbits.decorators import ( ) from lnbits.utils.exchange_rates import allowed_currencies, fiat_amount_as_satoshis +from .beancount_format import fiat_rate_metadata from .crud import ( approve_manual_payment_request, check_balance_assertion, @@ -1074,8 +1075,7 @@ async def api_create_expense_entry( metadata = { "fiat_currency": data.currency.upper(), "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, - "btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0, + **fiat_rate_metadata(amount_sats, data.amount), } # Get or create expense account @@ -1272,8 +1272,7 @@ async def api_create_income_entry( metadata = { "fiat_currency": fiat_currency, "fiat_amount": str(data.amount.quantize(Decimal("0.001"))), - "fiat_rate": float(amount_sats) / float(data.amount) if data.amount > 0 else 0, - "btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0, + **fiat_rate_metadata(amount_sats, data.amount), } # Submit to Fava @@ -1371,8 +1370,7 @@ async def api_create_receivable_entry( metadata = { "fiat_currency": data.currency.upper(), "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, - "btc_rate": float(data.amount) / float(amount_sats) * 100_000_000 if amount_sats > 0 else 0, + **fiat_rate_metadata(amount_sats, data.amount), } # Get or create revenue account @@ -1760,15 +1758,10 @@ async def api_generate_payment_invoice( proportion = Decimal(data.amount) / Decimal(total_sat_balance) 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({ "fiat_currency": fiat_currency, "fiat_amount": str(invoice_fiat_amount.quantize(Decimal("0.001"))), - "fiat_rate": fiat_rate, - "btc_rate": btc_rate, + **fiat_rate_metadata(data.amount, invoice_fiat_amount), }) logger.info(f"Invoice extra metadata: {invoice_extra}") @@ -1850,90 +1843,130 @@ async def api_record_payment( try: async with httpx.AsyncClient(timeout=5.0) as client: - # Get recent entries from Fava's journal endpoint + # Get recent entries from Fava's journal endpoint. base_url + # already ends in /api — the previous "/api/journal" path + # 404'd, so this duplicate check silently never ran. response = await client.get( - f"{fava.base_url}/api/journal", + f"{fava.base_url}/journal", params={"time": ""} # Get all entries ) + response.raise_for_status() + response_data = response.json() + entries = response_data.get('entries', []) - if response.status_code == 200: - response_data = response.json() - entries = response_data.get('entries', []) - - # Check if any entry has our payment link - for entry in entries: - entry_links = entry.get('links', []) - if link_to_find in entry_links: - # Payment already recorded, return existing entry - balance_data = await fava.get_user_balance_bql(target_user_id) - return { - "journal_entry_id": f"fava-exists-{data.payment_hash[:16]}", - "new_balance": balance_data["balance"], - "message": "Payment already recorded", - } - except Exception as e: + # Check if any entry has our payment link + for entry in entries: + entry_links = entry.get('links', []) + if link_to_find in entry_links: + # Payment already recorded, return existing entry + balance_data = await fava.get_user_balance_bql(target_user_id) + return { + "journal_entry_id": f"fava-exists-{data.payment_hash[:16]}", + "new_balance": balance_data["balance"], + "message": "Payment already recorded", + } + except httpx.HTTPError as e: + # Fail CLOSED: if Fava can't confirm the payment isn't already + # recorded, refuse to write — proceeding on a transient blip is + # how double entries happen. The client can simply retry. logger.warning(f"Could not check Fava for duplicate payment: {e}") - # Continue anyway - Fava/Beancount will catch duplicate if it exists - - # Convert amount from millisatoshis to satoshis - amount_sats = payment.amount // 1000 - - # Extract fiat metadata from invoice (if present) - fiat_currency = None - fiat_amount = None - if payment.extra and isinstance(payment.extra, dict): - logger.info(f"Payment.extra contents: {payment.extra}") - fiat_currency = payment.extra.get("fiat_currency") - fiat_amount_str = payment.extra.get("fiat_amount") - if fiat_amount_str: - from decimal import Decimal - fiat_amount = Decimal(str(fiat_amount_str)) - - logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}") - - # Get user's receivable account (what user owes) - user_receivable = await get_or_create_user_account( - target_user_id, AccountType.ASSET, "Accounts Receivable" - ) - - # Get lightning account - lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning") - if not lightning_account: raise HTTPException( - status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found" + status_code=HTTPStatus.SERVICE_UNAVAILABLE, + detail="Cannot verify payment duplicate status; try again shortly", ) - # Get unsettled receivable entries to link to this settlement - unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable") - settled_links = [e["link"] for e in unsettled if e.get("link")] - - # Format payment entry and submit to Fava - entry = format_payment_entry( - user_id=target_user_id, - payment_account=lightning_account.name, - payable_or_receivable_account=user_receivable.name, - amount_sats=amount_sats, - description=f"Lightning payment from user {target_user_id[:8]}", - entry_date=datetime.now().date(), - is_payable=False, # User paying libra (receivable settlement) - fiat_currency=fiat_currency, - fiat_amount=fiat_amount, - payment_hash=data.payment_hash, - reference=data.payment_hash, - settled_entry_links=settled_links + # Local idempotency gate shared with the background invoice listener + # (tasks.on_invoice_paid): exactly one claimant records a payment_hash. + from .crud import ( + claim_payment, + get_processed_payment, + mark_payment_done, + release_payment_claim, ) - logger.info(f"Formatted payment entry: {entry}") + if not await claim_payment(data.payment_hash): + existing = await get_processed_payment(data.payment_hash) + if existing and existing["status"] == "done": + balance_data = await fava.get_user_balance_bql(target_user_id) + return { + "journal_entry_id": existing.get("entry_id") + or f"fava-exists-{data.payment_hash[:16]}", + "new_balance": balance_data["balance"], + "message": "Payment already recorded", + } + raise HTTPException( + status_code=HTTPStatus.CONFLICT, + detail="Payment is being recorded; check balance shortly", + ) - # Submit to Fava - result = await fava.add_entry(entry) - logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}") + # Convert amount from millisatoshis to satoshis + try: + amount_sats = payment.amount // 1000 + + # Extract fiat metadata from invoice (if present) + fiat_currency = None + fiat_amount = None + if payment.extra and isinstance(payment.extra, dict): + logger.info(f"Payment.extra contents: {payment.extra}") + fiat_currency = payment.extra.get("fiat_currency") + fiat_amount_str = payment.extra.get("fiat_amount") + if fiat_amount_str: + from decimal import Decimal + fiat_amount = Decimal(str(fiat_amount_str)) + + logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}") + + # Get user's receivable account (what user owes) + user_receivable = await get_or_create_user_account( + target_user_id, AccountType.ASSET, "Accounts Receivable" + ) + + # Get lightning account + lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning") + if not lightning_account: + raise HTTPException( + status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found" + ) + + # Get unsettled receivable entries to link to this settlement + unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable") + settled_links = [e["link"] for e in unsettled if e.get("link")] + + # Format payment entry and submit to Fava + entry = format_payment_entry( + user_id=target_user_id, + payment_account=lightning_account.name, + payable_or_receivable_account=user_receivable.name, + amount_sats=amount_sats, + description=f"Lightning payment from user {target_user_id[:8]}", + entry_date=datetime.now().date(), + is_payable=False, # User paying libra (receivable settlement) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount, + payment_hash=data.payment_hash, + reference=data.payment_hash, + settled_entry_links=settled_links + ) + + logger.info(f"Formatted payment entry: {entry}") + + # Submit to Fava + result = await fava.add_entry(entry) + logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}") + + entry_id = f"ln-{data.payment_hash[:16]}" + await mark_payment_done(data.payment_hash, entry_id) + except BaseException: + # Release the claim so a retry (client or background listener) + # can record this payment. + await release_payment_claim(data.payment_hash) + raise # Get updated balance from Fava balance_data = await fava.get_user_balance_bql(target_user_id) return { - "journal_entry_id": f"fava-{datetime.now().timestamp()}", + "journal_entry_id": entry_id, "new_balance": balance_data["balance"], "message": "Payment recorded successfully", } @@ -2024,6 +2057,19 @@ async def api_settle_receivable( unsettled_payables = await fava.get_unsettled_entries_bql(data.user_id, "expense") 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( (Decimal(str(e["fiat_amount"])) for e in unsettled_payables), Decimal(0), @@ -2067,10 +2113,10 @@ async def api_settle_receivable( "net to clear all open entries, or pass " "`settled_entry_links` to settle a specific subset." ), - "cash_paid": float(cash_paid), - "net_obligation": float(net_obligation), - "receivable_total": float(receivable_total), - "payable_total": float(payable_total), + "cash_paid": str(cash_paid), + "net_obligation": str(net_obligation), + "receivable_total": str(receivable_total), + "payable_total": str(payable_total), "currency": data.currency.upper(), }, ) @@ -2857,8 +2903,7 @@ async def api_approve_expense_entry( This updates the transaction in the Beancount file via Fava API. """ - import httpx - from .fava_client import get_fava_client + from .fava_client import ChecksumConflictError, get_fava_client fava = get_fava_client() @@ -2893,57 +2938,29 @@ async def api_approve_expense_entry( detail="Entry metadata missing filename or lineno" ) - # 3. Get the source file from Fava - async with httpx.AsyncClient(timeout=fava.timeout) as client: - response = await client.get( - f"{fava.base_url}/source", - params={"filename": filename} - ) - response.raise_for_status() - source_data = response.json()["data"] + # 3. Flip the flag under FavaClient's write lock — the whole + # read-modify-write is atomic against every other ledger writer. + old_pattern = f"{date_str} !" - sha256sum = source_data["sha256sum"] - source = source_data["source"] - lines = source.split('\n') - - # 4. Find and modify the entry at the specified line - # Line numbers are 1-indexed, list is 0-indexed - entry_line_idx = lineno - 1 - - if entry_line_idx >= len(lines): + def _approve(line: str) -> str: + if old_pattern not in line: raise HTTPException( status_code=HTTPStatus.INTERNAL_SERVER_ERROR, - detail=f"Line {lineno} not found in source file" + detail=f"Line {lineno} does not contain expected pattern '{old_pattern}'. Found: {line}" ) + return line.replace(old_pattern, f"{date_str} *", 1) - entry_line = lines[entry_line_idx] - - # Check if the line contains the pending flag pattern - old_pattern = f"{date_str} !" - if old_pattern not in entry_line: - raise HTTPException( - status_code=HTTPStatus.INTERNAL_SERVER_ERROR, - detail=f"Line {lineno} does not contain expected pattern '{old_pattern}'. Found: {entry_line}" - ) - - # Replace the flag - new_pattern = f"{date_str} *" - new_line = entry_line.replace(old_pattern, new_pattern, 1) - lines[entry_line_idx] = new_line - - # 5. Write back the modified source - new_source = '\n'.join(lines) - - update_response = await client.put( - f"{fava.base_url}/source", - json={ - "file_path": filename, - "source": new_source, - "sha256sum": sha256sum - }, - headers={"Content-Type": "application/json"} + try: + await fava.transform_source_line(filename, lineno, _approve) + except ValueError as e: + raise HTTPException( + status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e) + ) + except ChecksumConflictError: + raise HTTPException( + status_code=HTTPStatus.CONFLICT, + detail="Ledger changed concurrently; retry the approval", ) - update_response.raise_for_status() logger.info(f"Entry {entry_id} approved (flag changed to *)") @@ -2966,8 +2983,7 @@ async def api_reject_expense_entry( Adds #voided tag for audit trail while keeping the '!' flag. Voided transactions are excluded from balances but preserved in the ledger. """ - import httpx - from .fava_client import get_fava_client + from .fava_client import ChecksumConflictError, get_fava_client fava = get_fava_client() @@ -3002,50 +3018,26 @@ async def api_reject_expense_entry( detail="Entry metadata missing filename or lineno" ) - # 3. Get the source file from Fava - async with httpx.AsyncClient(timeout=fava.timeout) as client: - response = await client.get( - f"{fava.base_url}/source", - params={"filename": filename} + # 3. Add the #voided tag under FavaClient's write lock — the whole + # read-modify-write is atomic against every other ledger writer. + def _void(line: str) -> str: + if "#voided" in line: + return line # already voided — no-op + return line.rstrip() + ' #voided' + + try: + changed = await fava.transform_source_line(filename, lineno, _void) + except ValueError as e: + raise HTTPException( + status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e) ) - response.raise_for_status() - source_data = response.json()["data"] - - sha256sum = source_data["sha256sum"] - source = source_data["source"] - lines = source.split('\n') - - # 4. Find and modify the entry at the specified line - add #voided tag - entry_line_idx = lineno - 1 - - if entry_line_idx >= len(lines): - raise HTTPException( - status_code=HTTPStatus.INTERNAL_SERVER_ERROR, - detail=f"Line {lineno} not found in source file" - ) - - entry_line = lines[entry_line_idx] - - # Add #voided tag if not already present - if "#voided" not in entry_line: - # Add #voided tag to the transaction line - new_line = entry_line.rstrip() + ' #voided' - lines[entry_line_idx] = new_line - - # 5. Write back the modified source - new_source = '\n'.join(lines) - - update_response = await client.put( - f"{fava.base_url}/source", - json={ - "file_path": filename, - "source": new_source, - "sha256sum": sha256sum - }, - headers={"Content-Type": "application/json"} - ) - update_response.raise_for_status() - logger.info(f"Entry {entry_id} rejected (added #voided tag)") + except ChecksumConflictError: + raise HTTPException( + status_code=HTTPStatus.CONFLICT, + detail="Ledger changed concurrently; retry the rejection", + ) + if changed: + logger.info(f"Entry {entry_id} rejected (added #voided tag)") return { "message": f"Entry {entry_id} rejected (marked as voided)", @@ -3774,8 +3766,14 @@ async def api_admin_add_chart_account( "already_existed": True, } - # Mirror into libra DB so permissions / metadata layer sees it. - synced = await sync_single_account_from_beancount(payload.name) + # Mirror into libra DB so permissions / metadata layer sees it. We just + # wrote the Open directive ourselves, so skip the verification + # round-trip through Fava (libra-#53). + synced = await sync_single_account_from_beancount( + payload.name, + description=payload.description, + assume_exists=True, + ) return { "success": True,