diff --git a/routstr/core/ledger_analytics.py b/routstr/core/ledger_analytics.py index 6308473c..949c564c 100644 --- a/routstr/core/ledger_analytics.py +++ b/routstr/core/ledger_analytics.py @@ -4,7 +4,7 @@ from copy import deepcopy from datetime import UTC, datetime, timedelta from typing import Any -from sqlalchemy import case +from sqlalchemy import case, text from sqlmodel import col, func, select from . import terminal_outcomes @@ -158,6 +158,13 @@ async def get_ledger_usage_dashboard( + col(TerminalOutcome.cache_creation_input_tokens) ) async with session_factory() as session: + # Keep all aggregates on one snapshot while queued outcomes are flushed. + if session.get_bind().dialect.name == "sqlite": + await session.execute(text("BEGIN")) + else: + await session.connection( + execution_options={"isolation_level": "REPEATABLE READ"} + ) bucket_rows = ( await session.exec( select(bucket, *measures) diff --git a/routstr/nostr/analytics_runtime.py b/routstr/nostr/analytics_runtime.py index 0b7c507c..c8c0abd7 100644 --- a/routstr/nostr/analytics_runtime.py +++ b/routstr/nostr/analytics_runtime.py @@ -4,14 +4,15 @@ import asyncio import time from datetime import UTC, datetime -from ..core.db import create_session +from ..core.db import NsecState, Secret, create_session from ..core.logging import get_logger -from ..core.settings import SettingsService, settings +from ..core.settings import SettingsService, derive_npub_from_nsec, settings from ..core.terminal_outcomes import ( start_terminal_outcome_writer, stop_terminal_outcome_writer, terminal_outcome_writer, ) +from ..core.vault import decrypt from .analytics_v2_delivery import ( AnalyticsV2Delivery, AnalyticsV2Producer, @@ -34,6 +35,15 @@ async def _read_state() -> DeliveryStateSnapshot: state = await get_analytics_v2_delivery_state(create_session) async with create_session() as session: await SettingsService.refresh(session, ("enable_analytics_sharing",)) + secret = await session.get(Secret, 1) + if secret is not None and secret.nsec_state != NsecState.legacy: + if secret.nsec_state == NsecState.encrypted: + if secret.encrypted_nsec is None: + raise RuntimeError("Encrypted Nostr identity has no ciphertext") + settings.nsec = decrypt(secret.encrypted_nsec) + else: + settings.nsec = "" + settings.npub = derive_npub_from_nsec(settings.nsec) or "" return state diff --git a/tests/unit/test_analytics_runtime.py b/tests/unit/test_analytics_runtime.py index 0da4229c..3a85bbd2 100644 --- a/tests/unit/test_analytics_runtime.py +++ b/tests/unit/test_analytics_runtime.py @@ -1,7 +1,9 @@ from __future__ import annotations import asyncio +import importlib import json +import time from collections.abc import AsyncIterator from contextlib import asynccontextmanager from datetime import UTC, datetime, timedelta @@ -10,14 +12,17 @@ from typing import Any import pytest import pytest_asyncio +from httpx import ASGITransport, AsyncClient from sqlalchemy.ext.asyncio import create_async_engine from sqlmodel import SQLModel, col, select, text from sqlmodel.ext.asyncio.session import AsyncSession -from routstr.core import terminal_outcomes -from routstr.core.db import TerminalOutcomeEpoch -from routstr.core.settings import SettingsService +from routstr.core import admin, terminal_outcomes, vault +from routstr.core.db import Secret, TerminalOutcomeEpoch +from routstr.core.main import app +from routstr.core.settings import Settings, SettingsService from routstr.nostr import analytics_runtime as runtime +from routstr.nostr import listing @pytest_asyncio.fixture @@ -165,6 +170,79 @@ async def test_identity_rotation_cancels_previous_publisher( assert node.writer.running +@pytest.mark.asyncio +async def test_saved_identity_reaches_other_workers_without_disabling_sharing( + node: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + settings_module = importlib.import_module("routstr.core.settings") + first = runtime.settings.copy(deep=True) + first.nsec = "" + first.npub = "" + first.enable_analytics_sharing = True + second = first.copy(deep=True) + + def bind_worker(settings: Settings) -> None: + for module in (runtime, admin, listing, settings_module): + monkeypatch.setattr(module, "settings", settings) + monkeypatch.setattr(SettingsService, "_current", settings) + + async with node.sessions() as session: + await session.exec( # type: ignore[call-overload] + text("INSERT INTO settings (id, data) VALUES (1, :data)").bindparams( + data=json.dumps(settings_module._strip_secret_fields(first.dict())) + ) + ) + await session.commit() + monkeypatch.setattr(admin, "create_session", node.sessions) + token = "stats-identity-test" + monkeypatch.setitem(admin.admin_sessions, token, int(time.time()) + 60) + other = runtime.AnalyticsCoordinator() + try: + for coordinator, settings in ((node.coordinator, first), (other, second)): + bind_worker(settings) + await coordinator.prepare_startup() + await coordinator.sync_once() + async with AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + headers={"Authorization": f"Bearer {token}"}, + ) as client: + for key in ("11" * 32, "22" * 32, ""): + bind_worker(first) + response = await client.patch("/admin/api/nsec", json={"nsec": key}) + assert response.status_code == 200 + saved = await client.patch( + "/admin/api/settings", json={"npub": response.json()["npub"]} + ) + assert saved.status_code == 200 + async with node.sessions() as session: + secret = await session.get(Secret, 1) + assert secret is not None + assert ( + vault.decrypt(secret.encrypted_nsec) + if secret.encrypted_nsec + else "" + ) == key + await node.coordinator.sync_once() + state = await runtime.get_analytics_v2_delivery_state(node.sessions) + for coordinator, settings in ( + (other, second), + (node.coordinator, first), + (other, second), + ): + bind_worker(settings) + await coordinator.sync_once() + current = await runtime.get_analytics_v2_delivery_state( + node.sessions + ) + assert current.sharing_enabled is bool(key) + assert current.generation == state.generation + assert settings.nsec == key + assert settings.npub == response.json()["npub"] + finally: + await other.close() + + @pytest.mark.asyncio async def test_restart_writer_failure_disables_publication_before_serving( node: Any, monkeypatch: Any diff --git a/tests/unit/test_ledger_analytics.py b/tests/unit/test_ledger_analytics.py index 1a08250d..a8c788ab 100644 --- a/tests/unit/test_ledger_analytics.py +++ b/tests/unit/test_ledger_analytics.py @@ -5,10 +5,11 @@ from contextlib import asynccontextmanager from copy import deepcopy from datetime import UTC, date, datetime, timedelta from pathlib import Path +from typing import Any import pytest from sqlalchemy.ext.asyncio import create_async_engine -from sqlmodel import SQLModel +from sqlmodel import SQLModel, text from sqlmodel.ext.asyncio.session import AsyncSession from routstr.core.db import ( @@ -225,6 +226,58 @@ async def test_dashboard_replaces_log_totals_and_matches_every_chart( } +@pytest.mark.parametrize("new_bucket", [False, True]) +async def test_dashboard_keeps_one_snapshot_while_queued_outcomes_flush( + sessions: SessionFactory, + monkeypatch: pytest.MonkeyPatch, + new_bucket: bool, +) -> None: + async with sessions() as session: + await session.execute(text("PRAGMA journal_mode=WAL")) + await _seed(sessions) + original_exec = AsyncSession.exec + flushed = False + + async def exec_then_flush( + session: AsyncSession, statement: Any, **kwargs: Any + ) -> Any: + nonlocal flushed + result = await original_exec(session, statement, **kwargs) + if not flushed: + flushed = True + async with sessions() as writer: + writer.add( + _row( + "queued", + NOW - timedelta(minutes=1 if new_bucket else 240), + ) + ) + await writer.commit() + return result + + monkeypatch.setattr(AsyncSession, "exec", exec_then_flush) + result = await get_ledger_usage_dashboard( + _legacy(), interval=60, hours=24, session_factory=sessions, now=NOW + ) + assert flushed + assert result["summary"]["successful_chat_completions"] == 3 + assert result["summary"]["revenue_msats"] == 1250 + assert ( + sum(model["successful"] for model in result["revenue_by_model"]["models"]) == 3 + ) + for point in result["model_usage_mix"]["metrics"]: + assert point["others"] >= 0 + assert point["others_revenue_msats"] >= 0 + assert ( + sum(point["model_counts"].values()) + point["others"] + == point["total_successful"] + ) + assert ( + sum(point["model_revenue_msats"].values()) + point["others_revenue_msats"] + == point["total_revenue_msats"] + ) + + async def test_historical_dates_do_not_return_recent_ledger_or_log_data( sessions: SessionFactory, ) -> None: