diff --git a/.gitignore b/.gitignore index d527b495..469e4d90 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,9 @@ __pycache__/ settings.toml settings.local.toml +# Claude Code runtime artifacts (the rest of .claude/ is tracked) +.claude/scheduled_tasks.lock + # Git worktrees/ diff --git a/nextcloud_mcp_server/alembic/versions/20260607_1200_007_add_usage_events.py b/nextcloud_mcp_server/alembic/versions/20260607_1200_007_add_usage_events.py new file mode 100644 index 00000000..6ce5bf32 --- /dev/null +++ b/nextcloud_mcp_server/alembic/versions/20260607_1200_007_add_usage_events.py @@ -0,0 +1,85 @@ +"""Add usage_events table for per-tenant usage metering. + +Deck #67 / control-plane usage-metering.md (pull model). Each tenant Pod +records billable usage (embedding queries, pages/chunks embedded) into its own +app DB; the control plane later pulls this table read-only into the billing +ledger and syncs to Stripe Meter Events. Writes are gated by +``USAGE_METERING_ENABLED`` (default off) — the recording hook is a no-op when +the flag is off, so an OSS self-hoster gets an empty table and zero write +overhead. + +Unlike the rest of this schema (unix-epoch ``BigInteger`` timestamps, JSON as +``Text``), this table uses real Postgres ``TIMESTAMPTZ``/``JSONB``/``UUID`` +because the control-plane rollup runs ``date_trunc('day', occurred_at AT TIME +ZONE 'UTC')`` and ``GROUP BY day, metric`` directly against Postgres, which +requires a genuine timestamptz column. SQLite (OSS/tests) uses portable +fallbacks (``TEXT``/``TIMESTAMP``); the control plane never queries SQLite. + +Revision ID: 007 +Revises: 006 +Create Date: 2026-06-07 12:00:00.000000 +""" + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +revision = "007" +down_revision = "006" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + is_pg = op.get_bind().dialect.name == "postgresql" + + # Retention: this table has no TTL by design — the control-plane rollup + # owns the lifecycle (it pulls rows read-only into usage_daily, then + # prunes once a day is reconciled). The data plane only appends. + op.create_table( + "usage_events", + # Pod-generated idempotency key. UUID on Postgres; TEXT on SQLite, + # which has no native UUID type. Stored/bound as a plain string in + # both backends (see usage/store.py). + sa.Column( + "event_id", + postgresql.UUID(as_uuid=False) if is_pg else sa.Text(), + primary_key=True, + ), + # Operation completion time (UTC). Real TIMESTAMPTZ on Postgres so the + # CP rollup's date_trunc(... AT TIME ZONE 'UTC') works; portable + # TIMESTAMP on SQLite (stored as ISO text, queryable in tests). + sa.Column( + "occurred_at", + postgresql.TIMESTAMP(timezone=True) if is_pg else sa.TIMESTAMP(), + nullable=False, + ), + # Catalog metric: 'embeddings_queries' or 'pages_chunks'. Deliberately + # an unconstrained Text (no CHECK/enum) — the metric catalog lives in + # control-plane config, not the app-DB schema. If a third metric is + # ever added, the CP-side catalog must learn it too, or its rollup will + # silently ignore the new rows; keep the two in sync. + sa.Column("metric", sa.Text(), nullable=False), + sa.Column("value", sa.BigInteger(), nullable=False), + # Rawest unit per request (provider, model, tokens, doc_type, ...). + # JSONB on Postgres so the CP can slice on dimensions later; TEXT + # (json.dumps) on SQLite. + sa.Column( + "metadata", + postgresql.JSONB() if is_pg else sa.Text(), + nullable=True, + ), + ) + # Serves the CP rollup's per-day range scan (occurred_at >= / <) plus the + # GROUP BY metric; leading occurred_at makes the range filter index-usable. + op.create_index( + "idx_usage_events_occurred_metric", + "usage_events", + ["occurred_at", "metric"], + ) + + +def downgrade() -> None: + op.drop_index("idx_usage_events_occurred_metric", table_name="usage_events") + op.drop_table("usage_events") diff --git a/nextcloud_mcp_server/auth/storage.py b/nextcloud_mcp_server/auth/storage.py index e6b95bac..e323a14b 100644 --- a/nextcloud_mcp_server/auth/storage.py +++ b/nextcloud_mcp_server/auth/storage.py @@ -39,7 +39,7 @@ import os import socket import sqlite3 import time -from contextlib import asynccontextmanager +from contextlib import AbstractAsyncContextManager, asynccontextmanager from pathlib import Path from typing import Any @@ -666,6 +666,21 @@ class RefreshTokenStorage: async with self.engine.connect() as conn: yield _DBConn(conn) + def acquire(self) -> AbstractAsyncContextManager["_DBConn"]: + """Public alias for :meth:`_db`: a backend-agnostic connection cm. + + Lets sibling stores (e.g. :class:`UsageEventStore`) reuse this + instance's engine, NullPool, and ``_DBConn`` shim without reaching + into the underscored internal. Use as ``async with storage.acquire() + as db:``. + """ + return self._db() + + @property + def dialect(self) -> str: + """Backend dialect name ("sqlite" / "postgresql"), or "unknown" pre-init.""" + return self._dialect + async def store_refresh_token( self, user_id: str, diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 8f298cba..b891df78 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -244,6 +244,11 @@ _DEFAULTS: dict[str, Any] = { # this before a real ACL backfill would silently drop legacy results. # verify-on-read remains the correctness backstop regardless. "acl_prefilter_enabled": False, + # Usage metering (Deck #67, control-plane usage-metering.md). OFF by + # default so OSS self-hosters don't accrue a metering table or write + # overhead; Astrolabe Cloud provisioning sets it true. When on, billable + # ops record rows into the app-DB usage_events table (best-effort). + "usage_metering_enabled": False, } @@ -829,6 +834,10 @@ class Settings: embedding_gateway_scope: str | None = None tenant_id: str | None = None # per-tenant identity (UUID form) acl_prefilter_enabled: bool = False # query-side ACL pre-filter (§11); OFF + # Usage metering (Deck #67); OFF by default. When true, billable ops + # record best-effort rows into the app-DB usage_events table for the + # control plane to pull. See nextcloud_mcp_server/usage/store.py. + usage_metering_enabled: bool = False def __post_init__(self): """Validate configuration and set defaults.""" @@ -1441,6 +1450,7 @@ def get_settings() -> Settings: "embedding_gateway_scope": "EMBEDDING_GATEWAY_SCOPE", "tenant_id": "TENANT_ID", "acl_prefilter_enabled": "ACL_PREFILTER_ENABLED", + "usage_metering_enabled": "USAGE_METERING_ENABLED", } # Only pass values that dynaconf actually has; omit unset keys so diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index b2b8ec0c..8d78849f 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -39,12 +39,22 @@ from nextcloud_mcp_server.search.access_filter import ( from nextcloud_mcp_server.search.bm25_hybrid import BM25HybridSearchAlgorithm from nextcloud_mcp_server.search.context import get_chunk_with_context from nextcloud_mcp_server.search.verification import verify_search_results +from nextcloud_mcp_server.usage import UsageEventStore from nextcloud_mcp_server.utils.validation import parse_modified_timestamp from nextcloud_mcp_server.vector.metrics_publisher import count_indexed from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client logger = logging.getLogger(__name__) +# Cap how many doc_types we copy into a usage-metering metadata row. doc_types +# is caller-supplied and (unlike path_prefixes) has no max_length on the tool +# signature, so an adversarial caller could pass a huge list. The CP rollup +# ignores metadata for billing (GROUP BY day, metric) and the value is bound +# parameterized, so this is not a billing/injection risk — the cap just keeps +# a single JSONB row from ballooning. 16 is generous headroom over the handful +# of real indexed doc types. +_USAGE_METADATA_MAX_DOC_TYPES = 16 + def configure_semantic_tools(mcp: FastMCP): """Configure semantic search tools for MCP server.""" @@ -517,6 +527,52 @@ def configure_semantic_tools(mcp: FastMCP): logger.info("Returning %d results from BM25 hybrid search", len(results)) + # Usage metering (Deck #67): one billable 'embeddings_queries' + # event per successful search (the query embedding is the metered + # cost). Best-effort and gated on the flag so the off-path touches + # no storage. nc_semantic_search_answer reuses this tool, so it + # records here too — do not add a second hook there. + # + # Privacy note: user_id stays tenant-local. The CP rollup + # aggregates GROUP BY (day, metric) into usage_daily, which has no + # metadata column, so nothing here propagates to Stripe; the value + # is retained only so Deck #67's "per-user attribution derivable + # from app-DB metadata later" stays possible without a re-migration. + if settings.usage_metering_enabled: + try: + store = await UsageEventStore.shared() + await store.record_usage_event( + metric="embeddings_queries", + value=1, + metadata={ + "user_id": username, + "fusion": fusion, + # Bounded copy — see _USAGE_METADATA_MAX_DOC_TYPES. + # Both None and [] normalize to null so a future + # metadata->'doc_types' IS NULL query counts the + # all-types case consistently. + "doc_types": ( + doc_types[:_USAGE_METADATA_MAX_DOC_TYPES] + if doc_types + else None + ), + }, + # The outer guard already confirmed the flag, so pass + # enabled=True directly — the store then skips a second + # uncached Settings build on this hot query path + # (ADR-024). + enabled=True, + ) + except Exception: + # Reached only when shared()/store construction itself + # raises (record_usage_event swallows its own write + # failures). Metering is on, so warn — a silent DEBUG line + # would hide "operator enabled metering but gets no data". + logger.warning( + "usage metering hook (embeddings_queries) skipped", + exc_info=True, + ) + return SemanticSearchResponse( results=results, query=query, diff --git a/nextcloud_mcp_server/usage/__init__.py b/nextcloud_mcp_server/usage/__init__.py new file mode 100644 index 00000000..dc127da6 --- /dev/null +++ b/nextcloud_mcp_server/usage/__init__.py @@ -0,0 +1,10 @@ +"""Per-tenant usage metering (data plane). + +Records billable operations into the app-DB ``usage_events`` table for the +control plane to pull. Gated by ``USAGE_METERING_ENABLED`` (default off). See +Deck #67 and control-plane ``usage-metering.md``. +""" + +from nextcloud_mcp_server.usage.store import UsageEventStore + +__all__ = ["UsageEventStore"] diff --git a/nextcloud_mcp_server/usage/store.py b/nextcloud_mcp_server/usage/store.py new file mode 100644 index 00000000..092b0c40 --- /dev/null +++ b/nextcloud_mcp_server/usage/store.py @@ -0,0 +1,158 @@ +"""Best-effort usage-event recording for per-tenant metering (Deck #67). + +A tenant Pod records billable operations (embedding queries, pages/chunks +embedded) into the app-DB ``usage_events`` table; the control plane later pulls +that table read-only into the billing ledger and syncs to Stripe Meter Events +(see control-plane ``usage-metering.md``). This module owns only the data-plane +recording side. + +Design contract: + +- **Flag-gated.** Writes are a no-op unless ``USAGE_METERING_ENABLED`` is true, + so OSS self-hosters and unmetered deployments do zero DB work. +- **Best-effort.** A metering-write failure is logged and dropped, never raised + into the user-facing operation. ``ON CONFLICT (event_id) DO NOTHING`` makes a + retried write a no-op. +- **Engine reuse.** Rather than opening its own engine, this store borrows the + process-wide :class:`RefreshTokenStorage` singleton (``get_shared_storage()``) + — same app DB, NullPool, dialect handling, and ``_DBConn`` shim. The shared + storage guarantees Alembic migrations (incl. ``usage_events``) already ran. +""" + +import json +import logging +import time +import uuid +from datetime import datetime, timezone +from typing import Any + +import anyio + +from nextcloud_mcp_server.auth.storage import RefreshTokenStorage, get_shared_storage +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.observability.metrics import record_db_operation + +logger = logging.getLogger(__name__) + + +# Parameters bind untyped through the ``sa.text(...)`` shim; asyncpg infers +# each placeholder's type from its target column. For ``occurred_at`` +# (TIMESTAMPTZ) it wants a real ``datetime`` (a string is rejected even with a +# CAST), so we bind the aware datetime object on Postgres; SQLite's sqlite3 +# driver can't bind a ``datetime`` on Python 3.12+, so we bind an ISO string +# there. ``metadata`` (JSONB) takes a JSON string on both — asyncpg's jsonb +# codec accepts ``str`` directly, so no cast is needed. Same SQL both ways; +# only the ``occurred_at`` bind value differs by dialect. +_INSERT_SQL = ( + "INSERT INTO usage_events (event_id, occurred_at, metric, value, metadata) " + "VALUES (?, ?, ?, ?, ?) " + "ON CONFLICT (event_id) DO NOTHING" +) + + +class UsageEventStore: + """Append-only writer for the app-DB ``usage_events`` table.""" + + # Process-wide cached instance returned by ``shared()`` so the hot search + # path doesn't allocate a fresh wrapper per metered query. The store is + # stateless beyond its storage handle, so one instance is reusable. + # ``anyio.Lock()`` doesn't bind to an event loop at construction, so a + # class-level instance is safe to define here (mirrors + # ``get_shared_storage``'s ``_shared_lock``). + _shared_instance: "UsageEventStore | None" = None + _shared_lock: anyio.Lock = anyio.Lock() + + def __init__(self, storage: RefreshTokenStorage) -> None: + self._storage = storage + + @classmethod + async def shared(cls) -> "UsageEventStore": + """Return the process-wide store backed by the storage singleton. + + Cached after first build: ``get_shared_storage()`` already returns the + cached :class:`RefreshTokenStorage` (running ``initialize()`` / Alembic + on first access, so ``usage_events`` exists), and the wrapper itself is + stateless, so reusing one instance avoids a per-call allocation on the + ``nc_semantic_search`` hot path. The lock mirrors ``get_shared_storage`` + so two concurrent cold-start callers don't both build (and one silently + overwrite) the instance. + + Tests should construct ``UsageEventStore(storage)`` directly rather than + via ``shared()``: the cache is a process global with no teardown hook, + so a test that called ``shared()`` would leak its storage into the next. + """ + async with cls._shared_lock: + if cls._shared_instance is None: + cls._shared_instance = cls(await get_shared_storage()) + return cls._shared_instance + + async def record_usage_event( + self, + *, + metric: str, + value: int, + occurred_at: datetime | None = None, + metadata: dict[str, Any] | None = None, + event_id: str | None = None, + enabled: bool | None = None, + ) -> None: + """Record one billable usage event (best-effort, flag-gated). + + Does nothing unless ``USAGE_METERING_ENABLED`` is true. Any failure is + logged and swallowed — this must never break the caller's operation. + + Args: + metric: Catalog metric, e.g. ``"embeddings_queries"`` or + ``"pages_chunks"``. + value: Count/quantity for this event. + occurred_at: Operation completion time; defaults to now (UTC). + metadata: Optional rawest-unit context (provider, model, tokens, + doc_type, ...). Stored as JSONB (Postgres) / JSON text (SQLite). + event_id: Optional idempotency key; defaults to a fresh UUID4. + enabled: The resolved ``USAGE_METERING_ENABLED`` value. ``None`` + (default) re-reads it via ``get_settings()`` so the store stays + self-gating for standalone/test use. Hot-path callers that + already hold the flag should pass it to avoid a second uncached + ``Settings`` build (``get_settings()`` is non-cached per + ADR-024 and ``nc_semantic_search`` is on the query path). + """ + if enabled is None: + enabled = get_settings().usage_metering_enabled + if not enabled: + return + + start = time.time() + try: + event_id = event_id or str(uuid.uuid4()) + when = occurred_at or datetime.now(timezone.utc) + # asyncpg takes the datetime object directly; sqlite3 needs a string. + when_bind = ( + when if self._storage.dialect == "postgresql" else when.isoformat() + ) + # json.dumps lives inside the best-effort try: a non-serializable + # metadata dict must be swallowed like any other write failure, not + # raised into the caller's operation (see the contract above). + params = ( + event_id, + when_bind, + metric, + value, + json.dumps(metadata, sort_keys=True) if metadata is not None else None, + ) + async with self._storage.acquire() as db: + await db.execute(_INSERT_SQL, params) + await db.commit() + record_db_operation( + self._storage.dialect, "insert", time.time() - start, "success" + ) + except Exception: + # Best-effort: never surface a metering failure to the user op. + record_db_operation( + self._storage.dialect, "insert", time.time() - start, "error" + ) + logger.warning( + "usage metering write dropped (metric=%s, value=%s)", + metric, + value, + exc_info=True, + ) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 4e99cb6e..b9bd2472 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -29,6 +29,7 @@ from nextcloud_mcp_server.observability.metrics import ( ) from nextcloud_mcp_server.observability.tracing import trace_operation from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter +from nextcloud_mcp_server.usage import UsageEventStore from nextcloud_mcp_server.vector import payload_keys from nextcloud_mcp_server.vector.document_chunker import ( DocumentChunker, @@ -821,6 +822,42 @@ async def _index_document( chunks=len(chunk_texts), chars=total_chars, ) + # Usage metering (Deck #67): record chunks embedded as a billable + # 'pages_chunks' event. Best-effort and gated on the flag so the + # off-path (OSS default) touches no storage; placed after the + # embedding succeeds so it can never affect the indexing path. + # + # Privacy note: user_id stays tenant-local — the CP rollup + # aggregates GROUP BY (day, metric) into usage_daily (no metadata + # column), so nothing here reaches Stripe; it is retained only to + # keep Deck #67's future per-user attribution derivable from the + # app DB without a re-migration. + if settings.usage_metering_enabled: + try: + store = await UsageEventStore.shared() + await store.record_usage_event( + metric="pages_chunks", + value=len(chunk_texts), + metadata={ + "provider": provider, + "model": settings.get_embedding_model_name(), + "doc_type": doc_task.doc_type, + "user_id": doc_task.user_id, + "total_chars": total_chars, + }, + # The outer guard already confirmed the flag, so pass + # enabled=True directly — the store then skips a second + # uncached Settings build here (ADR-024). + enabled=True, + ) + except Exception: + # Reached only when shared()/store construction itself + # raises (record_usage_event swallows its own write + # failures). Metering is on, so warn rather than hide the + # "enabled but no billing data" case in DEBUG logs. + logger.warning( + "usage metering hook (pages_chunks) skipped", exc_info=True + ) async def generate_sparse_embeddings(): """Generate sparse embeddings (BM25 for keyword matching).""" diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 42cc35b9..0b0f9dd5 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -157,6 +157,18 @@ class TestGetSettings: assert settings.vector_sync_processor_workers == 5 assert settings.vector_sync_queue_max_size == 5000 + @patch.dict(os.environ, {}, clear=True) + def test_usage_metering_disabled_by_default(self): + """USAGE_METERING_ENABLED defaults to False (OSS doesn't self-monitor).""" + _reload_config() + assert get_settings().usage_metering_enabled is False + + @patch.dict(os.environ, {"USAGE_METERING_ENABLED": "true"}, clear=True) + def test_usage_metering_enabled_via_env(self): + """USAGE_METERING_ENABLED=true maps to settings.usage_metering_enabled.""" + _reload_config() + assert get_settings().usage_metering_enabled is True + class TestChunkConfigValidation: """Test document chunking configuration validation.""" diff --git a/tests/unit/test_usage_store.py b/tests/unit/test_usage_store.py new file mode 100644 index 00000000..aa4b6944 --- /dev/null +++ b/tests/unit/test_usage_store.py @@ -0,0 +1,275 @@ +"""Unit tests for ``UsageEventStore`` (Deck #67 usage metering, data plane). + +Parametrized over both supported backends via the shared ``storage_backend`` +fixture: SQLite (default, always runs) and Postgres (opt-in, gated on +``TEST_DATABASE_URL`` — bring up ``docker compose --profile postgres up -d +postgres-test`` and export +``TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp``). + +Covers the recording contract: flag-gated no-op, insert roundtrip, ON CONFLICT +dedup, JSON metadata roundtrip, NULL metadata, and the best-effort guarantee +that a DB failure is swallowed instead of surfacing to the caller. +""" + +import json +import logging +import tempfile +import uuid +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from cryptography.fernet import Fernet + +import nextcloud_mcp_server.usage.store as store_module +from nextcloud_mcp_server.auth.storage import RefreshTokenStorage +from nextcloud_mcp_server.usage.store import UsageEventStore + +pytestmark = pytest.mark.unit + + +@pytest.fixture(autouse=True) +def _reset_shared_usage_store(): + """Keep the process-wide ``shared()`` cache from leaking across tests. + + These tests construct ``UsageEventStore(storage)`` directly, but a stray + ``shared()`` call (here or in a smoke test sharing the process) would + otherwise poison later tests with a stale storage handle. + """ + UsageEventStore._shared_instance = None + yield + UsageEventStore._shared_instance = None + + +@pytest.fixture +async def storage(storage_backend): + """Initialized RefreshTokenStorage backed by SQLite or Postgres.""" + key = Fernet.generate_key() + if storage_backend["kind"] == "sqlite": + with tempfile.TemporaryDirectory() as tmpdir: + db_path = Path(tmpdir) / "usage.db" + s = RefreshTokenStorage(db_path=str(db_path), encryption_key=key) + await s.initialize() + yield s + else: + s = RefreshTokenStorage(database_url=storage_backend["url"], encryption_key=key) + await s.initialize() + try: + yield s + finally: + await storage_backend["reset"]() + + +def _set_metering(monkeypatch, enabled: bool) -> None: + """Force the metering flag without mutating global dynaconf state. + + ``record_usage_event`` calls ``get_settings()`` (imported into the store + module's namespace), and ``get_settings()`` builds a fresh Settings per + call — so patching the symbol the store sees is the clean seam. + """ + + class _Settings: + usage_metering_enabled = enabled + + monkeypatch.setattr(store_module, "get_settings", lambda: _Settings()) + + +async def _count(storage: RefreshTokenStorage) -> int: + async with storage.acquire() as db: + cursor = await db.execute("SELECT COUNT(*) FROM usage_events") + row = await cursor.fetchone() + return row[0] + + +async def _fetch(storage: RefreshTokenStorage, event_id: str): + async with storage.acquire() as db: + cursor = await db.execute( + "SELECT event_id, occurred_at, metric, value, metadata " + "FROM usage_events WHERE event_id = ?", + (event_id,), + ) + return await cursor.fetchone() + + +async def test_flag_off_is_noop(storage, monkeypatch): + """With metering disabled, nothing is written (zero DB work).""" + _set_metering(monkeypatch, False) + store = UsageEventStore(storage) + await store.record_usage_event(metric="pages_chunks", value=5) + assert await _count(storage) == 0 + + +async def test_enabled_param_short_circuits_without_reading_settings( + storage, monkeypatch +): + """An explicit ``enabled`` flag is honored without touching get_settings(). + + Hot-path callers pass the already-resolved flag; the store must not rebuild + Settings when given one. ``enabled=False`` is a no-op; ``enabled=True`` + writes even though the (boobytrapped) settings lookup would raise. + """ + + def _boom(): + raise AssertionError("get_settings() must not be called when enabled is passed") + + monkeypatch.setattr(store_module, "get_settings", _boom) + store = UsageEventStore(storage) + + await store.record_usage_event(metric="pages_chunks", value=1, enabled=False) + assert await _count(storage) == 0 + + eid = str(uuid.uuid4()) + await store.record_usage_event( + metric="pages_chunks", value=1, event_id=eid, enabled=True + ) + assert await _count(storage) == 1 + + +async def test_insert_roundtrip(storage, monkeypatch): + """A recorded event lands and reads back with the right fields.""" + _set_metering(monkeypatch, True) + store = UsageEventStore(storage) + eid = str(uuid.uuid4()) + await store.record_usage_event( + metric="pages_chunks", + value=7, + event_id=eid, + metadata={"provider": "gateway"}, + ) + row = await _fetch(storage, eid) + assert row is not None + # Postgres returns event_id as a uuid.UUID; normalize to str for compare. + assert str(row[0]) == eid + assert row[2] == "pages_chunks" + assert row[3] == 7 + + +async def test_on_conflict_dedup(storage, monkeypatch): + """A duplicate event_id is a no-op; the first write is retained.""" + _set_metering(monkeypatch, True) + store = UsageEventStore(storage) + eid = str(uuid.uuid4()) + await store.record_usage_event(metric="pages_chunks", value=1, event_id=eid) + await store.record_usage_event(metric="embeddings_queries", value=99, event_id=eid) + assert await _count(storage) == 1 + row = await _fetch(storage, eid) + assert row[2] == "pages_chunks" # DO NOTHING, not DO UPDATE + assert row[3] == 1 + + +async def test_metadata_json_roundtrip(storage, monkeypatch): + """Nested metadata round-trips as JSON on both backends.""" + _set_metering(monkeypatch, True) + store = UsageEventStore(storage) + eid = str(uuid.uuid4()) + meta = {"provider": "gateway", "model": "titan", "nested": {"chunks": 3}} + await store.record_usage_event( + metric="pages_chunks", value=3, event_id=eid, metadata=meta + ) + row = await _fetch(storage, eid) + raw = row[4] + # Depending on the asyncpg/SQLAlchemy JSONB codec in play, Postgres may + # return JSONB as a Python dict or as a JSON str; SQLite stores TEXT. + # Handle both so the test is robust across driver/codec versions. + loaded = raw if isinstance(raw, dict) else json.loads(raw) + assert loaded == meta + + +async def test_occurred_at_roundtrip(storage, monkeypatch): + """occurred_at round-trips to the same instant on both backends. + + The store binds a datetime on Postgres and an ISO string on SQLite (the + only dialect-specific branch in the store); this pins that both read back + to the same instant regardless of the stored representation. + """ + _set_metering(monkeypatch, True) + store = UsageEventStore(storage) + eid = str(uuid.uuid4()) + when = datetime(2026, 1, 15, 12, 0, 0, tzinfo=timezone.utc) + await store.record_usage_event( + metric="pages_chunks", value=1, event_id=eid, occurred_at=when + ) + row = await _fetch(storage, eid) + stored = row[1] + # SQLite returns the ISO string we bound; Postgres returns a datetime. + parsed = stored if isinstance(stored, datetime) else datetime.fromisoformat(stored) + # Aware-datetime equality compares the instant, so a UTC value coming back + # in another session tz still matches; a naive value (none expected) is + # treated as UTC. + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + assert parsed == when + + +async def test_metadata_none_is_null(storage, monkeypatch): + """Omitting metadata stores SQL NULL, not the string 'null'.""" + _set_metering(monkeypatch, True) + store = UsageEventStore(storage) + eid = str(uuid.uuid4()) + await store.record_usage_event( + metric="embeddings_queries", value=1, event_id=eid, metadata=None + ) + row = await _fetch(storage, eid) + assert row[4] is None + + +async def test_best_effort_swallows_db_errors(storage, monkeypatch, caplog): + """A DB failure is logged + dropped, never raised into the caller.""" + _set_metering(monkeypatch, True) + store = UsageEventStore(storage) + + recorded: list[tuple] = [] + monkeypatch.setattr( + store_module, + "record_db_operation", + lambda *args, **kwargs: recorded.append(args), + ) + + def _boom(): + raise RuntimeError("db down") + + # ``acquire`` raises on call — the store must catch and continue. + monkeypatch.setattr(storage, "acquire", _boom) + + # Must not raise. + with caplog.at_level(logging.WARNING, logger="nextcloud_mcp_server.usage.store"): + await store.record_usage_event(metric="pages_chunks", value=1) + + assert recorded, "record_db_operation should be called on the error path" + assert recorded[-1][3] == "error" + # The observability contract: the dropped write surfaces at WARNING. + assert any( + r.levelno == logging.WARNING and "usage metering write dropped" in r.message + for r in caplog.records + ) + + +async def test_best_effort_swallows_unserializable_metadata( + storage, monkeypatch, caplog +): + """Non-serializable metadata is swallowed, not raised into the caller. + + json.dumps runs inside the best-effort try, so a metadata value the JSON + encoder can't handle must drop the event like any other write failure + rather than surfacing to the user op. + """ + _set_metering(monkeypatch, True) + store = UsageEventStore(storage) + + # An arbitrary object is not JSON-serializable; json.dumps raises TypeError. + bad_metadata = {"obj": object()} + + # Must not raise. + with caplog.at_level(logging.WARNING, logger="nextcloud_mcp_server.usage.store"): + await store.record_usage_event( + metric="pages_chunks", value=1, metadata=bad_metadata + ) + + # Nothing was written — the encode failed before the insert. + assert await _count(storage) == 0 + # Silent data loss would be a footgun once metering is on: same WARNING + # contract as the DB-error path. + assert any( + r.levelno == logging.WARNING and "usage metering write dropped" in r.message + for r in caplog.records + )