feat: CLI analytics via Umami event collector (#572)

Signed-off-by: phernandez <paul@basicmachines.co>

Author: bm-clawd

src/basic_memory/cli/analytics.py

@@ -0,0 +1,103 @@
+"""Lightweight CLI analytics via Umami event collector.
+
+Sends anonymous, non-blocking usage events to help understand how the
+CLI-to-cloud conversion funnel performs. No PII, no fingerprinting,
+no cookies. Respects the same opt-out mechanisms as promo messaging.
+
+Events are fire-and-forget — analytics never blocks or breaks the CLI.
+
+Setup:
+    Set these environment variables (or leave unset to disable):
+        BASIC_MEMORY_UMAMI_HOST     — Umami instance URL (e.g. https://analytics.basicmemory.com)
+        BASIC_MEMORY_UMAMI_SITE_ID  — Website ID from Umami dashboard
+"""
+
+import json
+import os
+import threading
+import urllib.request
+from typing import Optional
+
+import basic_memory
+
+
+# ---------------------------------------------------------------------------
+# Configuration — read from environment so nothing is hard-coded in source
+# ---------------------------------------------------------------------------
+
+def _umami_host() -> Optional[str]:
+    return os.getenv("BASIC_MEMORY_UMAMI_HOST", "").strip() or None
+
+
+def _umami_site_id() -> Optional[str]:
+    return os.getenv("BASIC_MEMORY_UMAMI_SITE_ID", "").strip() or None
+
+
+def _analytics_disabled() -> bool:
+    """True when analytics should not fire."""
+    value = os.getenv("BASIC_MEMORY_NO_PROMOS", "").strip().lower()
+    return value in {"1", "true", "yes"}
+
+
+def _is_configured() -> bool:
+    """True when both host and site ID are available."""
+    return _umami_host() is not None and _umami_site_id() is not None
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+# Well-known event names for the promo/cloud funnel
+EVENT_PROMO_SHOWN = "cli-promo-shown"
+EVENT_PROMO_OPTED_OUT = "cli-promo-opted-out"
+EVENT_CLOUD_LOGIN_STARTED = "cli-cloud-login-started"
+EVENT_CLOUD_LOGIN_SUCCESS = "cli-cloud-login-success"
+EVENT_CLOUD_LOGIN_SUB_REQUIRED = "cli-cloud-login-sub-required"
+
+
+def track(event_name: str, data: Optional[dict] = None) -> None:
+    """Send an analytics event to Umami. Non-blocking, silent on failure.
+
+    Parameters
+    ----------
+    event_name:
+        Short kebab-case name (e.g. "cli-promo-shown").
+    data:
+        Optional dict of event properties (all values should be strings/numbers).
+    """
+    if _analytics_disabled() or not _is_configured():
+        return
+
+    host = _umami_host()
+    site_id = _umami_site_id()
+
+    payload = {
+        "payload": {
+            "hostname": "cli.basicmemory.com",
+            "language": "en",
+            "url": f"/cli/{event_name}",
+            "website": site_id,
+            "name": event_name,
+            "data": {
+                "version": basic_memory.__version__,
+                **(data or {}),
+            },
+        }
+    }
+
+    def _send():
+        try:
+            req = urllib.request.Request(
+                f"{host}/api/send",
+                data=json.dumps(payload).encode("utf-8"),
+                headers={
+                    "Content-Type": "application/json",
+                    "User-Agent": f"basic-memory-cli/{basic_memory.__version__}",
+                },
+            )
+            urllib.request.urlopen(req, timeout=3)
+        except Exception:
+            pass  # Never break the CLI for analytics
+
+    threading.Thread(target=_send, daemon=True).start()

src/basic_memory/cli/commands/cloud/core_commands.py

@@ -6,6 +6,13 @@
 from basic_memory.cli.app import cloud_app
 from basic_memory.cli.commands.command_utils import run_with_cleanup
 from basic_memory.cli.auth import CLIAuth
+from basic_memory.cli.analytics import (
+    track,
+    EVENT_CLOUD_LOGIN_STARTED,
+    EVENT_CLOUD_LOGIN_SUCCESS,
+    EVENT_CLOUD_LOGIN_SUB_REQUIRED,
+    EVENT_PROMO_OPTED_OUT,
+)
 from basic_memory.cli.promo import OSS_DISCOUNT_CODE
 from basic_memory.config import ConfigManager
 from basic_memory.cli.commands.cloud.api_client import (
@@ -33,6 +40,7 @@ def login():
     """Authenticate with WorkOS using OAuth Device Authorization flow."""
 
     async def _login():
+        track(EVENT_CLOUD_LOGIN_STARTED)
         client_id, domain, host_url = get_cloud_config()
         auth = CLIAuth(client_id=client_id, authkit_domain=domain)
 
@@ -46,10 +54,12 @@ async def _login():
             console.print("[dim]Verifying subscription access...[/dim]")
             await make_api_request("GET", f"{host_url.rstrip('/')}/proxy/health")
 
+            track(EVENT_CLOUD_LOGIN_SUCCESS)
             console.print("[green]Cloud authentication successful[/green]")
             console.print(f"[dim]Cloud host ready: {host_url}[/dim]")
 
         except SubscriptionRequiredError as e:
+            track(EVENT_CLOUD_LOGIN_SUB_REQUIRED)
             console.print("\n[red]Subscription Required[/red]\n")
             console.print(f"[yellow]{e.args[0]}[/yellow]\n")
             console.print(
@@ -205,6 +215,7 @@ def promo(enabled: bool = typer.Option(True, "--on/--off", help="Enable or disab
     if enabled:
         console.print("[green]Cloud promo messages enabled[/green]")
     else:
+        track(EVENT_PROMO_OPTED_OUT)
         console.print("[yellow]Cloud promo messages disabled[/yellow]")
 
 

src/basic_memory/cli/promo.py

@@ -7,10 +7,14 @@
 from rich.panel import Panel
 
 import basic_memory
+from basic_memory.cli.analytics import track, EVENT_PROMO_SHOWN, EVENT_PROMO_OPTED_OUT
 from basic_memory.config import ConfigManager
 
 OSS_DISCOUNT_CODE = "BMFOSS"
-CLOUD_LEARN_MORE_URL = "https://basicmemory.com"
+CLOUD_LEARN_MORE_URL = (
+    "https://basicmemory.com"
+    "?utm_source=bm-cli&utm_medium=promo&utm_campaign=cloud-upsell"
+)
 
 
 def _promos_disabled_by_env() -> bool:
@@ -113,6 +117,9 @@ def maybe_show_cloud_promo(
     out.print(f"Learn more at [link={CLOUD_LEARN_MORE_URL}]{CLOUD_LEARN_MORE_URL}[/link]")
     out.print("[dim]Disable with: bm cloud promo --off[/dim]")
 
+    trigger = "first_run" if show_first_run else "version_bump"
+    track(EVENT_PROMO_SHOWN, {"trigger": trigger})
+
     config.cloud_promo_first_run_shown = True
     config.cloud_promo_last_version_shown = basic_memory.__version__
     manager.save_config(config)

tests/cli/test_analytics.py

@@ -0,0 +1,157 @@
+"""Tests for CLI analytics module."""
+
+import json
+import threading
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from basic_memory.cli.analytics import (
+    track,
+    _analytics_disabled,
+    _is_configured,
+    EVENT_PROMO_SHOWN,
+    EVENT_CLOUD_LOGIN_STARTED,
+    EVENT_CLOUD_LOGIN_SUCCESS,
+    EVENT_CLOUD_LOGIN_SUB_REQUIRED,
+    EVENT_PROMO_OPTED_OUT,
+)
+
+
+class TestAnalyticsDisabled:
+    def test_disabled_when_env_set(self, monkeypatch):
+        monkeypatch.setenv("BASIC_MEMORY_NO_PROMOS", "1")
+        assert _analytics_disabled() is True
+
+    def test_disabled_when_env_true(self, monkeypatch):
+        monkeypatch.setenv("BASIC_MEMORY_NO_PROMOS", "true")
+        assert _analytics_disabled() is True
+
+    def test_not_disabled_by_default(self, monkeypatch):
+        monkeypatch.delenv("BASIC_MEMORY_NO_PROMOS", raising=False)
+        assert _analytics_disabled() is False
+
+
+class TestIsConfigured:
+    def test_configured_when_both_set(self, monkeypatch):
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_HOST", "https://analytics.example.com")
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_SITE_ID", "abc-123")
+        assert _is_configured() is True
+
+    def test_not_configured_when_host_missing(self, monkeypatch):
+        monkeypatch.delenv("BASIC_MEMORY_UMAMI_HOST", raising=False)
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_SITE_ID", "abc-123")
+        assert _is_configured() is False
+
+    def test_not_configured_when_site_id_missing(self, monkeypatch):
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_HOST", "https://analytics.example.com")
+        monkeypatch.delenv("BASIC_MEMORY_UMAMI_SITE_ID", raising=False)
+        assert _is_configured() is False
+
+    def test_not_configured_when_empty_strings(self, monkeypatch):
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_HOST", "")
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_SITE_ID", "")
+        assert _is_configured() is False
+
+
+class TestTrack:
+    def test_no_op_when_disabled(self, monkeypatch):
+        monkeypatch.setenv("BASIC_MEMORY_NO_PROMOS", "1")
+        with patch("basic_memory.cli.analytics.threading.Thread") as mock_thread:
+            track("test-event")
+            mock_thread.assert_not_called()
+
+    def test_no_op_when_not_configured(self, monkeypatch):
+        monkeypatch.delenv("BASIC_MEMORY_NO_PROMOS", raising=False)
+        monkeypatch.delenv("BASIC_MEMORY_UMAMI_HOST", raising=False)
+        monkeypatch.delenv("BASIC_MEMORY_UMAMI_SITE_ID", raising=False)
+        with patch("basic_memory.cli.analytics.threading.Thread") as mock_thread:
+            track("test-event")
+            mock_thread.assert_not_called()
+
+    def test_sends_event_when_configured(self, monkeypatch):
+        monkeypatch.delenv("BASIC_MEMORY_NO_PROMOS", raising=False)
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_HOST", "https://analytics.example.com")
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_SITE_ID", "test-site-id")
+
+        captured_target = None
+
+        def fake_thread(target, daemon):
+            nonlocal captured_target
+            captured_target = target
+            mock = MagicMock()
+            return mock
+
+        with patch("basic_memory.cli.analytics.threading.Thread", side_effect=fake_thread):
+            track(EVENT_PROMO_SHOWN, {"trigger": "first_run"})
+
+        assert captured_target is not None
+
+    def test_send_hits_correct_url(self, monkeypatch):
+        monkeypatch.delenv("BASIC_MEMORY_NO_PROMOS", raising=False)
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_HOST", "https://analytics.example.com")
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_SITE_ID", "test-site-id")
+
+        captured_request = None
+
+        def fake_urlopen(req, timeout=None):
+            nonlocal captured_request
+            captured_request = req
+            return MagicMock()
+
+        # Run the send function directly instead of in a thread
+        with patch("basic_memory.cli.analytics.urllib.request.urlopen", fake_urlopen):
+            with patch("basic_memory.cli.analytics.threading.Thread") as mock_thread:
+                # Capture the target function and call it directly
+                def run_target(target, daemon):
+                    target()  # Execute synchronously
+                    return MagicMock()
+
+                mock_thread.side_effect = run_target
+                track(EVENT_CLOUD_LOGIN_STARTED)
+
+        assert captured_request is not None
+        assert captured_request.full_url == "https://analytics.example.com/api/send"
+        body = json.loads(captured_request.data)
+        assert body["payload"]["name"] == "cli-cloud-login-started"
+        assert body["payload"]["website"] == "test-site-id"
+        assert body["payload"]["hostname"] == "cli.basicmemory.com"
+        assert "version" in body["payload"]["data"]
+
+    def test_send_failure_is_silent(self, monkeypatch):
+        monkeypatch.delenv("BASIC_MEMORY_NO_PROMOS", raising=False)
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_HOST", "https://analytics.example.com")
+        monkeypatch.setenv("BASIC_MEMORY_UMAMI_SITE_ID", "test-site-id")
+
+        def fake_urlopen(req, timeout=None):
+            raise ConnectionError("Network down")
+
+        with patch("basic_memory.cli.analytics.urllib.request.urlopen", fake_urlopen):
+            with patch("basic_memory.cli.analytics.threading.Thread") as mock_thread:
+                def run_target(target, daemon):
+                    target()  # Should not raise
+                    return MagicMock()
+
+                mock_thread.side_effect = run_target
+                # Should not raise
+                track("test-event")
+
+
+class TestEventConstants:
+    """Verify event name constants exist and are kebab-case strings."""
+
+    @pytest.mark.parametrize(
+        "event",
+        [
+            EVENT_PROMO_SHOWN,
+            EVENT_PROMO_OPTED_OUT,
+            EVENT_CLOUD_LOGIN_STARTED,
+            EVENT_CLOUD_LOGIN_SUCCESS,
+            EVENT_CLOUD_LOGIN_SUB_REQUIRED,
+        ],
+    )
+    def test_event_names_are_kebab_case(self, event):
+        assert isinstance(event, str)
+        assert event == event.lower()
+        assert " " not in event
+        assert event.startswith("cli-")