diff --git a/__init__.py b/__init__.py index 42a2648..614a8ba 100644 --- a/__init__.py +++ b/__init__.py @@ -30,17 +30,6 @@ 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 870d1d7..3d82381 100644 --- a/account_sync.py +++ b/account_sync.py @@ -285,11 +285,7 @@ async def sync_accounts_from_beancount(force_full_sync: bool = False) -> dict: return stats -async def sync_single_account_from_beancount( - account_name: str, - description: Optional[str] = None, - assume_exists: bool = False, -) -> bool: +async def sync_single_account_from_beancount(account_name: str) -> bool: """ Sync a single account from Beancount to Libra DB. @@ -298,13 +294,6 @@ async def sync_single_account_from_beancount( 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 @@ -317,22 +306,6 @@ async def sync_single_account_from_beancount( 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 fbdfba9..f233ee5 100644 --- a/beancount_format.py +++ b/beancount_format.py @@ -115,18 +115,13 @@ def format_balance( account: str, amount: int, currency: str = "SATS" -) -> Dict[str, Any]: +) -> str: """ - Format a balance assertion directive for Fava's JSON API. + Format a balance assertion directive for Beancount. 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") @@ -134,15 +129,15 @@ def format_balance( currency: Currency code (default: "SATS") Returns: - Fava API Balance entry dict ready for `fava.add_entry`. + 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' """ - return { - "t": "Balance", - "date": date_val.strftime('%Y-%m-%d'), - "account": account, - "amount": {"number": str(amount), "currency": currency}, - "meta": {}, - } + date_str = date_val.strftime('%Y-%m-%d') + # Two spaces between account and amount (Beancount convention) + return f"{date_str} balance {account} {amount} {currency}" def format_posting_with_cost( @@ -257,10 +252,9 @@ 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 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") + # No cost + amount_str = f"{amount_sats} SATS {{}}" + logger.warning(f"format_posting_at_average_cost: cost_currency is None, using empty cost basis") posting_meta = metadata or {} @@ -310,25 +304,6 @@ 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, @@ -743,18 +718,15 @@ def format_net_settlement_entry( entry_date: date, payment_hash: Optional[str] = None, reference: Optional[str] = None, - settled_entry_links: Optional[List[str]] = None, - credit_account: Optional[str] = None, - credit_overflow_fiat: Decimal = Decimal(0), + settled_entry_links: Optional[List[str]] = None ) -> Dict[str, Any]: """ Format a net settlement payment entry (user paying net balance). - Creates a three- to four-posting transaction: + Creates a three-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 @@ -762,61 +734,25 @@ 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: Fiat value of the payment being recorded - total_receivable_fiat: Receivables cleared by this payment - total_payable_fiat: Payables cleared by this payment + net_fiat_amount: Net fiat amount (receivable - payable) + total_receivable_fiat: Total receivables to clear + total_payable_fiat: Total payables to clear 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. @@ -825,26 +761,20 @@ def format_net_settlement_entry( postings = [ { "account": payment_account, - "amount": f"{abs(amount_sats)} SATS @@ {net_fiat_amount:.2f} {fiat_currency}", + "amount": f"{abs(amount_sats)} SATS @@ {abs(net_fiat_amount):.2f} {fiat_currency}", "meta": {"payment-hash": payment_hash} if payment_hash else {} }, { "account": receivable_account, - "amount": f"-{total_receivable_fiat:.2f} {fiat_currency}", + "amount": f"-{abs(total_receivable_fiat):.2f} {fiat_currency}", "meta": {"sats-equivalent": str(abs(amount_sats))} }, { "account": payable_account, - "amount": f"{total_payable_fiat:.2f} {fiat_currency}", + "amount": f"{abs(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 913bb8a..c29f069 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, InvalidOperation +from decimal import Decimal from typing import Any, Dict, List, Optional @@ -278,14 +278,11 @@ def validate_metadata( } ) - # 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). + # Validate fiat amount is valid Decimal if has_fiat_amount: try: Decimal(str(metadata["fiat_amount"])) - except (ValueError, TypeError, InvalidOperation) as e: + except (ValueError, TypeError) as e: raise ValidationError( f"Invalid fiat_amount: {metadata['fiat_amount']}", {"error": str(e)} diff --git a/crud.py b/crud.py index b49bb9b..0692806 100644 --- a/crud.py +++ b/crud.py @@ -1696,73 +1696,3 @@ async def check_user_has_role_permission( return True return False - - -# ============================================================================= -# PROCESSED PAYMENTS (Lightning payment idempotency gate) -# ============================================================================= -# The Fava-side duplicate checks are read-then-write races; this table's -# primary key on payment_hash makes exactly one claimant win. Shared by the -# background invoice listener (tasks.on_invoice_paid) and the client-driven -# /record-payment endpoint. - - -async def claim_payment(payment_hash: str) -> bool: - """Atomically claim a Lightning payment for recording. - - Returns True when this caller owns the claim; False when the payment - is already recorded or another coroutine is recording it right now. - """ - result = await db.execute( - """ - INSERT INTO processed_payments (payment_hash, status) - VALUES (:payment_hash, 'processing') - ON CONFLICT (payment_hash) DO NOTHING - """, - {"payment_hash": payment_hash}, - ) - return result.rowcount == 1 - - -async def get_processed_payment(payment_hash: str) -> Optional[dict]: - row = await db.fetchone( - "SELECT payment_hash, status, entry_id FROM processed_payments" - " WHERE payment_hash = :payment_hash", - {"payment_hash": payment_hash}, - ) - return dict(row) if row else None - - -async def mark_payment_done(payment_hash: str, entry_id: Optional[str] = None) -> None: - await db.execute( - """ - UPDATE processed_payments SET status = 'done', entry_id = :entry_id - WHERE payment_hash = :payment_hash - """, - {"payment_hash": payment_hash, "entry_id": entry_id}, - ) - - -async def release_payment_claim(payment_hash: str) -> None: - """Compensating delete after a failed recording, so redelivery retries. - - Only removes an in-flight claim — a 'done' row is permanent. - """ - await db.execute( - "DELETE FROM processed_payments" - " WHERE payment_hash = :payment_hash AND status = 'processing'", - {"payment_hash": payment_hash}, - ) - - -async def clear_stale_payment_claims() -> int: - """Drop 'processing' claims left behind by a previous process life. - - A live claim only exists inside a running coroutine, so anything - still 'processing' at listener startup belongs to a crashed or - restarted process and would otherwise block that payment forever. - """ - result = await db.execute( - "DELETE FROM processed_payments WHERE status = 'processing'" - ) - return result.rowcount diff --git a/fava_client.py b/fava_client.py index faad8f7..eaed06b 100644 --- a/fava_client.py +++ b/fava_client.py @@ -20,8 +20,7 @@ See: https://github.com/beancount/fava/blob/main/src/fava/json_api.py import asyncio import re import httpx -from contextlib import asynccontextmanager -from typing import Any, AsyncIterator, Callable, Dict, List, Optional +from typing import Any, Dict, List, Optional from decimal import Decimal from datetime import date, datetime from loguru import logger @@ -45,34 +44,6 @@ 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. @@ -165,27 +136,6 @@ 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. @@ -210,7 +160,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 self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: resp = await client.get(f"{self.base_url}/options") resp.raise_for_status() main_file = resp.json()["data"]["beancount_options"]["filename"] @@ -286,7 +236,7 @@ class FavaClient: # Acquire global write lock to serialize ledger modifications async with self._write_lock: try: - async with self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.put( f"{self.base_url}/add_entries", json={"entries": [entry]}, @@ -400,11 +350,10 @@ 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 self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.get( f"{self.base_url}/query", params={"query_string": query} @@ -497,14 +446,14 @@ class FavaClient: import re # Try total price notation: "50.00 EUR @@ 50000 SATS" - total_price_match = _TOTAL_PRICE_RE.match(amount_str) + total_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?\d+)\s+SATS$', amount_str) # Try per-unit price notation: "50.00 EUR @ 1000.5 SATS" - unit_price_match = _UNIT_PRICE_RE.match(amount_str) + unit_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$', amount_str) if total_price_match: fiat_amount = Decimal(total_price_match.group(1)) fiat_currency = total_price_match.group(2) - sats_amount = _sats_to_int(total_price_match.group(3)) + sats_amount = int(total_price_match.group(3)) if fiat_currency not in fiat_balances: fiat_balances[fiat_currency] = Decimal(0) @@ -531,8 +480,8 @@ class FavaClient: accounts_dict[account_name]["sats"] += sats_amount # Try simple fiat format: "50.00 EUR" (check metadata for sats) - elif _FIAT_AMOUNT_RE.match(amount_str): - fiat_match = _FIAT_AMOUNT_RE.match(amount_str) + elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str): + fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', 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) @@ -553,9 +502,9 @@ class FavaClient: else: # Old format: SATS with cost/price notation - extract SATS amount - sats_match = _SATS_AMOUNT_RE.match(amount_str) + sats_match = re.match(r'^(-?\d+)\s+SATS', amount_str) if sats_match: - sats_amount = _sats_to_int(sats_match.group(1)) + sats_amount = int(sats_match.group(1)) total_sats += sats_amount # Track per account @@ -654,14 +603,14 @@ class FavaClient: import re # Try total price notation: "50.00 EUR @@ 50000 SATS" - total_price_match = _TOTAL_PRICE_RE.match(amount_str) + total_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@@\s+(-?\d+)\s+SATS$', amount_str) # Try per-unit price notation: "50.00 EUR @ 1000.5 SATS" - unit_price_match = _UNIT_PRICE_RE.match(amount_str) + unit_price_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})\s+@\s+([\d.]+)\s+SATS$', amount_str) if total_price_match: fiat_amount = Decimal(total_price_match.group(1)) fiat_currency = total_price_match.group(2) - sats_amount = _sats_to_int(total_price_match.group(3)) + sats_amount = 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) @@ -680,8 +629,8 @@ class FavaClient: user_data[user_id]["balance"] += sats_amount # Try simple fiat format: "50.00 EUR" (check metadata for sats) - elif _FIAT_AMOUNT_RE.match(amount_str): - fiat_match = _FIAT_AMOUNT_RE.match(amount_str) + elif re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', amount_str): + fiat_match = re.match(r'^(-?[\d.]+)\s+([A-Z]{3})$', 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) @@ -699,9 +648,9 @@ class FavaClient: else: # Old format: SATS with cost/price notation - sats_match = _SATS_AMOUNT_RE.match(amount_str) + sats_match = re.match(r'^(-?\d+)\s+SATS', amount_str) if sats_match: - sats_amount = _sats_to_int(sats_match.group(1)) + sats_amount = int(sats_match.group(1)) user_data[user_id]["balance"] += sats_amount # Extract fiat from cost syntax or metadata (backward compatibility) @@ -734,12 +683,9 @@ class FavaClient: True if Fava responds, False otherwise """ try: - async with self._client() as client: - # Health probes stay fast regardless of the configured - # request timeout. + async with httpx.AsyncClient(timeout=2.0) as client: response = await client.get( - f"{self.base_url}/changed", - timeout=2.0 + f"{self.base_url}/changed" ) return response.status_code == 200 except Exception as e: @@ -775,13 +721,12 @@ 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 self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.get( f"{self.base_url}/query", params={"query_string": query} @@ -862,7 +807,7 @@ class FavaClient: https://beancount.github.io/docs/beancount_query_language.html """ try: - async with self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.get( f"{self.base_url}/query", params={"query_string": query_string} @@ -1396,7 +1341,7 @@ class FavaClient: # (BQL's SELECT DISTINCT account only returns accounts with postings) account_names: set[str] = set() - async with self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: for endpoint in ("balance_sheet", "income_statement"): try: response = await client.get(f"{self.base_url}/{endpoint}") @@ -1495,7 +1440,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 self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.get(f"{self.base_url}/journal", params=params) response.raise_for_status() result = response.json() @@ -1537,7 +1482,7 @@ class FavaClient: sha256sum = context["sha256sum"] """ try: - async with self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.get( f"{self.base_url}/context", params={"entry_hash": entry_hash} @@ -1584,7 +1529,7 @@ class FavaClient: # Acquire global write lock to serialize ledger modifications async with self._write_lock: try: - async with self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.put( f"{self.base_url}/source_slice", json={ @@ -1599,78 +1544,11 @@ 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. @@ -1693,7 +1571,7 @@ class FavaClient: # Acquire global write lock to serialize ledger modifications async with self._write_lock: try: - async with self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.delete( f"{self.base_url}/source_slice", params={ @@ -1707,10 +1585,6 @@ 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}") @@ -1790,7 +1664,7 @@ class FavaClient: # Acquire global write lock to serialize ledger modifications async with self._write_lock: try: - async with self._client() as client: + async with httpx.AsyncClient(timeout=self.timeout) as client: # Step 1: Get current source file (fresh read on each attempt) response = await client.get( f"{self.base_url}/source", @@ -1943,7 +1817,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, currency, weight, links, + SELECT date, narration, account, number, weight, links, any_meta('entry-id') as entry_id WHERE account ~ '{account_pattern}' AND '{entry_tag}' IN tags @@ -1977,10 +1851,7 @@ class FavaClient: entries_by_link: Dict[str, Dict[str, Any]] = {} for row in original_result["rows"]: - ( - date_val, narration, account, number, currency, - weight, links, entry_id, - ) = row + date_val, narration, account, number, weight, links, entry_id = row # Skip if no links if not links or not isinstance(links, list): @@ -2004,11 +1875,9 @@ class FavaClient: if entry_link in entries_by_link: continue - # 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 amounts + fiat_amount = abs(float(number)) if number else 0.0 + fiat_currency = "EUR" # Default, could be extracted from posting # Parse SATS from weight column sats_amount = 0 diff --git a/migrations.py b/migrations.py index d8a4bba..9c38c55 100644 --- a/migrations.py +++ b/migrations.py @@ -34,33 +34,9 @@ Original migration sequence (Nov 2025): - m014: Removed legacy equity accounts (MemberEquity, RetainedEarnings) - m015: Converted entry_lines to single amount field - m016: Dropped journal_entries and entry_lines tables (Fava integration) - -IDEMPOTENCY CONTRACT: -Every statement here must be a silent no-op on re-run. The migration -version bump lands in the core LNbits DB (`dbversions`) while the DDL -lands in `ext_libra` — the two writes are not atomic. If the version -bump fails after the DDL commits, the whole migration re-runs on next -boot; a bare CREATE/ALTER/INSERT then crashes the extension until -manual dbversions surgery. """ -async def _alter_add_column_safe(db, sql: str) -> None: - """ALTER TABLE ADD COLUMN that swallows duplicate-column errors. - - Neither SQLite nor Postgres supports ADD COLUMN IF NOT EXISTS - portably, so re-runs are made no-ops by swallowing the error both - backends raise for an existing column. - """ - try: - await db.execute(sql) - except Exception as exc: - msg = str(exc).lower() - if "duplicate column" in msg or "already exists" in msg: - return - raise - - async def m001_initial(db): """ Initial Libra database schema (squashed from m001-m016). @@ -87,7 +63,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS accounts ( + CREATE TABLE accounts ( id TEXT PRIMARY KEY, name TEXT NOT NULL UNIQUE, account_type TEXT NOT NULL, @@ -100,13 +76,13 @@ async def m001_initial(db): await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_accounts_user_id ON accounts (user_id); + CREATE INDEX idx_accounts_user_id ON accounts (user_id); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_accounts_type ON accounts (account_type); + CREATE INDEX idx_accounts_type ON accounts (account_type); """ ) @@ -117,7 +93,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS extension_settings ( + CREATE TABLE extension_settings ( id TEXT NOT NULL PRIMARY KEY, libra_wallet_id TEXT, fava_url TEXT NOT NULL DEFAULT 'http://localhost:3333', @@ -135,7 +111,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS user_wallet_settings ( + CREATE TABLE user_wallet_settings ( id TEXT NOT NULL PRIMARY KEY, user_wallet_id TEXT, updated_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now} @@ -150,7 +126,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS manual_payment_requests ( + CREATE TABLE manual_payment_requests ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, amount INTEGER NOT NULL, @@ -167,14 +143,14 @@ async def m001_initial(db): await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_user_id + CREATE INDEX idx_manual_payment_requests_user_id ON manual_payment_requests (user_id); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_manual_payment_requests_status + CREATE INDEX idx_manual_payment_requests_status ON manual_payment_requests (status); """ ) @@ -187,7 +163,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS balance_assertions ( + CREATE TABLE balance_assertions ( id TEXT PRIMARY KEY, date TIMESTAMP NOT NULL, account_id TEXT NOT NULL, @@ -212,21 +188,21 @@ async def m001_initial(db): await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_balance_assertions_account_id + CREATE INDEX idx_balance_assertions_account_id ON balance_assertions (account_id); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_balance_assertions_status + CREATE INDEX idx_balance_assertions_status ON balance_assertions (status); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_balance_assertions_date + CREATE INDEX idx_balance_assertions_date ON balance_assertions (date); """ ) @@ -240,7 +216,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS user_equity_status ( + CREATE TABLE user_equity_status ( user_id TEXT PRIMARY KEY, is_equity_eligible BOOLEAN NOT NULL DEFAULT FALSE, equity_account_name TEXT, @@ -254,7 +230,7 @@ async def m001_initial(db): await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_user_equity_status_eligible + CREATE INDEX idx_user_equity_status_eligible ON user_equity_status (is_equity_eligible) WHERE is_equity_eligible = TRUE; """ @@ -269,7 +245,7 @@ async def m001_initial(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS account_permissions ( + CREATE TABLE account_permissions ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, account_id TEXT NOT NULL, @@ -286,7 +262,7 @@ async def m001_initial(db): # Index for looking up permissions by user await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_account_permissions_user_id + CREATE INDEX idx_account_permissions_user_id ON account_permissions (user_id); """ ) @@ -294,7 +270,7 @@ async def m001_initial(db): # Index for looking up permissions by account await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_account_permissions_account_id + CREATE INDEX idx_account_permissions_account_id ON account_permissions (account_id); """ ) @@ -302,7 +278,7 @@ async def m001_initial(db): # Composite index for checking specific user+account permissions await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_account_permissions_user_account + CREATE INDEX idx_account_permissions_user_account ON account_permissions (user_id, account_id); """ ) @@ -310,7 +286,7 @@ async def m001_initial(db): # Index for finding permissions by type await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_account_permissions_type + CREATE INDEX idx_account_permissions_type ON account_permissions (permission_type); """ ) @@ -318,7 +294,7 @@ async def m001_initial(db): # Index for finding expired permissions await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_account_permissions_expires + CREATE INDEX idx_account_permissions_expires ON account_permissions (expires_at) WHERE expires_at IS NOT NULL; """ @@ -344,7 +320,6 @@ async def m001_initial(db): f""" INSERT INTO accounts (id, name, account_type, description, created_at) VALUES (:id, :name, :type, :description, {db.timestamp_now}) - ON CONFLICT (name) DO NOTHING """, { "id": str(uuid.uuid4()), @@ -367,18 +342,17 @@ async def m002_add_account_is_active(db): Default: All existing accounts are marked as active (TRUE). """ - await _alter_add_column_safe( - db, + await db.execute( """ ALTER TABLE accounts ADD COLUMN is_active BOOLEAN NOT NULL DEFAULT TRUE - """, + """ ) # Create index for faster queries filtering by is_active await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_accounts_is_active ON accounts (is_active) + CREATE INDEX idx_accounts_is_active ON accounts (is_active) """ ) @@ -398,18 +372,17 @@ async def m003_add_account_is_virtual(db): Default: All existing accounts are real (is_virtual = FALSE). """ - await _alter_add_column_safe( - db, + await db.execute( """ ALTER TABLE accounts ADD COLUMN is_virtual BOOLEAN NOT NULL DEFAULT FALSE - """, + """ ) # Create index for faster queries filtering by is_virtual await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_accounts_is_virtual ON accounts (is_virtual) + CREATE INDEX idx_accounts_is_virtual ON accounts (is_virtual) """ ) @@ -429,7 +402,6 @@ async def m003_add_account_is_virtual(db): f""" INSERT INTO accounts (id, name, account_type, description, is_active, is_virtual, created_at) VALUES (:id, :name, :type, :description, TRUE, TRUE, {db.timestamp_now}) - ON CONFLICT (name) DO NOTHING """, { "id": str(uuid.uuid4()), @@ -466,7 +438,7 @@ async def m004_add_rbac_tables(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS roles ( + CREATE TABLE roles ( id TEXT PRIMARY KEY, name TEXT NOT NULL UNIQUE, description TEXT, @@ -479,13 +451,13 @@ async def m004_add_rbac_tables(db): await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_roles_name ON roles (name); + CREATE INDEX idx_roles_name ON roles (name); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_roles_is_default ON roles (is_default) + CREATE INDEX idx_roles_is_default ON roles (is_default) WHERE is_default = TRUE; """ ) @@ -497,7 +469,7 @@ async def m004_add_rbac_tables(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS role_permissions ( + CREATE TABLE role_permissions ( id TEXT PRIMARY KEY, role_id TEXT NOT NULL, account_id TEXT NOT NULL, @@ -512,19 +484,19 @@ async def m004_add_rbac_tables(db): await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_role_permissions_role_id ON role_permissions (role_id); + CREATE INDEX idx_role_permissions_role_id ON role_permissions (role_id); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_role_permissions_account_id ON role_permissions (account_id); + CREATE INDEX idx_role_permissions_account_id ON role_permissions (account_id); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_role_permissions_type ON role_permissions (permission_type); + CREATE INDEX idx_role_permissions_type ON role_permissions (permission_type); """ ) @@ -535,7 +507,7 @@ async def m004_add_rbac_tables(db): await db.execute( f""" - CREATE TABLE IF NOT EXISTS user_roles ( + CREATE TABLE user_roles ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, role_id TEXT NOT NULL, @@ -550,19 +522,19 @@ async def m004_add_rbac_tables(db): await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_user_roles_user_id ON user_roles (user_id); + CREATE INDEX idx_user_roles_user_id ON user_roles (user_id); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_user_roles_role_id ON user_roles (role_id); + CREATE INDEX idx_user_roles_role_id ON user_roles (role_id); """ ) await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_user_roles_expires ON user_roles (expires_at) + CREATE INDEX idx_user_roles_expires ON user_roles (expires_at) WHERE expires_at IS NOT NULL; """ ) @@ -570,7 +542,7 @@ async def m004_add_rbac_tables(db): # Composite index for checking specific user+role assignments await db.execute( """ - CREATE INDEX IF NOT EXISTS idx_user_roles_user_role ON user_roles (user_id, role_id); + CREATE INDEX idx_user_roles_user_role ON user_roles (user_id, role_id); """ ) @@ -614,7 +586,6 @@ async def m004_add_rbac_tables(db): f""" INSERT INTO roles (id, name, description, is_default, created_by, created_at) VALUES (:id, :name, :description, :is_default, :created_by, {db.timestamp_now}) - ON CONFLICT (name) DO NOTHING """, { "id": str(uuid.uuid4()), @@ -624,30 +595,3 @@ async def m004_add_rbac_tables(db): "created_by": "system", # System-created default roles }, ) - - -async def m005_add_processed_payments(db): - """ - Local idempotency gate for Lightning payment recording. - - The Fava-side duplicate check (`add_entry_idempotent`, journal-link - scan) is a read-then-write race: the background invoice listener and - the client-driven /record-payment endpoint can both pass the "not - present" check for the same payment_hash and both insert. The - primary key on payment_hash makes exactly one claimant win. - - status lifecycle: 'processing' (claimed, write in flight) → 'done' - (entry recorded). Failed claims are deleted so redelivery retries; - 'processing' rows from a crashed process are cleared at listener - startup. - """ - await db.execute( - f""" - CREATE TABLE IF NOT EXISTS processed_payments ( - payment_hash TEXT PRIMARY KEY, - status TEXT NOT NULL DEFAULT 'processing', - entry_id TEXT, - created_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now} - ); - """ - ) diff --git a/tasks.py b/tasks.py index 23dd0b5..8ed5a33 100644 --- a/tasks.py +++ b/tasks.py @@ -179,31 +179,12 @@ async def wait_for_paid_invoices(): This ensures payments are recorded even if the user closes their browser before the payment is detected by client-side polling. """ - from .crud import clear_stale_payment_claims - invoice_queue = Queue() register_invoice_listener(invoice_queue, "ext_libra") - # Claims from a previous process life can't be live anymore — clear - # them so those payments aren't blocked forever. - cleared = await clear_stale_payment_claims() - if cleared: - logger.warning( - f"[LIBRA] Cleared {cleared} stale in-flight payment claim(s) " - "from a previous run" - ) - while True: payment = await invoice_queue.get() - try: - await on_invoice_paid(payment) - except Exception: - # One bad payment must not kill the listener for the rest of - # the process lifetime; its claim was released, so redelivery - # can retry it. - logger.exception( - f"[LIBRA] Failed to record payment {payment.payment_hash}" - ) + await on_invoice_paid(payment) async def on_invoice_paid(payment: Payment) -> None: @@ -229,20 +210,8 @@ async def on_invoice_paid(payment: Payment) -> None: logger.warning(f"Libra invoice {payment.payment_hash} missing user_id in metadata") return - from .crud import claim_payment, mark_payment_done, release_payment_claim from .fava_client import get_fava_client - # Local idempotency gate: exactly one claimant (this listener or the - # /record-payment endpoint) gets to record a given payment_hash. The - # Fava-side idempotent write below stays as a second layer for - # entries recorded before this table existed. - if not await claim_payment(payment.payment_hash): - logger.info( - f"Payment {payment.payment_hash} already recorded or being " - "recorded; skipping" - ) - return - fava = get_fava_client() # Use idempotency key based on payment hash - this ensures duplicate @@ -276,36 +245,26 @@ 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 what this payment clears + # Get user's current balance to determine receivables and payables balance = await fava.get_user_balance(user_id) fiat_balances = balance.get("fiat_balances", {}) total_fiat_balance = fiat_balances.get(fiat_currency, Decimal(0)) - # 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 + # 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) - logger.info( - f"Settlement: {fiat_amount} {fiat_currency} " - f"(clears receivable: {total_receivable}, credit: {credit_overflow})" - ) + logger.info(f"Settlement: {fiat_amount} {fiat_currency} (Receivable: {total_receivable}, Payable: {total_payable})") # Get account names user_receivable = await get_or_create_user_account( @@ -314,35 +273,22 @@ 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 - # 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. + # Query for unsettled entries to link this settlement back to them + # Net settlement can settle both expenses and receivables settled_links = [] - 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 + 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 # Format as net settlement transaction entry = format_net_settlement_entry( @@ -359,9 +305,7 @@ 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, - credit_account=user_credit.name if user_credit else None, - credit_overflow_fiat=credit_overflow, + settled_entry_links=settled_links if settled_links else None ) # Submit to Fava using idempotent method to prevent duplicates @@ -380,11 +324,6 @@ async def on_invoice_paid(payment: Payment) -> None: f"{result.get('data', 'Unknown')}" ) - await mark_payment_done(payment.payment_hash, idempotency_key) - except Exception as e: logger.error(f"Error recording Libra payment {payment.payment_hash}: {e}") - # Release the claim so redelivery (or the /record-payment - # endpoint) can retry this payment. - await release_payment_claim(payment.payment_hash) raise diff --git a/tests/test_migrations.py b/tests/test_migrations.py deleted file mode 100644 index 1586473..0000000 --- a/tests/test_migrations.py +++ /dev/null @@ -1,110 +0,0 @@ -"""Migration idempotency tests. - -The migration version bump lands in the core LNbits DB (`dbversions`) -while the DDL lands in `ext_libra` — the two writes are not atomic. If -the bump fails after the DDL commits, the whole migration re-runs on -next boot, so every statement must be a silent no-op on re-run instead -of crashing with `duplicate column` / `table exists` / UNIQUE -violations (which bricks the extension until manual dbversions -surgery). - -These tests run the full migration chain twice against a fresh SQLite -database — the second pass simulates the lost-version-bump re-run. -""" -import importlib -import re -from uuid import uuid4 - -import pytest - -from lnbits.db import Database - -pytestmark = pytest.mark.anyio - - -def _module(name: str): - """Import a libra submodule under whichever path the active LNbits layout - uses (default `lnbits.extensions.libra` or bare `libra`).""" - for prefix in ("lnbits.extensions.libra", "libra"): - try: - return importlib.import_module(f"{prefix}.{name}") - except ModuleNotFoundError: - continue - raise ModuleNotFoundError(f"libra.{name}: tried both import paths") - - -migrations = _module("migrations") - -# Same discovery as lnbits.core.helpers.run_migration: m### prefix, -# module definition order. -_MIGRATION_RE = re.compile(r"^m(\d\d\d)_") -MIGRATION_FUNCTIONS = [ - fn for name, fn in vars(migrations).items() if _MIGRATION_RE.match(name) -] - -# Tables the chain must leave behind — one probe row read per table -# proves both existence and queryability after a double run. -EXPECTED_TABLES = [ - "accounts", - "extension_settings", - "user_wallet_settings", - "manual_payment_requests", - "balance_assertions", - "user_equity_status", - "account_permissions", - "roles", - "role_permissions", - "user_roles", - "processed_payments", -] - - -async def _run_all_migrations(db: Database) -> None: - async with db.connect() as conn: - for migrate in MIGRATION_FUNCTIONS: - await migrate(conn) - - -async def _seed_counts(db: Database) -> dict: - async with db.connect() as conn: - accounts = await conn.fetchall("SELECT id, name FROM accounts") - roles = await conn.fetchall("SELECT id, name FROM roles") - return { - "account_names": sorted(r["name"] for r in accounts), - "account_ids": sorted(r["id"] for r in accounts), - "role_names": sorted(r["name"] for r in roles), - "role_ids": sorted(r["id"] for r in roles), - } - - -async def test_migrations_rerun_is_noop(): - """Full chain twice: second run must not raise and not re-seed.""" - db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}") - - await _run_all_migrations(db) - first = await _seed_counts(db) - - # Simulate the lost dbversions bump: everything runs again. - await _run_all_migrations(db) - second = await _seed_counts(db) - - # Seeds must not duplicate (names) and must not be replaced (ids). - assert second == first - assert first["account_names"], "seed accounts missing after migration" - assert "Employee" in first["role_names"] - - # Every table exists and is queryable after the double run. - async with db.connect() as conn: - for table in EXPECTED_TABLES: - await conn.fetchall(f"SELECT * FROM {table} LIMIT 1") # noqa: S608 - - -async def test_single_migration_rerun_is_noop(): - """Each migration individually survives an immediate re-run (the - version bump fails right after that one migration committed).""" - db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}") - - async with db.connect() as conn: - for migrate in MIGRATION_FUNCTIONS: - await migrate(conn) - await migrate(conn) # re-run before "bumping" to the next diff --git a/tests/test_payment_idempotency.py b/tests/test_payment_idempotency.py deleted file mode 100644 index 969107b..0000000 --- a/tests/test_payment_idempotency.py +++ /dev/null @@ -1,281 +0,0 @@ -"""Lightning payment idempotency — the `processed_payments` claim gate. - -The background invoice listener (`tasks.on_invoice_paid`) and the -client-driven `POST /record-payment` endpoint can both fire for the -same `payment_hash` (queue redelivery after restart, webhook + poller). -The Fava-side duplicate checks are read-then-write races; the local -`processed_payments` primary key makes exactly one claimant win. - -These tests bypass invoice generation (blocked by libra/issues/40) by -delivering synthetic paid `Payment` objects straight to -`on_invoice_paid` and by inserting paid payment rows via the LNbits -core crud for the endpoint tests. -""" -import asyncio -import importlib -from uuid import uuid4 - -import pytest - -from lnbits.core.crud.payments import create_payment -from lnbits.core.models.payments import CreatePayment, Payment, PaymentState - -from .helpers import list_user_entries, post_receivable - -pytestmark = pytest.mark.anyio - - -def _module(name: str): - """Import a libra submodule under whichever path the active LNbits layout - uses (default `lnbits.extensions.libra` or bare `libra`).""" - for prefix in ("lnbits.extensions.libra", "libra"): - try: - return importlib.import_module(f"{prefix}.{name}") - except ModuleNotFoundError: - continue - raise ModuleNotFoundError(f"libra.{name}: tried both import paths") - - -tasks = _module("tasks") -libra_crud = _module("crud") - - -def _paid_payment( - wallet_id: str, - user_id: str, - *, - fiat_amount: str = "100.00", - fiat_currency: str = "EUR", - sats: int = 100_000, -) -> Payment: - payment_hash = uuid4().hex + uuid4().hex[:32] - return Payment( - checking_id=payment_hash, - payment_hash=payment_hash, - wallet_id=wallet_id, - amount=sats * 1000, - fee=0, - bolt11="lnbcfake", - status=PaymentState.SUCCESS, - extra={ - "tag": "libra", - "user_id": user_id, - "fiat_currency": fiat_currency, - "fiat_amount": fiat_amount, - }, - ) - - -async def _setup_receivable( - client, super_user_headers, configured_user, standard_accounts, - amount: str = "100.00", -): - user, wallet = configured_user - await post_receivable( - client, - super_user_headers=super_user_headers, - user_id=user.id, - amount=amount, - description=f"Idempotency setup {uuid4().hex[:6]}", - revenue_account=standard_accounts["revenue_rent"]["name"], - ) - # Force a Fava reload before downstream balance reads (see #37). - await list_user_entries(client, wallet_inkey=wallet.inkey) - return user, wallet - - -async def _entries_with_link(client, wallet_inkey: str, link: str) -> list: - payload = await list_user_entries(client, wallet_inkey=wallet_inkey) - return [ - e for e in payload["entries"] if link in (e.get("links") or []) - ] - - -# --------------------------------------------------------------------------- -# on_invoice_paid — the background listener path -# --------------------------------------------------------------------------- - - -async def test_double_delivery_records_exactly_once( - client, super_user_headers, configured_user, standard_accounts -): - """Same payment delivered twice (queue redelivery) → one ledger entry.""" - user, wallet = await _setup_receivable( - client, super_user_headers, configured_user, standard_accounts - ) - payment = _paid_payment(wallet.id, user.id) - - await tasks.on_invoice_paid(payment) - await tasks.on_invoice_paid(payment) - - link = f"ln-{payment.payment_hash[:16]}" - assert len(await _entries_with_link(client, wallet.inkey, link)) == 1 - - row = await libra_crud.get_processed_payment(payment.payment_hash) - assert row is not None and row["status"] == "done" - - -async def test_failed_recording_releases_claim_and_retry_succeeds( - client, super_user_headers, configured_user, standard_accounts, monkeypatch -): - """A Fava failure mid-write must not permanently block the payment.""" - user, wallet = await _setup_receivable( - client, super_user_headers, configured_user, standard_accounts - ) - payment = _paid_payment(wallet.id, user.id) - - fava_client = _module("fava_client") - fava = fava_client.get_fava_client() - - async def _boom(*args, **kwargs): - raise RuntimeError("fava down") - - monkeypatch.setattr(fava, "add_entry_idempotent", _boom) - with pytest.raises(RuntimeError): - await tasks.on_invoice_paid(payment) - monkeypatch.undo() - - # Claim released → nothing recorded, retry allowed. - assert await libra_crud.get_processed_payment(payment.payment_hash) is None - - await tasks.on_invoice_paid(payment) - row = await libra_crud.get_processed_payment(payment.payment_hash) - assert row is not None and row["status"] == "done" - link = f"ln-{payment.payment_hash[:16]}" - assert len(await _entries_with_link(client, wallet.inkey, link)) == 1 - - -async def test_listener_survives_poison_payment_and_clears_stale_claims( - client, super_user_headers, configured_user, standard_accounts, monkeypatch -): - """One bad payment must not kill the listener; stale 'processing' - claims from a previous process life are cleared at startup.""" - user, wallet = await _setup_receivable( - client, super_user_headers, configured_user, standard_accounts - ) - - # A claim left behind by a "crashed" previous run. - stale_hash = uuid4().hex + uuid4().hex[:32] - assert await libra_crud.claim_payment(stale_hash) - - captured: dict = {} - monkeypatch.setattr( - tasks, - "register_invoice_listener", - lambda queue, name: captured.update(queue=queue), - ) - - listener = asyncio.create_task(tasks.wait_for_paid_invoices()) - try: - for _ in range(50): - if "queue" in captured: - break - await asyncio.sleep(0.05) - assert "queue" in captured, "listener never registered its queue" - - poison = _paid_payment(wallet.id, user.id, fiat_amount="not-a-number") - good = _paid_payment(wallet.id, user.id) - captured["queue"].put_nowait(poison) - captured["queue"].put_nowait(good) - - row = None - for _ in range(100): - row = await libra_crud.get_processed_payment(good.payment_hash) - if row and row["status"] == "done": - break - await asyncio.sleep(0.1) - assert row is not None and row["status"] == "done", ( - "good payment was not recorded after the poison payment" - ) - finally: - listener.cancel() - - # Startup cleared the stale claim; the poison payment's claim was - # released on failure so redelivery could retry it. - assert await libra_crud.get_processed_payment(stale_hash) is None - assert await libra_crud.get_processed_payment(poison.payment_hash) is None - - -# --------------------------------------------------------------------------- -# POST /record-payment — the client-driven path -# --------------------------------------------------------------------------- - - -async def _insert_paid_payment_row(wallet_id: str, user_id: str) -> str: - payment_hash = uuid4().hex + uuid4().hex[:32] - await create_payment( - checking_id=payment_hash, - data=CreatePayment( - wallet_id=wallet_id, - payment_hash=payment_hash, - bolt11="lnbcfake", - amount_msat=100_000_000, - memo="idempotency test", - extra={ - "tag": "libra", - "user_id": user_id, - "fiat_currency": "EUR", - "fiat_amount": "100.00", - }, - ), - status=PaymentState.SUCCESS, - ) - return payment_hash - - -async def test_record_payment_conflicts_while_in_flight( - client, super_user_headers, configured_user, standard_accounts -): - user, wallet = await _setup_receivable( - client, super_user_headers, configured_user, standard_accounts - ) - payment_hash = await _insert_paid_payment_row(wallet.id, user.id) - - # Another claimant (e.g. the background listener) is mid-recording. - assert await libra_crud.claim_payment(payment_hash) - - r = await client.post( - "/libra/api/v1/record-payment", - headers={"X-Api-Key": wallet.inkey}, - json={"payment_hash": payment_hash}, - ) - assert r.status_code == 409, r.text - - # Once that claimant finishes, a replay reports "already recorded" - # instead of writing a second entry. - await libra_crud.mark_payment_done(payment_hash, f"ln-{payment_hash[:16]}") - r = await client.post( - "/libra/api/v1/record-payment", - headers={"X-Api-Key": wallet.inkey}, - json={"payment_hash": payment_hash}, - ) - assert r.status_code == 200, r.text - assert "already recorded" in r.json()["message"].lower() - - -async def test_record_payment_records_once_then_replays_safely( - client, super_user_headers, configured_user, standard_accounts -): - user, wallet = await _setup_receivable( - client, super_user_headers, configured_user, standard_accounts - ) - payment_hash = await _insert_paid_payment_row(wallet.id, user.id) - - r = await client.post( - "/libra/api/v1/record-payment", - headers={"X-Api-Key": wallet.inkey}, - json={"payment_hash": payment_hash}, - ) - assert r.status_code == 200, r.text - assert r.json()["message"] == "Payment recorded successfully" - - r = await client.post( - "/libra/api/v1/record-payment", - headers={"X-Api-Key": wallet.inkey}, - json={"payment_hash": payment_hash}, - ) - assert r.status_code == 200, r.text - assert "already recorded" in r.json()["message"].lower() - - link = f"ln-{payment_hash[:16]}" - assert len(await _entries_with_link(client, wallet.inkey, link)) == 1 diff --git a/tests/test_reconciliation_api.py b/tests/test_reconciliation_api.py index a168ff8..66757be 100644 --- a/tests/test_reconciliation_api.py +++ b/tests/test_reconciliation_api.py @@ -18,6 +18,19 @@ 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) # --------------------------------------------------------------------------- @@ -45,6 +58,7 @@ async def _create_assertion( # --------------------------------------------------------------------------- +@ASSERTION_CREATE_BROKEN @pytest.mark.anyio async def test_assertion_against_empty_account_passes( client, super_user_headers, standard_accounts, @@ -65,6 +79,7 @@ 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, @@ -87,6 +102,7 @@ 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, @@ -103,6 +119,7 @@ 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, @@ -128,6 +145,7 @@ 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, @@ -149,6 +167,7 @@ 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, @@ -171,6 +190,7 @@ 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 f7de090..442a01e 100644 --- a/tests/test_settlement_api.py +++ b/tests/test_settlement_api.py @@ -10,7 +10,6 @@ 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 @@ -269,12 +268,10 @@ 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}" - # 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") + 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 @pytest.mark.anyio diff --git a/tests/test_unit.py b/tests/test_unit.py index 11779e9..1c7dabc 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -553,6 +553,11 @@ 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"}) @@ -565,111 +570,3 @@ 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 0f2b817..66e2180 100644 --- a/tests/test_void_reject_api.py +++ b/tests/test_void_reject_api.py @@ -210,69 +210,3 @@ 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 1c3269f..1b3149a 100644 --- a/views_api.py +++ b/views_api.py @@ -13,7 +13,6 @@ 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, @@ -1075,7 +1074,8 @@ 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_metadata(amount_sats, data.amount), + "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, } # Get or create expense account @@ -1272,7 +1272,8 @@ async def api_create_income_entry( metadata = { "fiat_currency": fiat_currency, "fiat_amount": str(data.amount.quantize(Decimal("0.001"))), - **fiat_rate_metadata(amount_sats, data.amount), + "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, } # Submit to Fava @@ -1370,7 +1371,8 @@ 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_metadata(amount_sats, data.amount), + "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, } # Get or create revenue account @@ -1758,10 +1760,15 @@ 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_metadata(data.amount, invoice_fiat_amount), + "fiat_rate": fiat_rate, + "btc_rate": btc_rate, }) logger.info(f"Invoice extra metadata: {invoice_extra}") @@ -1843,130 +1850,90 @@ async def api_record_payment( try: async with httpx.AsyncClient(timeout=5.0) as client: - # Get recent entries from Fava's journal endpoint. base_url - # already ends in /api — the previous "/api/journal" path - # 404'd, so this duplicate check silently never ran. + # Get recent entries from Fava's journal endpoint response = await client.get( - f"{fava.base_url}/journal", + f"{fava.base_url}/api/journal", params={"time": ""} # Get all entries ) - response.raise_for_status() - response_data = response.json() - entries = response_data.get('entries', []) - # Check if any entry has our payment link - for entry in entries: - entry_links = entry.get('links', []) - if link_to_find in entry_links: - # Payment already recorded, return existing entry - balance_data = await fava.get_user_balance_bql(target_user_id) - return { - "journal_entry_id": f"fava-exists-{data.payment_hash[:16]}", - "new_balance": balance_data["balance"], - "message": "Payment already recorded", - } - except httpx.HTTPError as e: - # Fail CLOSED: if Fava can't confirm the payment isn't already - # recorded, refuse to write — proceeding on a transient blip is - # how double entries happen. The client can simply retry. + if response.status_code == 200: + response_data = response.json() + entries = response_data.get('entries', []) + + # Check if any entry has our payment link + for entry in entries: + entry_links = entry.get('links', []) + if link_to_find in entry_links: + # Payment already recorded, return existing entry + balance_data = await fava.get_user_balance_bql(target_user_id) + return { + "journal_entry_id": f"fava-exists-{data.payment_hash[:16]}", + "new_balance": balance_data["balance"], + "message": "Payment already recorded", + } + except Exception as e: logger.warning(f"Could not check Fava for duplicate payment: {e}") - raise HTTPException( - status_code=HTTPStatus.SERVICE_UNAVAILABLE, - detail="Cannot verify payment duplicate status; try again shortly", - ) - - # Local idempotency gate shared with the background invoice listener - # (tasks.on_invoice_paid): exactly one claimant records a payment_hash. - from .crud import ( - claim_payment, - get_processed_payment, - mark_payment_done, - release_payment_claim, - ) - - if not await claim_payment(data.payment_hash): - existing = await get_processed_payment(data.payment_hash) - if existing and existing["status"] == "done": - balance_data = await fava.get_user_balance_bql(target_user_id) - return { - "journal_entry_id": existing.get("entry_id") - or f"fava-exists-{data.payment_hash[:16]}", - "new_balance": balance_data["balance"], - "message": "Payment already recorded", - } - raise HTTPException( - status_code=HTTPStatus.CONFLICT, - detail="Payment is being recorded; check balance shortly", - ) + # Continue anyway - Fava/Beancount will catch duplicate if it exists # Convert amount from millisatoshis to satoshis - try: - amount_sats = payment.amount // 1000 + amount_sats = payment.amount // 1000 - # Extract fiat metadata from invoice (if present) - fiat_currency = None - fiat_amount = None - if payment.extra and isinstance(payment.extra, dict): - logger.info(f"Payment.extra contents: {payment.extra}") - fiat_currency = payment.extra.get("fiat_currency") - fiat_amount_str = payment.extra.get("fiat_amount") - if fiat_amount_str: - from decimal import Decimal - fiat_amount = Decimal(str(fiat_amount_str)) + # Extract fiat metadata from invoice (if present) + fiat_currency = None + fiat_amount = None + if payment.extra and isinstance(payment.extra, dict): + logger.info(f"Payment.extra contents: {payment.extra}") + fiat_currency = payment.extra.get("fiat_currency") + fiat_amount_str = payment.extra.get("fiat_amount") + if fiat_amount_str: + from decimal import Decimal + fiat_amount = Decimal(str(fiat_amount_str)) - logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}") + logger.info(f"Extracted fiat metadata - currency: {fiat_currency}, amount: {fiat_amount}") - # Get user's receivable account (what user owes) - user_receivable = await get_or_create_user_account( - target_user_id, AccountType.ASSET, "Accounts Receivable" + # Get user's receivable account (what user owes) + user_receivable = await get_or_create_user_account( + target_user_id, AccountType.ASSET, "Accounts Receivable" + ) + + # Get lightning account + lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning") + if not lightning_account: + raise HTTPException( + status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found" ) - # Get lightning account - lightning_account = await get_account_by_name("Assets:Bitcoin:Lightning") - if not lightning_account: - raise HTTPException( - status_code=HTTPStatus.NOT_FOUND, detail="Lightning account not found" - ) + # Get unsettled receivable entries to link to this settlement + unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable") + settled_links = [e["link"] for e in unsettled if e.get("link")] - # Get unsettled receivable entries to link to this settlement - unsettled = await fava.get_unsettled_entries_bql(target_user_id, "receivable") - settled_links = [e["link"] for e in unsettled if e.get("link")] + # Format payment entry and submit to Fava + entry = format_payment_entry( + user_id=target_user_id, + payment_account=lightning_account.name, + payable_or_receivable_account=user_receivable.name, + amount_sats=amount_sats, + description=f"Lightning payment from user {target_user_id[:8]}", + entry_date=datetime.now().date(), + is_payable=False, # User paying libra (receivable settlement) + fiat_currency=fiat_currency, + fiat_amount=fiat_amount, + payment_hash=data.payment_hash, + reference=data.payment_hash, + settled_entry_links=settled_links + ) - # Format payment entry and submit to Fava - entry = format_payment_entry( - user_id=target_user_id, - payment_account=lightning_account.name, - payable_or_receivable_account=user_receivable.name, - amount_sats=amount_sats, - description=f"Lightning payment from user {target_user_id[:8]}", - entry_date=datetime.now().date(), - is_payable=False, # User paying libra (receivable settlement) - fiat_currency=fiat_currency, - fiat_amount=fiat_amount, - payment_hash=data.payment_hash, - reference=data.payment_hash, - settled_entry_links=settled_links - ) + logger.info(f"Formatted payment entry: {entry}") - logger.info(f"Formatted payment entry: {entry}") - - # Submit to Fava - result = await fava.add_entry(entry) - logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}") - - entry_id = f"ln-{data.payment_hash[:16]}" - await mark_payment_done(data.payment_hash, entry_id) - except BaseException: - # Release the claim so a retry (client or background listener) - # can record this payment. - await release_payment_claim(data.payment_hash) - raise + # Submit to Fava + result = await fava.add_entry(entry) + logger.info(f"Payment entry submitted to Fava: {result.get('data', 'Unknown')}") # Get updated balance from Fava balance_data = await fava.get_user_balance_bql(target_user_id) return { - "journal_entry_id": entry_id, + "journal_entry_id": f"fava-{datetime.now().timestamp()}", "new_balance": balance_data["balance"], "message": "Payment recorded successfully", } @@ -2057,19 +2024,6 @@ 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), @@ -2113,10 +2067,10 @@ async def api_settle_receivable( "net to clear all open entries, or pass " "`settled_entry_links` to settle a specific subset." ), - "cash_paid": str(cash_paid), - "net_obligation": str(net_obligation), - "receivable_total": str(receivable_total), - "payable_total": str(payable_total), + "cash_paid": float(cash_paid), + "net_obligation": float(net_obligation), + "receivable_total": float(receivable_total), + "payable_total": float(payable_total), "currency": data.currency.upper(), }, ) @@ -2903,7 +2857,8 @@ async def api_approve_expense_entry( This updates the transaction in the Beancount file via Fava API. """ - from .fava_client import ChecksumConflictError, get_fava_client + import httpx + from .fava_client import get_fava_client fava = get_fava_client() @@ -2938,29 +2893,57 @@ async def api_approve_expense_entry( detail="Entry metadata missing filename or lineno" ) - # 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} !" + # 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"] - def _approve(line: str) -> str: - if old_pattern not in line: + 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): raise HTTPException( status_code=HTTPStatus.INTERNAL_SERVER_ERROR, - detail=f"Line {lineno} does not contain expected pattern '{old_pattern}'. Found: {line}" + detail=f"Line {lineno} not found in source file" ) - return line.replace(old_pattern, f"{date_str} *", 1) - 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", + 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"} ) + update_response.raise_for_status() logger.info(f"Entry {entry_id} approved (flag changed to *)") @@ -2983,7 +2966,8 @@ 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. """ - from .fava_client import ChecksumConflictError, get_fava_client + import httpx + from .fava_client import get_fava_client fava = get_fava_client() @@ -3018,26 +3002,50 @@ async def api_reject_expense_entry( detail="Entry metadata missing filename or lineno" ) - # 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' + # 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"] - 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) - ) - 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)") + 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)") return { "message": f"Entry {entry_id} rejected (marked as voided)", @@ -3766,14 +3774,8 @@ async def api_admin_add_chart_account( "already_existed": True, } - # 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, - ) + # Mirror into libra DB so permissions / metadata layer sees it. + synced = await sync_single_account_from_beancount(payload.name) return { "success": True,