feat(api): add a read-only API key and a /api/stats endpoint (#1419)

This is the read-only API key from #1410, where you said to go ahead.

I run Shelfmark behind Homepage and wanted more on the dashboard tile
than up or down, without putting an admin API key in the dashboard's
config. This adds a second key, `SHELFMARK_API_KEY_READONLY`, that can
read one new endpoint.

`GET /api/stats` returns counts only: books added over the last 7 and 30
days by format, the queue, requests by outcome, and download failures
over the last 7 days. No titles and no user names. The read-only key,
the admin key or an admin session can read it.

The read-only key gets a 403 on every other path and on any write,
including a POST to `/api/stats` itself, and it never gets a session or
a cookie. If a request carries both keys, the admin one wins, so a
proxy's own `Authorization` header can't downgrade a correct
`X-Api-Key`. Unset means off, like the existing key. The endpoint exists
either way, but without a key only an admin session can read it.

It also adds a `db_path` property on `UserDB`, so the stats module can
open the database read-only.

Tests cover the key scope, the 403s for the read-only key, both keys at
once, who can reach the endpoint, and the counters. The full suite
passes, and I broke each rule on purpose to check a test catches it. A
version of this has been running on my own install behind Homepage.

If you'd rather have the stats in a different shape or under a different
path, I'm happy to change it.

docs/api-access.md

@@ -70,3 +70,16 @@ curl -s -H "Authorization: Bearer $SHELFMARK_API_KEY" https://shelfmark.example.
 - The key is a root-equivalent credential: an admin can configure a custom
   post-download script that the server executes, so treat it like a root
   password and send it only over HTTPS.
+
+## Read-only key and dashboard stats
+
+Set `SHELFMARK_API_KEY_READONLY` to a second secret for dashboards and monitors. It can read one endpoint,
+`GET /api/stats`, and nothing else. Writes and every other path return 403, and the key never gets a session.
+The admin key reaches `/api/stats` too.
+
+```bash
+curl -s -H "X-Api-Key: $SHELFMARK_API_KEY_READONLY" https://shelfmark.example.com/api/stats
+```
+
+The response has counts only: books added in the last 7 and 30 days by format, the queue, requests by outcome,
+and download failures in the last 7 days. No titles and no user names.

docs/environment-variables.md

@@ -49,6 +49,7 @@ These environment variables are used at startup before the settings system loads
 | `HIDE_LOCAL_AUTH` | Hide the username/password login form when OIDC is active. | boolean | `false` |
 | `DISABLE_LOCAL_AUTH` | Disable username/password login and remove the local-admin prerequisite for OIDC. Implies HIDE_LOCAL_AUTH; with AUTH_METHOD=builtin, everyone is locked out until auth env vars are changed. | boolean | `false` |
 | `SHELFMARK_API_KEY` | Optional static API key. When set, requests carrying it as 'Authorization: Bearer <key>' (or X-Api-Key) are authenticated as an admin; browser sessions keep working. Unset = off. | string | `unset` |
+| `SHELFMARK_API_KEY_READONLY` | Optional read-only API key. Can only GET `/api/stats`; never an admin. Unset = off. | string | `unset` |
 | `OIDC_AUTO_REDIRECT` | Automatically redirect to the OIDC provider instead of showing the login page. | boolean | `false` |
 | `DOCKERMODE` | Indicates the application is running inside a Docker container. | boolean | `false` |
 | `ONBOARDING` | Show the onboarding wizard on first run. Set to false to skip (useful for ephemeral storage). | boolean | `true` |
@@ -133,6 +134,13 @@ Optional static API key. When set, requests carrying it as 'Authorization: Beare
 - **Type:** string
 - **Default:** `unset`
 
+#### `SHELFMARK_API_KEY_READONLY`
+
+Optional second API key for dashboards. Send it like `SHELFMARK_API_KEY`. It can only `GET /api/stats` (counters only, no titles or user names). Every other request, and any write, is refused with 403, and it never gets a session. Unset = off.
+
+- **Type:** string
+- **Default:** `unset`
+
 #### `OIDC_AUTO_REDIRECT`
 
 Automatically redirect to the OIDC provider instead of showing the login page.

shelfmark/config/env.py

@@ -169,6 +169,8 @@ DISABLE_LOCAL_AUTH = string_to_bool(os.getenv("DISABLE_LOCAL_AUTH", "false"))
 # Optional static API key. When set, requests carrying it as a Bearer token
 # (or X-Api-Key) are authenticated as an admin for that request only.
 SHELFMARK_API_KEY = os.getenv("SHELFMARK_API_KEY", "").strip()
+# Optional second key that can only read /api/stats (dashboards). Never an admin.
+SHELFMARK_API_KEY_READONLY = os.getenv("SHELFMARK_API_KEY_READONLY", "").strip()
 OIDC_AUTO_REDIRECT = string_to_bool(os.getenv("OIDC_AUTO_REDIRECT", "false"))
 
 

shelfmark/core/api_key.py

@@ -12,7 +12,7 @@ from __future__ import annotations
 
 import hmac
 
-from shelfmark.config.env import SHELFMARK_API_KEY
+from shelfmark.config.env import SHELFMARK_API_KEY, SHELFMARK_API_KEY_READONLY
 
 
 def extract_api_key_candidates(
@@ -37,3 +37,20 @@ def matches_api_key(candidate: str) -> bool:
     if not SHELFMARK_API_KEY or not candidate:
         return False
     return hmac.compare_digest(candidate.encode("utf-8"), SHELFMARK_API_KEY.encode("utf-8"))
+
+
+def key_scope(candidate: str) -> str | None:
+    """``"admin"``, ``"readonly"`` or None for a presented credential.
+
+    The read-only key (``SHELFMARK_API_KEY_READONLY``) is for dashboards: it reaches
+    ``/api/stats`` and nothing else. Admin wins if both are set to the same value.
+    """
+    if not candidate:
+        return None
+    if matches_api_key(candidate):
+        return "admin"
+    if SHELFMARK_API_KEY_READONLY and hmac.compare_digest(
+        candidate.encode("utf-8"), SHELFMARK_API_KEY_READONLY.encode("utf-8")
+    ):
+        return "readonly"
+    return None

shelfmark/core/stats.py

@@ -0,0 +1,97 @@
+"""Dashboard counters for ``/api/stats``: what was added, what is queued, what went wrong.
+
+Counts only, no titles or user names, read straight from the request and history tables
+on a read-only connection.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+_FORMATS = ("ebook", "audiobook")
+
+
+def _since(now: datetime, days: int) -> str:
+    return (now - timedelta(days=days)).isoformat()
+
+
+def _added(conn: sqlite3.Connection, now: datetime) -> dict[str, int]:
+    out: dict[str, int] = {}
+    for days in (7, 30):
+        rows = conn.execute(
+            """
+            SELECT LOWER(COALESCE(content_type, '')) AS kind, COUNT(*) AS n
+            FROM download_history
+            WHERE final_status = 'complete' AND datetime(terminal_at) >= datetime(?)
+            GROUP BY kind
+            """,
+            (_since(now, days),),
+        ).fetchall()
+        by_kind = {row["kind"]: int(row["n"]) for row in rows}
+        for kind in _FORMATS:
+            out[f"{kind}_{days}d"] = by_kind.get(kind, 0)
+        out[f"total_{days}d"] = sum(by_kind.values())
+    return out
+
+
+def _count(conn: sqlite3.Connection, sql: str, params: tuple[Any, ...] = ()) -> int:
+    row = conn.execute(sql, params).fetchone()
+    return int(row[0]) if row else 0
+
+
+def collect(db_path: str, *, now: datetime | None = None) -> dict[str, Any]:
+    """Counters for the dashboard, as of ``now`` (UTC)."""
+    now = now or datetime.now(UTC)
+    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
+    conn.row_factory = sqlite3.Row
+    try:
+        week = _since(now, 7)
+        awaiting = _count(
+            conn,
+            "SELECT COUNT(*) FROM download_requests WHERE status = 'pending' AND delivery_state = 'none'",
+        )
+        queued = _count(
+            conn, "SELECT COUNT(*) FROM download_requests WHERE delivery_state = 'queued'"
+        )
+        by_delivery = {
+            state: _count(
+                conn, "SELECT COUNT(*) FROM download_requests WHERE delivery_state = ?", (state,)
+            )
+            for state in ("complete", "error", "cancelled")
+        }
+        return {
+            "generated_at": now.isoformat(),
+            "added": _added(conn, now),
+            "queue": {
+                "active": _count(
+                    conn, "SELECT COUNT(*) FROM download_history WHERE final_status = 'active'"
+                ),
+                "awaiting_pickup": awaiting,
+                "queued": queued,
+            },
+            "requests": {
+                "pending": _count(
+                    conn, "SELECT COUNT(*) FROM download_requests WHERE status = 'pending'"
+                ),
+                "delivered": by_delivery["complete"],
+                "failed": by_delivery["error"],
+                "cancelled": by_delivery["cancelled"],
+                "rejected": _count(
+                    conn, "SELECT COUNT(*) FROM download_requests WHERE status = 'rejected'"
+                ),
+            },
+            "errors": {
+                "failed_7d": _count(
+                    conn,
+                    """
+                    SELECT COUNT(*) FROM download_history
+                    WHERE final_status = 'error' AND datetime(terminal_at) >= datetime(?)
+                    """,
+                    (week,),
+                ),
+            },
+        }
+    finally:
+        conn.close()

shelfmark/core/user_db.py

@@ -188,6 +188,11 @@ class UserDB:
         self._db_path = db_path
         self._lock = threading.Lock()
 
+    @property
+    def db_path(self) -> str:
+        """Filesystem path of the SQLite database."""
+        return self._db_path
+
     def _connect(self) -> sqlite3.Connection:
         conn = sqlite3.connect(self._db_path)
         conn.row_factory = sqlite3.Row

shelfmark/main.py

@@ -728,6 +728,7 @@ def _proxy_default_is_admin(db: UserDB) -> bool:
 
 _API_KEY_EXEMPT_PREFIXES = ("/api/auth/",)
 _API_KEY_EXEMPT_PATHS = frozenset({"/api/health"})
+_READ_ONLY_KEY_PATHS = frozenset({"/api/stats"})
 
 
 @app.before_request
@@ -746,7 +747,7 @@ def api_key_auth_middleware() -> Response | tuple[Response, int] | None:
         return None
     if request.path in _API_KEY_EXEMPT_PATHS or request.path.startswith(_API_KEY_EXEMPT_PREFIXES):
         return None
-    if not api_key_module.SHELFMARK_API_KEY:
+    if not api_key_module.SHELFMARK_API_KEY and not api_key_module.SHELFMARK_API_KEY_READONLY:
         return None
 
     candidates = api_key_module.extract_api_key_candidates(
@@ -755,7 +756,9 @@ def api_key_auth_middleware() -> Response | tuple[Response, int] | None:
     if not candidates:
         return None
 
-    if not any(api_key_module.matches_api_key(candidate) for candidate in candidates):
+    scopes = {api_key_module.key_scope(candidate) for candidate in candidates}
+    scope = "admin" if "admin" in scopes else "readonly" if "readonly" in scopes else None
+    if scope is None:
         return None
     if get_auth_mode() == "none":
         return None
@@ -764,6 +767,14 @@ def api_key_auth_middleware() -> Response | tuple[Response, int] | None:
     # reset also covers the error path below.
     g.api_key_auth = True
 
+    if scope == "readonly":
+        # A read-only key is a dashboard credential: one GET endpoint, no session.
+        g.api_key_scope = "readonly"
+        if request.method != "GET" or request.path not in _READ_ONLY_KEY_PATHS:
+            return jsonify({"error": "This API key is read-only"}), 403
+        return None
+    g.api_key_scope = "admin"
+
     try:
         admin = user_db.get_first_admin() if user_db is not None else None
     except _OPERATIONAL_ERRORS:
@@ -1353,6 +1364,28 @@ def api_config() -> Response | tuple[Response, int]:
         return jsonify({"error": str(e)}), 500
 
 
+@app.route("/api/stats", methods=["GET"])
+def api_stats() -> Response | tuple[Response, int]:
+    """Counters for dashboards. Reachable with the read-only API key, an admin key or an admin session."""
+    from shelfmark.core import stats
+
+    scope = g.get("api_key_scope")
+    if get_auth_mode() != "none" and scope is None:
+        if "user_id" not in session:
+            return jsonify({"error": "Unauthorized"}), 401
+        if not session.get("is_admin", False):
+            return jsonify({"error": "Admin access required"}), 403
+
+    if user_db is None:
+        return jsonify({"error": "User database unavailable"}), 503
+    try:
+        payload = stats.collect(user_db.db_path)
+    except sqlite3.Error:
+        logger.exception("stats: could not read the database")
+        return jsonify({"error": "Internal Server Error"}), 500
+    return jsonify(payload)
+
+
 @app.route("/api/health", methods=["GET"])
 def api_health() -> Response | tuple[Response, int]:
     """Health check endpoint for container orchestration.

tests/core/test_api_key_env.py

@@ -421,3 +421,132 @@ class TestIdentityAndRouting:
         assert response.status_code == 400
         assert response.get_json() == {"error": "source_id is required"}
         assert "Set-Cookie" not in response.headers
+
+
+class TestKeyScope:
+    def test_admin_key(self, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "adm")
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "ro")
+        assert api_key.key_scope("adm") == "admin"
+
+    def test_read_only_key(self, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "adm")
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "ro")
+        assert api_key.key_scope("ro") == "readonly"
+
+    def test_unknown_and_empty(self, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "adm")
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "ro")
+        assert api_key.key_scope("nope") is None
+        assert api_key.key_scope("") is None
+
+    def test_unset_keys_never_match_an_empty_candidate(self, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "")
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "")
+        assert api_key.key_scope("") is None
+        assert api_key.key_scope("x") is None
+
+    def test_same_value_in_both_is_admin(self, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "same")
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "same")
+        assert api_key.key_scope("same") == "admin"
+
+    def test_read_only_key_alone(self, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "")
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "ro")
+        assert api_key.key_scope("ro") == "readonly"
+
+
+class TestReadOnlyKeyRequests:
+    @pytest.fixture
+    def ro(self, wired, monkeypatch):
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "ro-key")
+        return wired
+
+    def test_reads_stats(self, ro, user_db):
+        user_db.create_user(username="root", role="admin")
+        response = ro.app.test_client().get("/api/stats", headers=_bearer("ro-key"))
+        assert response.status_code == 200
+
+    def test_x_api_key_works_too(self, ro, user_db):
+        user_db.create_user(username="root", role="admin")
+        response = ro.app.test_client().get("/api/stats", headers=_x_api_key("ro-key"))
+        assert response.status_code == 200
+
+    @pytest.mark.parametrize(
+        "path", ["/api/settings", "/api/downloads/active", "/api/users", "/api/requests"]
+    )
+    def test_cannot_read_anything_else(self, ro, user_db, path):
+        user_db.create_user(username="root", role="admin")
+        response = ro.app.test_client().get(path, headers=_bearer("ro-key"))
+        assert response.status_code == 403
+
+    @pytest.mark.parametrize("method", ["post", "put", "patch", "delete"])
+    def test_cannot_write_even_to_stats(self, ro, user_db, method):
+        user_db.create_user(username="root", role="admin")
+        response = getattr(ro.app.test_client(), method)("/api/stats", headers=_bearer("ro-key"))
+        assert response.status_code == 403
+
+    def test_never_gets_an_admin_session(self, ro, user_db):
+        user_db.create_user(username="root", role="admin")
+        with ro.app.test_request_context("/api/stats", headers=_bearer("ro-key")):
+            assert ro.api_key_auth_middleware() is None
+            from flask import session
+
+            assert "is_admin" not in session
+            assert "user_id" not in session
+
+    def test_sets_no_cookie(self, ro, user_db):
+        user_db.create_user(username="root", role="admin")
+        response = ro.app.test_client().get("/api/stats", headers=_bearer("ro-key"))
+        assert "Set-Cookie" not in response.headers
+
+    def test_admin_key_still_reads_stats(self, ro, user_db):
+        user_db.create_user(username="root", role="admin")
+        response = ro.app.test_client().get("/api/stats", headers=_bearer("s3cret"))
+        assert response.status_code == 200
+
+    def test_admin_key_beside_a_read_only_key_keeps_admin_access(self, ro, user_db):
+        """A proxy's own Authorization value must not downgrade a correct admin X-Api-Key."""
+        user_db.create_user(username="root", role="admin")
+        headers = {**_bearer("ro-key"), **_x_api_key("s3cret")}
+        assert ro.app.test_client().get("/api/settings", headers=headers).status_code == 200
+
+    def test_no_key_is_refused(self, ro, user_db):
+        user_db.create_user(username="root", role="admin")
+        assert ro.app.test_client().get("/api/stats").status_code == 401
+
+    def test_wrong_key_is_refused(self, ro, user_db):
+        user_db.create_user(username="root", role="admin")
+        response = ro.app.test_client().get("/api/stats", headers=_bearer("guess"))
+        assert response.status_code == 401
+
+    def test_read_only_key_alone_enables_the_middleware(self, main_module, user_db, monkeypatch):
+        monkeypatch.setattr(main_module, "user_db", user_db)
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY", "")
+        monkeypatch.setattr(api_key, "SHELFMARK_API_KEY_READONLY", "ro-key")
+        with patch.object(main_module, "get_auth_mode", return_value="builtin"):
+            client = main_module.app.test_client()
+            assert client.get("/api/stats", headers=_bearer("ro-key")).status_code == 200
+            assert client.get("/api/settings", headers=_bearer("ro-key")).status_code == 403
+
+
+class TestStatsEndpoint:
+    def test_stats_payload_carries_counters(self, wired, user_db):
+        user_db.create_user(username="root", role="admin")
+        body = wired.app.test_client().get("/api/stats", headers=_bearer("s3cret")).get_json()
+
+        assert set(body) == {"generated_at", "added", "queue", "requests", "errors"}
+        assert body["added"]["total_7d"] == 0
+
+    def test_stats_is_closed_to_a_plain_non_admin_session(self, wired, user_db):
+        user = user_db.create_user(username="reader", role="user")
+        client = _cookie_client(wired.app, user, is_admin=False)
+
+        assert client.get("/api/stats").status_code == 403
+
+    def test_stats_is_open_to_an_admin_session(self, wired, user_db):
+        admin = user_db.create_user(username="root", role="admin")
+        client = _cookie_client(wired.app, admin, is_admin=True)
+
+        assert client.get("/api/stats").status_code == 200

tests/core/test_stats.py

@@ -0,0 +1,189 @@
+"""Tests for the dashboard counters behind /api/stats."""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import tempfile
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from shelfmark.core import stats
+from shelfmark.core.download_history_service import DownloadHistoryService
+from shelfmark.core.user_db import UserDB
+
+NOW = datetime(2026, 10, 1, 12, 0, tzinfo=UTC)
+
+
+@pytest.fixture
+def env():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        path = os.path.join(tmpdir, "users.db")
+        db = UserDB(path)
+        db.initialize()
+        user = db.create_user(username="reader", role="user")
+        yield db, DownloadHistoryService(path), user, path
+
+
+def _history(env, task, content_type, status, *, age_days=0.0, message=None):
+    _db, history, user, path = env
+    history.record_download(
+        task_id=task,
+        user_id=user["id"],
+        username="reader",
+        request_id=None,
+        source="direct_download",
+        source_display_name="x",
+        title=task,
+        author="a",
+        file_format="epub",
+        size=None,
+        preview=None,
+        content_type=content_type,
+        downloads=None,
+        origin="requested",
+    )
+    if status != "active":
+        history.finalize_download(task_id=task, final_status=status, status_message=message)
+    stamp = (NOW - timedelta(days=age_days)).isoformat()
+    conn = sqlite3.connect(path)
+    conn.execute("UPDATE download_history SET terminal_at = ? WHERE task_id = ?", (stamp, task))
+    conn.commit()
+    conn.close()
+
+
+def _request(env, content_type="ebook", *, status="pending", delivery="none"):
+    db, _history_svc, user, _path = env
+    row = db.create_request(
+        user_id=user["id"],
+        content_type=content_type,
+        request_level="book",
+        policy_mode="request_book",
+        book_data={"title": "t", "provider": "hardcover", "provider_id": str(id(object()))},
+    )
+    conn = sqlite3.connect(env[3])
+    conn.execute(
+        "UPDATE download_requests SET status = ?, delivery_state = ? WHERE id = ?",
+        (status, delivery, row["id"]),
+    )
+    conn.commit()
+    conn.close()
+
+
+def test_an_empty_database_reports_zeroes(env):
+    result = stats.collect(env[3], now=NOW)
+
+    assert result["added"] == {
+        "ebook_7d": 0,
+        "audiobook_7d": 0,
+        "total_7d": 0,
+        "ebook_30d": 0,
+        "audiobook_30d": 0,
+        "total_30d": 0,
+    }
+    assert result["queue"] == {"active": 0, "awaiting_pickup": 0, "queued": 0}
+    assert result["requests"] == {
+        "pending": 0,
+        "delivered": 0,
+        "failed": 0,
+        "cancelled": 0,
+        "rejected": 0,
+    }
+    assert result["errors"] == {"failed_7d": 0}
+    assert result["generated_at"] == NOW.isoformat()
+
+
+def test_added_counts_completed_downloads_by_format_and_window(env):
+    _history(env, "e1", "ebook", "complete", age_days=1)
+    _history(env, "e2", "ebook", "complete", age_days=6.9)
+    _history(env, "e3", "ebook", "complete", age_days=20)
+    _history(env, "a1", "audiobook", "complete", age_days=2)
+    _history(env, "a2", "audiobook", "complete", age_days=29)
+    _history(env, "old", "audiobook", "complete", age_days=31)
+
+    added = stats.collect(env[3], now=NOW)["added"]
+
+    assert added["ebook_7d"] == 2
+    assert added["audiobook_7d"] == 1
+    assert added["total_7d"] == 3
+    assert added["ebook_30d"] == 3
+    assert added["audiobook_30d"] == 2
+    assert added["total_30d"] == 5
+
+
+def test_added_ignores_unfinished_and_failed_downloads(env):
+    _history(env, "x1", "ebook", "error", age_days=1, message="boom")
+    _history(env, "x2", "ebook", "cancelled", age_days=1)
+    _history(env, "x3", "ebook", "active")
+
+    assert stats.collect(env[3], now=NOW)["added"]["total_7d"] == 0
+
+
+def test_an_unknown_content_type_counts_in_the_total_only(env):
+    _history(env, "c1", "comic", "complete", age_days=1)
+
+    added = stats.collect(env[3], now=NOW)["added"]
+
+    assert (added["ebook_7d"], added["audiobook_7d"], added["total_7d"]) == (0, 0, 1)
+
+
+def test_active_downloads_are_the_queue(env):
+    _history(env, "q1", "ebook", "active")
+    _history(env, "q2", "audiobook", "active")
+    _history(env, "done", "ebook", "complete")
+
+    assert stats.collect(env[3], now=NOW)["queue"]["active"] == 2
+
+
+def test_requests_are_bucketed_by_status_and_delivery(env):
+    _request(env)  # pending, nothing started
+    _request(env)
+    _request(env, delivery="queued")
+    _request(env, status="fulfilled", delivery="complete")
+    _request(env, status="fulfilled", delivery="error")
+    _request(env, status="fulfilled", delivery="cancelled")
+    _request(env, status="rejected")
+
+    result = stats.collect(env[3], now=NOW)
+
+    assert result["queue"]["awaiting_pickup"] == 2
+    assert result["queue"]["queued"] == 1
+    assert result["requests"] == {
+        "pending": 3,
+        "delivered": 1,
+        "failed": 1,
+        "cancelled": 1,
+        "rejected": 1,
+    }
+
+
+def test_failures_are_counted_over_the_last_seven_days_only(env):
+    _history(env, "f1", "ebook", "error", age_days=1, message="source said no")
+    _history(env, "f2", "ebook", "error", age_days=3, message="timeout")
+    _history(env, "old", "ebook", "error", age_days=8, message="timeout")
+    _history(env, "ok", "ebook", "complete", age_days=1)
+
+    assert stats.collect(env[3], now=NOW)["errors"] == {"failed_7d": 2}
+
+
+def test_an_error_without_a_message_is_a_failure(env):
+    _history(env, "f1", "ebook", "error", age_days=1)
+
+    assert stats.collect(env[3], now=NOW)["errors"]["failed_7d"] == 1
+
+
+def test_the_database_is_opened_read_only(env, monkeypatch):
+    opened = []
+    real = sqlite3.connect
+
+    def spy(target, *args, **kwargs):
+        opened.append(str(target))
+        return real(target, *args, **kwargs)
+
+    monkeypatch.setattr(stats.sqlite3, "connect", spy)
+
+    stats.collect(env[3], now=NOW)
+
+    assert opened
+    assert all("mode=ro" in target for target in opened)