diff --git a/blueprints/ynab/sync.py b/blueprints/ynab/sync.py index a551e0c..3ddb27e 100644 --- a/blueprints/ynab/sync.py +++ b/blueprints/ynab/sync.py @@ -53,8 +53,15 @@ def ynab_configured() -> bool: return bool(os.getenv("YNAB_ACCESS_TOKEN")) -def _configured_budget_id() -> str: - return os.getenv("YNAB_BUDGET_ID") or "last-used" +# YNAB accepts these in place of a budget id; they can point at different +# budgets over time, so the mirror resolves them once and keeps the real id +BUDGET_ID_ALIASES = ("last-used", "default") + + +def _configured_budget_id() -> str | None: + """YNAB_BUDGET_ID if it names a specific budget, else None.""" + budget_id = os.getenv("YNAB_BUDGET_ID", "").strip() + return None if budget_id in ("", *BUDGET_ID_ALIASES) else budget_id def _money(milliunits: int | None) -> Decimal | None: @@ -266,6 +273,17 @@ def _budget_entities(budget: dict) -> dict[str, list[dict]]: } +# Id fields that reference a named entity (account, payee, category, group); +# the change log records these by name +NAMED_ID_FIELDS = ( + "account_id", + "payee_id", + "category_id", + "category_group_id", + "transfer_account_id", +) + + class _Labeler: """Resolves ids to names for change log labels and diffs.""" @@ -289,9 +307,9 @@ class _Labeler: return row.get("name") or row["id"] def diff_entry(self, field: str, old: Any, new: Any) -> tuple[str, list]: - """Log ``category_id`` changes as ``category`` names, not opaque ids.""" - if field.endswith("_id"): - return field[:-3], [self.name(old), self.name(new)] + """Log e.g. ``category_id`` changes as ``category`` names, not ids.""" + if field in NAMED_ID_FIELDS: + return field.removesuffix("_id"), [self.name(old), self.name(new)] return field, [_jsonable(old), _jsonable(new)] @@ -480,23 +498,26 @@ async def apply_budget( async def _current_state() -> YnabSyncState | None: """The sync state for the mirrored budget. - The mirror holds a single budget. With ``YNAB_BUDGET_ID`` unset, the first - sync resolves YNAB's "last-used" alias and the mirror stays pinned to that - budget, so switching budgets in YNAB can't mix two budgets' rows. + The mirror holds a single budget, keyed by its real YNAB id. With + ``YNAB_BUDGET_ID`` unset or an alias, the first sync resolves "last-used" + and the mirror stays pinned to that budget, so switching budgets in YNAB + can't mix two budgets' rows. """ - configured = os.getenv("YNAB_BUDGET_ID") - states = await YnabSyncState.all() - if configured: - others = [s.budget_id for s in states if s.budget_id != configured] - if others: - raise ValueError( - f"YNAB mirror holds budget {others[0]} but YNAB_BUDGET_ID is " - f"{configured}; clear the ynab_* tables to switch budgets" - ) - return next((s for s in states if s.budget_id == configured), None) + configured = _configured_budget_id() + states = await YnabSyncState.all().order_by("budget_id") + ids = [s.budget_id for s in states] + if len(ids) > 1 or (configured and ids and ids[0] != configured): + raise ValueError( + f"YNAB mirror holds budget(s) {', '.join(ids)} but YNAB_BUDGET_ID " + f"is {configured or 'unset'}; clear the ynab_* tables to switch budgets" + ) return states[0] if states else None +def _backing_off() -> bool: + return time.monotonic() - _last_failure < FAILURE_BACKOFF_SECONDS + + def _is_fresh(state: YnabSyncState | None, max_age_seconds: int) -> bool: return ( state is not None @@ -509,8 +530,9 @@ def _is_fresh(state: YnabSyncState | None, max_age_seconds: int) -> bool: async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> dict: """Sync YNAB into Postgres. Returns per-entity counts of changed rows. - ``max_age_seconds`` skips the sync if the mirror is already that fresh, - checked under the lock so parallel tool calls share one sync. + ``max_age_seconds`` (on-demand syncs from tools) skips the sync if the + mirror is already that fresh or YNAB failed recently. Both are checked + under the lock, so parallel tool calls share one sync or one failure. The YNAB request runs before the DB transaction. An asyncio lock serializes syncs in this process; a Postgres advisory lock plus a @@ -520,7 +542,9 @@ async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> d global _last_failure async with _sync_lock: state = await _current_state() - if max_age_seconds is not None and _is_fresh(state, max_age_seconds): + if max_age_seconds is not None and ( + _is_fresh(state, max_age_seconds) or _backing_off() + ): return {"skipped": True} now = datetime.now(timezone.utc) @@ -533,6 +557,7 @@ async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> d or now - _norm(state.last_full_sync_at) > FULL_RESYNC_INTERVAL ) budget_id = state.budget_id if state else _configured_budget_id() + budget_id = budget_id or "last-used" try: data = await YNABService().fetch_budget( @@ -540,6 +565,10 @@ async def sync_ynab(full: bool = False, max_age_seconds: int | None = None) -> d ) budget = data["budget"] knowledge = data["server_knowledge"] + if state is not None and budget["id"] != state.budget_id: + raise ValueError( + f"YNAB returned budget {budget['id']} for {state.budget_id}" + ) async with in_transaction() as conn: await conn.execute_query( "SELECT pg_advisory_xact_lock(hashtext('ynab_sync'))" @@ -588,8 +617,7 @@ async def ensure_fresh(max_age_seconds: int = 120) -> YnabSyncState | None: been populated. """ state = await _current_state() - backing_off = time.monotonic() - _last_failure < FAILURE_BACKOFF_SECONDS - if not _is_fresh(state, max_age_seconds) and not backing_off: + if not _is_fresh(state, max_age_seconds) and not _backing_off(): try: await sync_ynab(max_age_seconds=max_age_seconds) except Exception: diff --git a/blueprints/ynab/tools.py b/blueprints/ynab/tools.py index 7fa5ee3..86b60ee 100644 --- a/blueprints/ynab/tools.py +++ b/blueprints/ynab/tools.py @@ -33,6 +33,7 @@ ENTITY_TYPE_ALIASES = { } ENTITY_TYPES = ( "transaction", + "month", "subtransaction", "month_category", "category", @@ -42,6 +43,12 @@ ENTITY_TYPES = ( ) +def _singular(word: str) -> str: + if word.endswith("ies"): + return word[:-3] + "y" # categories -> category + return word.removesuffix("s") + + def ynab_tool_errors(fn): """Return errors to the model as text: LangGraph's ToolNode re-raises anything else, which would fail the whole chat turn.""" @@ -450,14 +457,14 @@ async def ynab_recent_changes(days: int = 7, entity_type: str = "") -> str: Args: days: Look back this many days (default 7) - entity_type: Optional filter: transaction, assignment, category, payee, account + entity_type: Optional filter: transaction, assignment, category, payee, account, month """ if await ensure_fresh() is None: return NOT_SYNCED - entity_type = entity_type.strip().lower().removesuffix("s") + entity_type = _singular(entity_type.strip().lower().replace(" ", "_")) entity_type = ENTITY_TYPE_ALIASES.get(entity_type, entity_type) if entity_type and entity_type not in ENTITY_TYPES: - return f"Unknown entity_type '{entity_type}'. Use one of: transaction, assignment, category, category_group, payee, account." + return f"Unknown entity_type '{entity_type}'. Use one of: transaction, assignment, category, category_group, payee, account, month." rows = await _query( """SELECT synced_at, entity_type, action, label, changes FROM ynab_changes WHERE synced_at >= now() - make_interval(days => $1) diff --git a/tests/unit/test_ynab_mirror.py b/tests/unit/test_ynab_mirror.py index cd19517..f988fc5 100644 --- a/tests/unit/test_ynab_mirror.py +++ b/tests/unit/test_ynab_mirror.py @@ -1,8 +1,10 @@ """Tests for the YNAB Postgres mirror: delta application, change log, SQL guard.""" +import asyncio from decimal import Decimal import pytest +import sqlglot from tortoise import Tortoise from tortoise.transactions import in_transaction @@ -248,7 +250,12 @@ def test_executed_sql_drops_comments_and_semicolon(): ) assert error is None assert "--" not in sql and not sql.endswith(";") - assert f"SELECT * FROM ({sql}) AS q" # wraps cleanly + # The LIMIT wrapper must still parse: a leftover line comment would + # swallow its closing parenthesis + wrapped = sqlglot.parse_one( + f"SELECT * FROM ({sql}) AS q LIMIT 100", read="postgres" + ) + assert wrapped.args["limit"] is not None async def test_full_sync_marks_missing_rows_deleted(db): @@ -328,8 +335,14 @@ async def test_recent_changes_accepts_display_names(monkeypatch): monkeypatch.setattr(tools, "ensure_fresh", fresh) monkeypatch.setattr(tools, "_query", fake_query) - await tools.ynab_recent_changes.ainvoke({"entity_type": "Assignments"}) - assert seen[-1][1] == "month_category" + for given, stored in [ + ("Assignments", "month_category"), + ("categories", "category"), + ("category groups", "category_group"), + ("month", "month"), + ]: + await tools.ynab_recent_changes.ainvoke({"entity_type": given}) + assert seen[-1][1] == stored result = await tools.ynab_recent_changes.ainvoke({"entity_type": "bogus"}) assert result.startswith("Unknown entity_type") @@ -345,3 +358,77 @@ async def test_budget_summary_rejects_bad_month(monkeypatch): assert await tools.ynab_budget_summary.ainvoke({"month": month}) == ( "Month must be YYYY-MM." ) + + +async def test_subtransaction_parent_id_is_not_renamed(db): + split = { + "id": "sub-1", + "transaction_id": "txn-1", + "amount": -10_000, + "memo": None, + "payee_id": None, + "category_id": "cat-1", + "transfer_account_id": None, + "deleted": False, + } + budget = _budget([_txn()]) + budget["subtransactions"] = [split] + await _apply(budget, 1, log_changes=False) + budget["subtransactions"] = [{**split, "amount": -12_000}] + await _apply(budget, 2) + change = await YnabChange.get(entity_type="subtransaction") + assert change.changes == {"amount": [-10.0, -12.0]} + + +class TestBudgetPinning: + @pytest.fixture(autouse=True) + async def _db(self, db): + yield + + async def _state(self, budget_id): + from blueprints.ynab.models import YnabSyncState + + await YnabSyncState.create(budget_id=budget_id, server_knowledge=1) + + @pytest.mark.parametrize("configured", ["", "last-used", "default"]) + async def test_alias_or_unset_uses_stored_budget(self, monkeypatch, configured): + from blueprints.ynab.sync import _current_state + + monkeypatch.setenv("YNAB_BUDGET_ID", configured) + await self._state("uuid-a") + assert (await _current_state()).budget_id == "uuid-a" + + async def test_different_configured_budget_is_an_error(self, monkeypatch): + from blueprints.ynab.sync import _current_state + + monkeypatch.setenv("YNAB_BUDGET_ID", "uuid-b") + await self._state("uuid-a") + with pytest.raises(ValueError, match="clear the ynab_"): + await _current_state() + + async def test_multiple_stored_budgets_is_an_error(self, monkeypatch): + from blueprints.ynab.sync import _current_state + + monkeypatch.delenv("YNAB_BUDGET_ID", raising=False) + await self._state("uuid-a") + await self._state("uuid-b") + with pytest.raises(ValueError): + await _current_state() + + +async def test_parallel_syncs_share_one_failure(db, monkeypatch): + from blueprints.ynab import sync + + calls = [] + + class FailingYNAB: + async def fetch_budget(self, budget_id, since): + calls.append(budget_id) + raise TimeoutError("ynab down") + + monkeypatch.setattr(sync, "YNABService", FailingYNAB) + monkeypatch.setattr(sync, "_last_failure", float("-inf")) + await asyncio.gather( + *(sync.ensure_fresh() for _ in range(3)), return_exceptions=True + ) + assert len(calls) == 1