From 1c6b1a84ea11beae2e2aa851301a6f32fbcf17f9 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 7 Jun 2026 15:13:14 +0200 Subject: [PATCH 1/7] feat(usage): record per-tenant usage events into the app DB Deck #67 data-plane slice: tenant Pods record billable operations (embedding queries, pages/chunks embedded) into an app-DB usage_events table that the control plane later pulls read-only into the billing ledger and syncs to Stripe Meter Events. - migration 007: usage_events table (Postgres TIMESTAMPTZ/JSONB/UUID with portable SQLite fallbacks), indexed (occurred_at, metric) for the CP rollup's per-day range scan + GROUP BY metric. - UsageEventStore: best-effort, flag-gated writer reusing the shared RefreshTokenStorage engine; ON CONFLICT (event_id) DO NOTHING for idempotent retries; dialect-branched occurred_at bind. All work (incl. metadata JSON encode) is swallowed so a metering failure never surfaces to the user op. - USAGE_METERING_ENABLED flag (default off) wired through Settings + env map; off-path touches no storage, so OSS self-hosters get an empty table and zero write overhead. - two recording hooks: embeddings_queries (per nc_semantic_search, which nc_semantic_search_answer reuses) and pages_chunks (after dense embedding succeeds, covering both in-process and procrastinate paths). - storage.acquire()/.dialect public seams so the sibling store doesn't reach into the underscored internal. - tests parametrized over SQLite + Postgres: flag-off no-op, roundtrip, ON CONFLICT dedup, JSON/NULL metadata, and the best-effort swallow of both DB errors and unserializable metadata. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../20260610_1200_007_add_usage_events.py | 78 +++++++ nextcloud_mcp_server/auth/storage.py | 17 +- nextcloud_mcp_server/config.py | 10 + nextcloud_mcp_server/server/semantic.py | 24 +++ nextcloud_mcp_server/usage/__init__.py | 10 + nextcloud_mcp_server/usage/store.py | 127 ++++++++++++ nextcloud_mcp_server/vector/processor.py | 23 +++ tests/unit/test_config.py | 12 ++ tests/unit/test_usage_store.py | 191 ++++++++++++++++++ 9 files changed, 491 insertions(+), 1 deletion(-) create mode 100644 nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py create mode 100644 nextcloud_mcp_server/usage/__init__.py create mode 100644 nextcloud_mcp_server/usage/store.py create mode 100644 tests/unit/test_usage_store.py diff --git a/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py b/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py new file mode 100644 index 00000000..53205772 --- /dev/null +++ b/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py @@ -0,0 +1,78 @@ +"""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-10 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" + + 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'. + 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..5ce62d69 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -39,6 +39,7 @@ 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 @@ -517,6 +518,29 @@ 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. + 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, + "doc_types": doc_types, + }, + ) + except Exception: + logger.debug( + "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..3f1b7298 --- /dev/null +++ b/nextcloud_mcp_server/usage/store.py @@ -0,0 +1,127 @@ +"""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 + +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.""" + + def __init__(self, storage: RefreshTokenStorage) -> None: + self._storage = storage + + @classmethod + async def shared(cls) -> "UsageEventStore": + """Build a store backed by the process-wide storage singleton. + + ``get_shared_storage()`` runs ``initialize()`` (and thus Alembic + migrations) on first access, so the ``usage_events`` table is present + by the time any event is recorded. + """ + return cls(await get_shared_storage()) + + 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, + ) -> 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. + """ + if not get_settings().usage_metering_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..863ae9e9 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,28 @@ 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. + 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, + }, + ) + except Exception: + logger.debug( + "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..f14aca6d --- /dev/null +++ b/tests/unit/test_usage_store.py @@ -0,0 +1,191 @@ +"""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 tempfile +import uuid +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 +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_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] + # asyncpg returns JSONB as a JSON string (no codec); SQLite stores TEXT. + loaded = raw if isinstance(raw, dict) else json.loads(raw) + assert loaded == meta + + +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): + """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. + 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" + + +async def test_best_effort_swallows_unserializable_metadata(storage, monkeypatch): + """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. + 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 From 702f66e6b168ef5e7fe1581effad3d29df4299d8 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 7 Jun 2026 15:20:41 +0200 Subject: [PATCH 2/7] refactor(usage): address round-1 review on PR #871 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - store: add optional `enabled` param to record_usage_event so hot-path callers (nc_semantic_search) pass the already-resolved flag instead of forcing a second uncached Settings build (ADR-024); falls back to get_settings() when None so the store stays self-gating for standalone use. - hooks: thread enabled= through both call sites; bump the outer shared()/construction failure log from debug → warning so "metering enabled but no billing data" is visible at the default INFO level. - migration: instantiate postgresql.JSONB() to match the sibling TIMESTAMP(timezone=True) column. - tests: fix the misleading "asyncpg returns JSONB as a JSON string" comment; add occurred_at dialect round-trip test and an enabled-param short-circuit test. Co-Authored-By: Claude Opus 4.8 (1M context) --- .claude/scheduled_tasks.lock | 1 + .../20260610_1200_007_add_usage_events.py | 2 +- nextcloud_mcp_server/server/semantic.py | 9 ++- nextcloud_mcp_server/usage/store.py | 11 +++- nextcloud_mcp_server/vector/processor.py | 9 ++- tests/unit/test_usage_store.py | 57 ++++++++++++++++++- 6 files changed, 84 insertions(+), 5 deletions(-) create mode 100644 .claude/scheduled_tasks.lock diff --git a/.claude/scheduled_tasks.lock b/.claude/scheduled_tasks.lock new file mode 100644 index 00000000..4e015a00 --- /dev/null +++ b/.claude/scheduled_tasks.lock @@ -0,0 +1 @@ +{"sessionId":"fdfa73f5-8734-459a-8f23-67e1eeaa804c","pid":13329,"procStart":"23713","acquiredAt":1780838157352} \ No newline at end of file diff --git a/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py b/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py index 53205772..17c38d40 100644 --- a/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py +++ b/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py @@ -60,7 +60,7 @@ def upgrade() -> None: # (json.dumps) on SQLite. sa.Column( "metadata", - postgresql.JSONB if is_pg else sa.Text, + postgresql.JSONB() if is_pg else sa.Text, nullable=True, ), ) diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 5ce62d69..9640573d 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -534,9 +534,16 @@ def configure_semantic_tools(mcp: FastMCP): "fusion": fusion, "doc_types": doc_types, }, + # Pass the already-resolved flag so the store doesn't + # rebuild Settings on this hot query path (ADR-024). + enabled=settings.usage_metering_enabled, ) except Exception: - logger.debug( + # 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, ) diff --git a/nextcloud_mcp_server/usage/store.py b/nextcloud_mcp_server/usage/store.py index 3f1b7298..c7551b61 100644 --- a/nextcloud_mcp_server/usage/store.py +++ b/nextcloud_mcp_server/usage/store.py @@ -72,6 +72,7 @@ class UsageEventStore: 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). @@ -86,8 +87,16 @@ class UsageEventStore: 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 not get_settings().usage_metering_enabled: + if enabled is None: + enabled = get_settings().usage_metering_enabled + if not enabled: return start = time.time() diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 863ae9e9..0565cd75 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -839,9 +839,16 @@ async def _index_document( "user_id": doc_task.user_id, "total_chars": total_chars, }, + # Pass the already-resolved flag so the store doesn't + # rebuild Settings here (ADR-024). + enabled=settings.usage_metering_enabled, ) except Exception: - logger.debug( + # 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 ) diff --git a/tests/unit/test_usage_store.py b/tests/unit/test_usage_store.py index f14aca6d..11ab5083 100644 --- a/tests/unit/test_usage_store.py +++ b/tests/unit/test_usage_store.py @@ -14,6 +14,7 @@ that a DB failure is swallowed instead of surfacing to the caller. import json import tempfile import uuid +from datetime import datetime, timezone from pathlib import Path import pytest @@ -84,6 +85,32 @@ async def test_flag_off_is_noop(storage, monkeypatch): 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) @@ -127,11 +154,39 @@ async def test_metadata_json_roundtrip(storage, monkeypatch): ) row = await _fetch(storage, eid) raw = row[4] - # asyncpg returns JSONB as a JSON string (no codec); SQLite stores TEXT. + # 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, 6, 10, 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) From 2bbf4ed96794a3d3f05aa9dee9cccd5780a144c7 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 7 Jun 2026 15:28:24 +0200 Subject: [PATCH 3/7] refactor(usage): address round-2 review on PR #871 - remove accidentally-committed .claude/scheduled_tasks.lock (Claude Code runtime artifact swept in by `git add -A`) and gitignore it; the rest of .claude/ stays tracked. - store: cache UsageEventStore.shared() as a process-wide instance so the hot search path doesn't allocate a fresh wrapper per metered query (the wrapper is stateless beyond its storage handle). - hooks: pass enabled=True directly (the outer guard already confirmed the flag) instead of re-reading settings.usage_metering_enabled. - migration: document the no-TTL retention design (control-plane rollup owns the lifecycle; the data plane only appends). - tests: assert the best-effort error path logs at WARNING (observability contract). Co-Authored-By: Claude Opus 4.8 (1M context) --- .claude/scheduled_tasks.lock | 1 - .gitignore | 3 +++ .../20260610_1200_007_add_usage_events.py | 3 +++ nextcloud_mcp_server/server/semantic.py | 8 +++++--- nextcloud_mcp_server/usage/store.py | 19 ++++++++++++++----- nextcloud_mcp_server/vector/processor.py | 7 ++++--- tests/unit/test_usage_store.py | 11 +++++++++-- 7 files changed, 38 insertions(+), 14 deletions(-) delete mode 100644 .claude/scheduled_tasks.lock diff --git a/.claude/scheduled_tasks.lock b/.claude/scheduled_tasks.lock deleted file mode 100644 index 4e015a00..00000000 --- a/.claude/scheduled_tasks.lock +++ /dev/null @@ -1 +0,0 @@ -{"sessionId":"fdfa73f5-8734-459a-8f23-67e1eeaa804c","pid":13329,"procStart":"23713","acquiredAt":1780838157352} \ No newline at end of file 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/20260610_1200_007_add_usage_events.py b/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py index 17c38d40..fcfefa45 100644 --- a/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py +++ b/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py @@ -34,6 +34,9 @@ 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, diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 9640573d..4b3574f3 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -534,9 +534,11 @@ def configure_semantic_tools(mcp: FastMCP): "fusion": fusion, "doc_types": doc_types, }, - # Pass the already-resolved flag so the store doesn't - # rebuild Settings on this hot query path (ADR-024). - enabled=settings.usage_metering_enabled, + # 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 diff --git a/nextcloud_mcp_server/usage/store.py b/nextcloud_mcp_server/usage/store.py index c7551b61..b2298e78 100644 --- a/nextcloud_mcp_server/usage/store.py +++ b/nextcloud_mcp_server/usage/store.py @@ -51,18 +51,27 @@ _INSERT_SQL = ( 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. + _shared_instance: "UsageEventStore | None" = None + def __init__(self, storage: RefreshTokenStorage) -> None: self._storage = storage @classmethod async def shared(cls) -> "UsageEventStore": - """Build a store backed by the process-wide storage singleton. + """Return the process-wide store backed by the storage singleton. - ``get_shared_storage()`` runs ``initialize()`` (and thus Alembic - migrations) on first access, so the ``usage_events`` table is present - by the time any event is recorded. + 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. """ - return cls(await get_shared_storage()) + if cls._shared_instance is None: + cls._shared_instance = cls(await get_shared_storage()) + return cls._shared_instance async def record_usage_event( self, diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 0565cd75..87e94a6d 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -839,9 +839,10 @@ async def _index_document( "user_id": doc_task.user_id, "total_chars": total_chars, }, - # Pass the already-resolved flag so the store doesn't - # rebuild Settings here (ADR-024). - enabled=settings.usage_metering_enabled, + # 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 diff --git a/tests/unit/test_usage_store.py b/tests/unit/test_usage_store.py index 11ab5083..e15222a8 100644 --- a/tests/unit/test_usage_store.py +++ b/tests/unit/test_usage_store.py @@ -12,6 +12,7 @@ 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 @@ -199,7 +200,7 @@ async def test_metadata_none_is_null(storage, monkeypatch): assert row[4] is None -async def test_best_effort_swallows_db_errors(storage, monkeypatch): +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) @@ -218,10 +219,16 @@ async def test_best_effort_swallows_db_errors(storage, monkeypatch): monkeypatch.setattr(storage, "acquire", _boom) # Must not raise. - await store.record_usage_event(metric="pages_chunks", value=1) + 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): From 3a8ea893c6f0425467a6d6c5f43ab0fbb31719b4 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 7 Jun 2026 15:35:21 +0200 Subject: [PATCH 4/7] refactor(usage): address round-3 review on PR #871 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - store: guard UsageEventStore.shared() with a class-level anyio.Lock so two concurrent cold-start callers don't both build (and one silently overwrite) the cached instance — mirrors get_shared_storage(). Document that tests should construct the store directly to avoid singleton leak. - migration: rename 20260610 -> 20260607 and fix Create Date to today so `alembic history` isn't future-dated (revision id 007 / down_revision 006 unchanged; single head verified). Co-Authored-By: Claude Opus 4.8 (1M context) --- ... => 20260607_1200_007_add_usage_events.py} | 2 +- nextcloud_mcp_server/usage/store.py | 19 ++++++++++++++++--- 2 files changed, 17 insertions(+), 4 deletions(-) rename nextcloud_mcp_server/alembic/versions/{20260610_1200_007_add_usage_events.py => 20260607_1200_007_add_usage_events.py} (98%) diff --git a/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py b/nextcloud_mcp_server/alembic/versions/20260607_1200_007_add_usage_events.py similarity index 98% rename from nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py rename to nextcloud_mcp_server/alembic/versions/20260607_1200_007_add_usage_events.py index fcfefa45..e2a41937 100644 --- a/nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py +++ b/nextcloud_mcp_server/alembic/versions/20260607_1200_007_add_usage_events.py @@ -17,7 +17,7 @@ fallbacks (``TEXT``/``TIMESTAMP``); the control plane never queries SQLite. Revision ID: 007 Revises: 006 -Create Date: 2026-06-10 12:00:00.000000 +Create Date: 2026-06-07 12:00:00.000000 """ import sqlalchemy as sa diff --git a/nextcloud_mcp_server/usage/store.py b/nextcloud_mcp_server/usage/store.py index b2298e78..092b0c40 100644 --- a/nextcloud_mcp_server/usage/store.py +++ b/nextcloud_mcp_server/usage/store.py @@ -26,6 +26,8 @@ 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 @@ -54,7 +56,11 @@ class UsageEventStore: # 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 @@ -67,10 +73,17 @@ class UsageEventStore: 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. + ``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. """ - if cls._shared_instance is None: - cls._shared_instance = cls(await get_shared_storage()) + 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( From 9c2f9fac46dd8b0a5e3729944455362c816021bc Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 7 Jun 2026 15:42:12 +0200 Subject: [PATCH 5/7] refactor(usage): address round-4 review on PR #871 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - hooks: document why user_id in metadata is safe — it stays tenant-local (the CP rollup aggregates GROUP BY (day, metric) into usage_daily, which has no metadata column, so it never reaches Stripe) and is retained to keep Deck #67's future per-user attribution derivable from the app DB. - migration: instantiate the SQLite-side column types (sa.Text() etc.) for visual parity with the instantiated Postgres types. - tests: assert the WARNING contract in the unserializable-metadata test too; add an autouse fixture that resets UsageEventStore._shared_instance so a stray shared() call can't leak across tests. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../20260607_1200_007_add_usage_events.py | 10 +++---- nextcloud_mcp_server/server/semantic.py | 6 ++++ nextcloud_mcp_server/vector/processor.py | 6 ++++ tests/unit/test_usage_store.py | 30 ++++++++++++++++--- 4 files changed, 43 insertions(+), 9 deletions(-) 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 index e2a41937..dc61317a 100644 --- 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 @@ -44,7 +44,7 @@ def upgrade() -> None: # both backends (see usage/store.py). sa.Column( "event_id", - postgresql.UUID(as_uuid=False) if is_pg else sa.Text, + postgresql.UUID(as_uuid=False) if is_pg else sa.Text(), primary_key=True, ), # Operation completion time (UTC). Real TIMESTAMPTZ on Postgres so the @@ -52,18 +52,18 @@ def upgrade() -> None: # TIMESTAMP on SQLite (stored as ISO text, queryable in tests). sa.Column( "occurred_at", - postgresql.TIMESTAMP(timezone=True) if is_pg else sa.TIMESTAMP, + postgresql.TIMESTAMP(timezone=True) if is_pg else sa.TIMESTAMP(), nullable=False, ), # Catalog metric: 'embeddings_queries' or 'pages_chunks'. - sa.Column("metric", sa.Text, nullable=False), - sa.Column("value", sa.BigInteger, nullable=False), + 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, + postgresql.JSONB() if is_pg else sa.Text(), nullable=True, ), ) diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 4b3574f3..f8d36221 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -523,6 +523,12 @@ def configure_semantic_tools(mcp: FastMCP): # 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() diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 87e94a6d..b9bd2472 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -826,6 +826,12 @@ async def _index_document( # '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() diff --git a/tests/unit/test_usage_store.py b/tests/unit/test_usage_store.py index e15222a8..475254c8 100644 --- a/tests/unit/test_usage_store.py +++ b/tests/unit/test_usage_store.py @@ -28,6 +28,19 @@ 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.""" @@ -231,7 +244,9 @@ async def test_best_effort_swallows_db_errors(storage, monkeypatch, caplog): ) -async def test_best_effort_swallows_unserializable_metadata(storage, monkeypatch): +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 @@ -245,9 +260,16 @@ async def test_best_effort_swallows_unserializable_metadata(storage, monkeypatch bad_metadata = {"obj": object()} # Must not raise. - await store.record_usage_event( - metric="pages_chunks", value=1, metadata=bad_metadata - ) + 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 + ) From c89f724585fca4961ded3f60d39bdb8bfc49a4c0 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 7 Jun 2026 15:47:57 +0200 Subject: [PATCH 6/7] refactor(usage): close out round-5 nits on PR #871 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Non-blocking follow-ups from the merge-ready review: - semantic.py: bound the doc_types copied into embeddings_queries metadata to _USAGE_METADATA_MAX_DOC_TYPES (16). doc_types is caller-supplied with no max_length on the tool signature; capping the stored copy keeps one JSONB row from ballooning (not a billing/injection risk — CP ignores metadata, binds are parameterized). - migration: note that `metric` is intentionally unconstrained Text and that adding a third metric requires keeping the CP-side catalog in sync, else the rollup silently ignores the new rows. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../20260607_1200_007_add_usage_events.py | 6 +++++- nextcloud_mcp_server/server/semantic.py | 16 +++++++++++++++- 2 files changed, 20 insertions(+), 2 deletions(-) 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 index dc61317a..6ce5bf32 100644 --- 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 @@ -55,7 +55,11 @@ def upgrade() -> None: postgresql.TIMESTAMP(timezone=True) if is_pg else sa.TIMESTAMP(), nullable=False, ), - # Catalog metric: 'embeddings_queries' or 'pages_chunks'. + # 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, ...). diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index f8d36221..ec93a851 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -46,6 +46,15 @@ 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.""" @@ -538,7 +547,12 @@ def configure_semantic_tools(mcp: FastMCP): metadata={ "user_id": username, "fusion": fusion, - "doc_types": doc_types, + # Bounded copy — see _USAGE_METADATA_MAX_DOC_TYPES. + "doc_types": ( + doc_types[:_USAGE_METADATA_MAX_DOC_TYPES] + if doc_types + else doc_types + ), }, # The outer guard already confirmed the flag, so pass # enabled=True directly — the store then skips a second From 98de8f331f02c7efcb749736552858dd0d994219 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 7 Jun 2026 15:52:54 +0200 Subject: [PATCH 7/7] refactor(usage): final round-6 nits on PR #871 - semantic.py: normalize both None and [] doc_types to null in the metadata so a future `metadata->'doc_types' IS NULL` query counts the all-types case consistently. - test: use a fixed past date in test_occurred_at_roundtrip instead of a future literal (deterministic, no "why this date" confusion). Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/server/semantic.py | 5 ++++- tests/unit/test_usage_store.py | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index ec93a851..8d78849f 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -548,10 +548,13 @@ def configure_semantic_tools(mcp: FastMCP): "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 doc_types + else None ), }, # The outer guard already confirmed the flag, so pass diff --git a/tests/unit/test_usage_store.py b/tests/unit/test_usage_store.py index 475254c8..aa4b6944 100644 --- a/tests/unit/test_usage_store.py +++ b/tests/unit/test_usage_store.py @@ -185,7 +185,7 @@ async def test_occurred_at_roundtrip(storage, monkeypatch): _set_metering(monkeypatch, True) store = UsageEventStore(storage) eid = str(uuid.uuid4()) - when = datetime(2026, 6, 10, 12, 0, 0, tzinfo=timezone.utc) + 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 )