FILE: reportkit/__init__.py
```python
"""reportkit — multi-tenant report service."""
```

FILE: reportkit/schema.sql
```sql
-- reportkit schema
-- Apply with: sqlite3 reportkit.db < reportkit/schema.sql

PRAGMA journal_mode = WAL;
PRAGMA foreign_keys = ON;

-- ------------------------------------------------------------
-- Tenants
-- ------------------------------------------------------------
CREATE TABLE IF NOT EXISTS tenants (
    id          TEXT PRIMARY KEY,
    name        TEXT NOT NULL,
    suspended   INTEGER NOT NULL DEFAULT 0,  -- 0=active, 1=suspended
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

-- ------------------------------------------------------------
-- Users
-- ------------------------------------------------------------
CREATE TABLE IF NOT EXISTS users (
    id          TEXT PRIMARY KEY,
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    email       TEXT NOT NULL,
    role        TEXT NOT NULL CHECK(role IN ('viewer', 'editor', 'admin')),
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    UNIQUE(tenant_id, email)
);

CREATE INDEX IF NOT EXISTS idx_users_tenant ON users(tenant_id);

-- ------------------------------------------------------------
-- Sessions  (bearer-token → user)
-- ------------------------------------------------------------
CREATE TABLE IF NOT EXISTS sessions (
    token       TEXT PRIMARY KEY,
    user_id     TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    expires_at  TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);

-- ------------------------------------------------------------
-- Reports
-- ------------------------------------------------------------
CREATE TABLE IF NOT EXISTS reports (
    id          TEXT PRIMARY KEY,
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    owner_id    TEXT NOT NULL REFERENCES users(id)   ON DELETE CASCADE,
    title       TEXT NOT NULL,
    body        TEXT NOT NULL DEFAULT '',
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')),
    updated_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE INDEX IF NOT EXISTS idx_reports_owner   ON reports(owner_id);
CREATE INDEX IF NOT EXISTS idx_reports_tenant  ON reports(tenant_id);
CREATE INDEX IF NOT EXISTS idx_reports_created ON reports(owner_id, created_at DESC);

-- ------------------------------------------------------------
-- Attachments
-- ------------------------------------------------------------
CREATE TABLE IF NOT EXISTS attachments (
    id          TEXT PRIMARY KEY,
    report_id   TEXT NOT NULL REFERENCES reports(id) ON DELETE CASCADE,
    tenant_id   TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    filename    TEXT NOT NULL,
    mime_type   TEXT NOT NULL DEFAULT 'application/octet-stream',
    size_bytes  INTEGER NOT NULL,
    data        BLOB NOT NULL,
    created_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

CREATE INDEX IF NOT EXISTS idx_attachments_report ON attachments(report_id);

-- ------------------------------------------------------------
-- Billing ledger
-- ------------------------------------------------------------
CREATE TABLE IF NOT EXISTS billing_events (
    id              TEXT PRIMARY KEY,   -- provider's event id
    tenant_id       TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
    event_type      TEXT NOT NULL,
    amount_cents    INTEGER NOT NULL DEFAULT 0,
    payload         TEXT NOT NULL,      -- raw JSON from provider
    received_at     TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
);

-- Unique on provider id prevents double-application on retry
CREATE UNIQUE INDEX IF NOT EXISTS idx_billing_events_id ON billing_events(id);

-- ------------------------------------------------------------
-- Permissions view  (hot path — covered by indexes above)
-- ------------------------------------------------------------
-- Joining here keeps permission logic in one place; the query
-- planner uses idx_sessions_user + idx_users_tenant.
CREATE VIEW IF NOT EXISTS v_caller_permissions AS
SELECT
    s.token,
    u.id          AS user_id,
    u.tenant_id,
    u.role,
    t.suspended   AS tenant_suspended,
    CASE u.role
        WHEN 'viewer' THEN 'report:read'
        WHEN 'editor' THEN 'report:read,report:write'
        WHEN 'admin'  THEN 'report:read,report:write,report:delete,user:manage'
    END           AS permissions_csv,
    s.expires_at
FROM sessions s
JOIN users   u ON u.id        = s.user_id
JOIN tenants t ON t.id        = u.tenant_id;
```

FILE: reportkit/db.py
```python
"""Database helpers — connection factory and schema bootstrap."""

import sqlite3
import pathlib

_SCHEMA = pathlib.Path(__file__).with_name("schema.sql").read_text()


def open_db(path: str = ":memory:") -> sqlite3.Connection:
    """Return a configured sqlite3 connection.

    Row factory is set so rows behave like dicts.
    WAL mode and foreign-key enforcement are applied here as well as in the
    schema so they take effect even when the schema has already been applied.
    """
    conn = sqlite3.connect(path, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode = WAL")
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def apply_schema(conn: sqlite3.Connection) -> None:
    """Idempotently apply the DDL schema."""
    conn.executescript(_SCHEMA)
    conn.commit()
```

FILE: reportkit/errors.py
```python
"""Domain exceptions used throughout reportkit."""


class ReportkitError(Exception):
    """Base class; carries an HTTP-ish status code."""
    status: int = 500

    def __init__(self, message: str):
        super().__init__(message)
        self.message = message


class AuthenticationError(ReportkitError):
    status = 401


class PermissionError(ReportkitError):
    status = 403


class NotFoundError(ReportkitError):
    status = 404


class ConflictError(ReportkitError):
    status = 409


class ValidationError(ReportkitError):
    status = 422


class AttachmentTooLargeError(ValidationError):
    status = 413


class DisallowedContentError(ValidationError):
    status = 422


class WebhookSignatureError(ReportkitError):
    status = 400
```

FILE: reportkit/auth.py
```python
"""Session resolution and permission checking."""

import sqlite3
from datetime import datetime, timezone

from .errors import AuthenticationError, PermissionError

# Role → frozenset of permissions
_ROLE_PERMISSIONS: dict[str, frozenset[str]] = {
    "viewer": frozenset({"report:read"}),
    "editor": frozenset({"report:read", "report:write"}),
    "admin":  frozenset({"report:read", "report:write", "report:delete", "user:manage"}),
}


def _now_utc() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")


def resolve_caller(conn: sqlite3.Connection, token: str) -> dict:
    """Return caller info dict or raise AuthenticationError.

    Uses v_caller_permissions (the hot-path view) so a single query covers
    the session → user → tenant join.  The view is backed by indexes on
    sessions.user_id and users.tenant_id.
    """
    if not token:
        raise AuthenticationError("Missing bearer token")

    row = conn.execute(
        """
        SELECT user_id, tenant_id, role, tenant_suspended, expires_at
        FROM   v_caller_permissions
        WHERE  token = ?
        """,
        (token,),
    ).fetchone()

    if row is None:
        raise AuthenticationError("Invalid or expired token")

    if row["expires_at"] < _now_utc():
        raise AuthenticationError("Token has expired")

    if row["tenant_suspended"]:
        raise AuthenticationError("Tenant is suspended")

    return {
        "user_id":   row["user_id"],
        "tenant_id": row["tenant_id"],
        "role":      row["role"],
        "permissions": _ROLE_PERMISSIONS[row["role"]],
    }


def require_permission(caller: dict, permission: str) -> None:
    """Raise PermissionError if *caller* does not hold *permission*."""
    if permission not in caller["permissions"]:
        raise PermissionError(
            f"Role '{caller['role']}' does not grant '{permission}'"
        )


def _extract_token(request: dict) -> str:
    """Pull the bearer token out of request['headers']['Authorization']."""
    headers: dict = request.get("headers") or {}
    auth: str = headers.get("Authorization") or headers.get("authorization") or ""
    if auth.lower().startswith("bearer "):
        return auth[7:].strip()
    return ""
```

FILE: reportkit/events.py
```python
"""Lightweight synchronous event dispatcher.

Handlers register against a string event name.  The dispatcher calls every
registered handler in registration order; errors in one handler do not prevent
subsequent handlers from running (they are logged to stderr).

Usage
-----
    registry = EventRegistry()

    @registry.on("report.created")
    def notify(payload):
        ...

    registry.dispatch("report.created", {"report_id": "..."})
"""

import sys
from collections import defaultdict
from typing import Callable

Handler = Callable[[dict], None]


class EventRegistry:
    def __init__(self) -> None:
        self._handlers: dict[str, list[Handler]] = defaultdict(list)

    def on(self, event_name: str) -> Callable[[Handler], Handler]:
        """Decorator: register *fn* for *event_name*."""
        def decorator(fn: Handler) -> Handler:
            self._handlers[event_name].append(fn)
            return fn
        return decorator

    def register(self, event_name: str, fn: Handler) -> None:
        """Imperative alternative to the decorator."""
        self._handlers[event_name].append(fn)

    def dispatch(self, event_name: str, payload: dict) -> None:
        """Call every handler registered for *event_name*."""
        for handler in self._handlers.get(event_name, []):
            try:
                handler(payload)
            except Exception as exc:  # noqa: BLE001
                print(
                    f"[events] handler {handler!r} raised for '{event_name}': {exc}",
                    file=sys.stderr,
                )


# Module-level default registry used by the service.
default_registry = EventRegistry()
```

FILE: reportkit/scanner.py
```python
"""Attachment content scanner.

Checks uploaded bytes for disallowed content markers before storage.
This is a structural placeholder: real deployments would shell out to
ClamAV, call a vendor API, or apply more sophisticated byte-pattern
matching.  The interface is intentionally narrow so it can be swapped.
"""

from .errors import DisallowedContentError

# Byte sequences that are unconditionally rejected.
# Extend this list or replace the function body with a real scanner.
_DISALLOWED_MARKERS: list[bytes] = [
    b"EICAR-STANDARD-ANTIVIRUS-TEST-FILE",   # EICAR test string
    b"X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR",  # EICAR variant prefix
]

MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024  # 25 MB


def scan(data: bytes, filename: str = "") -> None:
    """Raise DisallowedContentError if *data* contains a disallowed marker.

    Raises AttachmentTooLargeError (a subclass of ValidationError) if the
    payload exceeds MAX_ATTACHMENT_BYTES.  Size is checked first so we never
    scan oversized blobs.
    """
    from .errors import AttachmentTooLargeError  # local to avoid circular

    if len(data) > MAX_ATTACHMENT_BYTES:
        raise AttachmentTooLargeError(
            f"Attachment '{filename}' exceeds the 25 MB limit "
            f"({len(data):,} bytes)"
        )

    for marker in _DISALLOWED_MARKERS:
        if marker in data:
            raise DisallowedContentError(
                f"Attachment '{filename}' contains disallowed content"
            )
```

FILE: reportkit/billing.py
```python
"""Billing webhook handler.

The provider POSTs signed `invoice.paid` events.  Requests are authenticated
with an HMAC-SHA256 signature over the raw request body; the shared secret is
read from the environment variable BILLING_WEBHOOK_SECRET.

Idempotency: the provider may retry deliveries it considers failed (non-2xx or
timeout).  We deduplicate on the provider's event id using a UNIQUE index on
billing_events.id, so a second delivery of the same event is a no-op (we
return 200 to stop retries).
"""

import hashlib
import hmac
import json
import os
import sqlite3
import uuid
from datetime import datetime, timezone

from .errors import ValidationError, WebhookSignatureError


def _secret() -> bytes:
    val = os.environ.get("BILLING_WEBHOOK_SECRET", "")
    if not val:
        raise RuntimeError("BILLING_WEBHOOK_SECRET is not set")
    return val.encode()


def _verify_signature(raw_body: bytes, signature_header: str) -> None:
    """Verify HMAC-SHA256(secret, raw_body) == signature_header.

    The provider sends the hex digest in the 'X-Billing-Signature' header.
    """
    expected = hmac.new(_secret(), raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature_header.lower()):
        raise WebhookSignatureError("Billing webhook signature mismatch")


def handle_webhook(conn: sqlite3.Connection, request: dict) -> dict:
    """Process an inbound billing webhook.

    request dict keys
    -----------------
    body_bytes  : bytes   — raw request body (used for signature verification)
    headers     : dict    — HTTP headers (must contain X-Billing-Signature)

    Returns a response dict with 'status' and 'body'.
    """
    raw_body: bytes = request.get("body_bytes") or b""
    headers: dict   = request.get("headers") or {}
    sig = headers.get("X-Billing-Signature") or headers.get("x-billing-signature") or ""

    if not sig:
        raise WebhookSignatureError("Missing X-Billing-Signature header")

    _verify_signature(raw_body, sig)

    try:
        payload = json.loads(raw_body)
    except json.JSONDecodeError as exc:
        raise ValidationError(f"Invalid JSON in webhook body: {exc}") from exc

    event_type  = payload.get("type", "")
    event_id    = payload.get("id") or str(uuid.uuid4())
    tenant_id   = payload.get("tenant_id", "")
    amount_cents = int(payload.get("amount_cents", 0))

    if event_type != "invoice.paid":
        # Unknown event types are acknowledged but not processed.
        return {"status": 200, "body": {"acknowledged": True, "processed": False}}

    if not tenant_id:
        raise ValidationError("invoice.paid event missing tenant_id")

    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")

    try:
        conn.execute(
            """
            INSERT INTO billing_events (id, tenant_id, event_type, amount_cents, payload, received_at)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (event_id, tenant_id, event_type, amount_cents, raw_body.decode(errors="replace"), now),
        )
        conn.commit()
    except sqlite3.IntegrityError:
        # Duplicate delivery — idempotent, return 200 to stop retries.
        return {"status": 200, "body": {"acknowledged": True, "processed": False, "duplicate": True}}

    return {"status": 200, "body": {"acknowledged": True, "processed": True}}
```

FILE: reportkit/ops.py
```python
"""Internal ops-console operations.

These functions are called by the internal ops console which runs inside the
private network.  They receive an explicit db connection and actor_id (for
audit purposes) rather than going through the HTTP surface.

purge_tenant  — delete every report (and cascade to attachments) for a tenant
set_role      — change a user's role; only admins may call this via the console
"""

import sqlite3

from .errors import NotFoundError, ValidationError

_VALID_ROLES = {"viewer", "editor", "admin"}


def purge_tenant(conn: sqlite3.Connection, actor_id: str, tenant_id: str) -> int:
    """Delete all reports (and cascading attachments) for *tenant_id*.

    Returns the number of reports deleted.  Does NOT delete the tenant row
    itself — that is a separate lifecycle decision.

    Raises NotFoundError if the tenant does not exist.
    """
    row = conn.execute(
        "SELECT id FROM tenants WHERE id = ?", (tenant_id,)
    ).fetchone()
    if row is None:
        raise NotFoundError(f"Tenant '{tenant_id}' not found")

    cursor = conn.execute(
        "DELETE FROM reports WHERE tenant_id = ?", (tenant_id,)
    )
    conn.commit()
    return cursor.rowcount


def set_role(
    conn: sqlite3.Connection,
    actor_id: str,
    user_id: str,
    new_role: str,
) -> None:
    """Set *user_id*'s role to *new_role*.

    Raises ValidationError for an unknown role, NotFoundError if the user
    does not exist.
    """
    if new_role not in _VALID_ROLES:
        raise ValidationError(
            f"Invalid role '{new_role}'. Must be one of: {', '.join(sorted(_VALID_ROLES))}"
        )

    row = conn.execute(
        "SELECT id FROM users WHERE id = ?", (user_id,)
    ).fetchone()
    if row is None:
        raise NotFoundError(f"User '{user_id}' not found")

    conn.execute(
        "UPDATE users SET role = ? WHERE id = ?", (new_role, user_id)
    )
    conn.commit()
```

FILE: reportkit/service.py
```python
"""ReportService — the HTTP surface of reportkit.

Every public method accepts a *request* dict and returns a *response* dict
with at least a ``status`` (int) and ``body`` (dict) key.  No web framework
is involved; a thin WSGI/ASGI adapter (not included) would call these methods.

request dict shape
------------------
{
    "headers": {"Authorization": "Bearer <token>", ...},
    "query":   {"sort": "title", "q": "...", ...},
    "body":    {...},           # parsed JSON for JSON endpoints
    "body_bytes": b"...",       # raw bytes for attachment upload / webhooks
}
"""

import sqlite3
import uuid
from datetime import datetime, timezone

from .auth import resolve_caller, require_permission, _extract_token
from .errors import (
    NotFoundError,
    ValidationError,
    ReportkitError,
)
from .events import default_registry
from .scanner import scan


def _now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")


def _row_to_dict(row: sqlite3.Row) -> dict:
    return dict(row)


def _ok(body: dict, status: int = 200) -> dict:
    return {"status": status, "body": body}


def _err(exc: ReportkitError) -> dict:
    return {"status": exc.status, "body": {"error": exc.message}}


_ALLOWED_SORT_COLUMNS = {"created_at", "title"}


class ReportService:
    """Stateless service object; receives a db connection at construction."""

    def __init__(self, conn: sqlite3.Connection) -> None:
        self._conn = conn

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _authenticate(self, request: dict) -> dict:
        token = _extract_token(request)
        return resolve_caller(self._conn, token)

    def _get_report_or_404(self, report_id: str, caller: dict) -> sqlite3.Row:
        row = self._conn.execute(
            "SELECT * FROM reports WHERE id = ? AND tenant_id = ?",
            (report_id, caller["tenant_id"]),
        ).fetchone()
        if row is None:
            raise NotFoundError(f"Report '{report_id}' not found")
        return row

    # ------------------------------------------------------------------
    # Public HTTP surface
    # ------------------------------------------------------------------

    def create_report(self, request: dict) -> dict:
        """POST /reports — create a report for the authenticated caller."""
        try:
            caller = self._authenticate(request)
            require_permission(caller, "report:write")

            body: dict = request.get("body") or {}
            title = (body.get("title") or "").strip()
            if not title:
                raise ValidationError("'title' is required")
            content = body.get("body") or ""

            report_id = str(uuid.uuid4())
            now = _now()

            self._conn.execute(
                """
                INSERT INTO reports (id, tenant_id, owner_id, title, body, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (report_id, caller["tenant_id"], caller["user_id"], title, content, now, now),
            )
            self._conn.commit()

            default_registry.dispatch("report.created", {
                "report_id": report_id,
                "tenant_id": caller["tenant_id"],
                "owner_id":  caller["user_id"],
                "title":     title,
            })

            return _ok(
                {
                    "id":         report_id,
                    "title":      title,
                    "body":       content,
                    "owner_id":   caller["user_id"],
                    "tenant_id":  caller["tenant_id"],
                    "created_at": now,
                    "updated_at": now,
                },
                status=201,
            )
        except ReportkitError as exc:
            return _err(exc)

    def get_report(self, request: dict, report_id: str) -> dict:
        """GET /reports/<id> — return a single report."""
        try:
            caller = self._authenticate(request)
            require_permission(caller, "report:read")
            row = self._get_report_or_404(report_id, caller)
            return _ok(_row_to_dict(row))
        except ReportkitError as exc:
            return _err(exc)

    def list_reports(self, request: dict) -> dict:
        """GET /reports — the caller's own reports, newest first."""
        try:
            caller = self._authenticate(request)
            require_permission(caller, "report:read")

            rows = self._conn.execute(
                """
                SELECT * FROM reports
                WHERE  owner_id  = ?
                  AND  tenant_id = ?
                ORDER  BY created_at DESC
                """,
                (caller["user_id"], caller["tenant_id"]),
            ).fetchall()

            return _ok({"reports": [_row_to_dict(r) for r in rows]})
        except ReportkitError as exc:
            return _err(exc)

    def search_reports(self, request: dict) -> dict:
        """GET /reports/search?q=...&sort=created_at|title

        Searches the caller's own reports by title substring.
        The ``sort`` parameter is validated against an allowlist before being
        interpolated into the ORDER BY clause, preventing SQL injection from
        the user-controlled sort control.
        """
        try:
            caller = self._authenticate(request)
            require_permission(caller, "report:read")

            query_params: dict = request.get("query") or {}
            q    = (query_params.get("q") or "").strip()
            sort = (query_params.get("sort") or "created_at").strip()

            # Validate sort against explicit allowlist — never interpolate
            # user input into SQL without this check.
            if sort not in _ALLOWED_SORT_COLUMNS:
                raise ValidationError(
                    f"Invalid sort '{sort}'. Allowed: {', '.join(sorted(_ALLOWED_SORT_COLUMNS))}"
                )

            pattern = f"%{q}%"

            rows = self._conn.execute(
                f"""
                SELECT * FROM reports
                WHERE  owner_id  = ?
                  AND  tenant_id = ?
                  AND  title LIKE ?
                ORDER  BY {sort} DESC
                """,  # sort is allowlist-validated above
                (caller["user_id"], caller["tenant_id"], pattern),
            ).fetchall()

            return _ok({"reports": [_row_to_dict(r) for r in rows]})
        except ReportkitError as exc:
            return _err(exc)

    def delete_report(self, request: dict, report_id: str) -> dict:
        """DELETE /reports/<id> — delete a report."""
        try:
            caller = self._authenticate(request)
            require_permission(caller, "report:delete")

            # Verify the report exists and belongs to this tenant.
            self._get_report_or_404(report_id, caller)

            self._conn.execute(
                "DELETE FROM reports WHERE id = ? AND tenant_id = ?",
                (report_id, caller["tenant_id"]),
            )
            self._conn.commit()

            default_registry.dispatch("report.deleted", {
                "report_id": report_id,
                "tenant_id": caller["tenant_id"],
                "actor_id":  caller["user_id"],
            })

            return _ok({"deleted": True})
        except ReportkitError as exc:
            return _err(exc)

    def upload_attachment(self, request: dict) -> dict:
        """POST /attachments — store an attachment linked to a report.

        request body keys (parsed from multipart or JSON envelope):
            report_id : str
            filename  : str
            mime_type : str  (optional)

        request["body_bytes"] : raw file bytes
        """
        try:
            caller = self._authenticate(request)
            require_permission(caller, "report:write")

            body: dict  = request.get("body") or {}
            report_id   = (body.get("report_id") or "").strip()
            filename    = (body.get("filename") or "upload").strip()
            mime_type   = (body.get("mime_type") or "application/octet-stream").strip()
            data: bytes = request.get("body_bytes") or b""

            if not report_id:
                raise ValidationError("'report_id' is required")

            # Confirm the report exists and belongs to this tenant.
            self._get_report_or_404(report_id, caller)

            # Scan before storing — raises on disallowed content or oversize.
            scan(data, filename)

            attachment_id = str(uuid.uuid4())
            now = _now()

            self._conn.execute(
                """
                INSERT INTO attachments
                    (id, report_id, tenant_id, filename, mime_type, size_bytes, data, created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    attachment_id,
                    report_id,
                    caller["tenant_id"],
                    filename,
                    mime_type,
                    len(data),
                    data,
                    now,
                ),
            )
            self._conn.commit()

            return _ok(
                {
                    "id":         attachment_id,
                    "report_id":  report_id,
                    "filename":   filename,
                    "mime_type":  mime_type,
                    "size_bytes": len(data),
                    "created_at": now,
                },
                status=201,
            )
        except ReportkitError as exc:
            return _err(exc)
```

FILE: reportkit/fixtures.py
```python
"""Test fixture helpers — create tenants, users, sessions, reports in a db."""

import uuid
from datetime import datetime, timedelta, timezone


def _now_plus(seconds: int = 3600) -> str:
    dt = datetime.now(timezone.utc) + timedelta(seconds=seconds)
    return dt.strftime("%Y-%m-%dT%H:%M:%S.000Z")


def make_tenant(conn, name: str = "Acme") -> str:
    tid = str(uuid.uuid4())
    conn.execute(
        "INSERT INTO tenants (id, name) VALUES (?, ?)", (tid, name)
    )
    conn.commit()
    return tid


def make_user(conn, tenant_id: str, role: str = "editor", email: str | None = None) -> str:
    uid = str(uuid.uuid4())
    email = email or f"{uid[:8]}@example.com"
    conn.execute(
        "INSERT INTO users (id, tenant_id, email, role) VALUES (?, ?, ?, ?)",
        (uid, tenant_id, email, role),
    )
    conn.commit()
    return uid


def make_session(conn, user_id: str, expires_in: int = 3600) -> str:
    token = str(uuid.uuid4())
    conn.execute(
        "INSERT INTO sessions (token, user_id, expires_at) VALUES (?, ?, ?)",
        (token, user_id, _now_plus(expires_in)),
    )
    conn.commit()
    return token


def make_report(conn, owner_id: str, tenant_id: str, title: str = "Report") -> str:
    rid = str(uuid.uuid4())
    conn.execute(
        """
        INSERT INTO reports (id, tenant_id, owner_id, title, body)
        VALUES (?, ?, ?, ?, '')
        """,
        (rid, tenant_id, owner_id, title),
    )
    conn.commit()
    return rid


def bearer(token: str) -> dict:
    """Return a minimal request dict with an Authorization header."""
    return {"headers": {"Authorization": f"Bearer {token}"}}
```

FILE: tests/__init__.py
```python
```

FILE: tests/conftest.py
```python
"""Shared pytest fixtures."""

import pytest

from reportkit.db import open_db, apply_schema
from reportkit.service import ReportService
from reportkit.fixtures import make_tenant, make_user, make_session, make_report


@pytest.fixture()
def conn():
    db = open_db(":memory:")
    apply_schema(db)
    return db


@pytest.fixture()
def svc(conn):
    return ReportService(conn)


# ---------------------------------------------------------------------------
# A ready-made tenant with one user of each role
# ---------------------------------------------------------------------------

@pytest.fixture()
def tenant(conn):
    return make_tenant(conn, "TestCo")


@pytest.fixture()
def viewer_token(conn, tenant):
    uid = make_user(conn, tenant, role="viewer")
    return make_session(conn, uid)


@pytest.fixture()
def editor_token(conn, tenant):
    uid = make_user(conn, tenant, role="editor")
    return make_session(conn, uid)


@pytest.fixture()
def admin_token(conn, tenant):
    uid = make_user(conn, tenant, role="admin")
    return make_session(conn, uid)
```

FILE: tests/test_auth.py
```python
"""Authentication and permission tests."""

import pytest

from reportkit.auth import resolve_caller, require_permission
from reportkit.errors import AuthenticationError, PermissionError
from reportkit.db import open_db, apply_schema
from reportkit.fixtures import make_tenant, make_user, make_session


@pytest.fixture()
def conn():
    db = open_db(":memory:")
    apply_schema(db)
    return db


def test_resolve_caller_success(conn):
    tid = make_tenant(conn)
    uid = make_user(conn, tid, role="editor")
    token = make_session(conn, uid)

    caller = resolve_caller(conn, token)
    assert caller["user_id"] == uid
    assert caller["tenant_id"] == tid
    assert caller["role"] == "editor"
    assert "report:read" in caller["permissions"]
    assert "report:write" in caller["permissions"]
    assert "report:delete" not in caller["permissions"]


def test_resolve_caller_bad_token(conn):
    with pytest.raises(AuthenticationError):
        resolve_caller(conn, "not-a-real-token")


def test_resolve_caller_empty_token(conn):
    with pytest.raises(AuthenticationError):
        resolve_caller(conn, "")


def test_resolve_caller_expired(conn):
    tid = make_tenant(conn)
    uid = make_user(conn, tid, role="viewer")
    # Insert an already-expired session
    token = "expired-token"
    conn.execute(
        "INSERT INTO sessions (token, user_id, expires_at) VALUES (?, ?, ?)",
        (token, uid, "2000-01-01T00:00:00.000Z"),
    )
    conn.commit()
    with pytest.raises(AuthenticationError, match="expired"):
        resolve_caller(conn, token)


def test_resolve_caller_suspended_tenant(conn):
    tid = make_tenant(conn)
    conn.execute("UPDATE tenants SET suspended = 1 WHERE id = ?", (tid,))
    conn.commit()
    uid = make_user(conn, tid, role="admin")
    token = make_session(conn, uid)
    with pytest.raises(AuthenticationError, match="suspended"):
        resolve_caller(conn, token)


def test_viewer_permissions(conn):
    tid = make_tenant(conn)
    uid = make_user(conn, tid, role="viewer")
    token = make_session(conn, uid)
    caller = resolve_caller(conn, token)
    assert caller["permissions"] == frozenset({"report:read"})


def test_admin_permissions(conn):
    tid = make_tenant(conn)
    uid = make_user(conn, tid, role="admin")
    token = make_session(conn, uid)
    caller = resolve_caller(conn, token)
    assert {"report:read", "report:write", "report:delete", "user:manage"}.issubset(
        caller["permissions"]
    )


def test_require_permission_ok(conn):
    tid = make_tenant(conn)
    uid = make_user(conn, tid, role="editor")
    token = make_session(conn, uid)
    caller = resolve_caller(conn, token)
    # Should not raise
    require_permission(caller, "report:read")
    require_permission(caller, "report:write")


def test_require_permission_denied(conn):
    tid = make_tenant(conn)
    uid = make_user(conn, tid, role="editor")
    token = make_session(conn, uid)
    caller = resolve_caller(conn, token)
    with pytest.raises(PermissionError):
        require_permission(caller, "report:delete")
```

FILE: tests/test_reports.py
```python
"""Tests for create_report, get_report, list_reports, delete_report."""

import pytest

from reportkit.fixtures import make_report, bearer
from reportkit.errors import PermissionError


def test_create_report_success(svc, editor_token):
    req = {**bearer(editor_token), "body": {"title": "Q1 Summary", "body": "content"}}
    resp = svc.create_report(req)
    assert resp["status"] == 201
    assert resp["body"]["title"] == "Q1 Summary"
    assert "id" in resp["body"]


def test_create_report_viewer_forbidden(svc, viewer_token):
    req = {**bearer(viewer_token), "body": {"title": "Nope"}}
    resp = svc.create_report(req)
    assert resp["status"] == 403


def test_create_report_missing_title(svc, editor_token):
    req = {**bearer(editor_token), "body": {"title": "  "}}
    resp = svc.create_report(req)
    assert resp["status"] == 422


def test_create_report_no_auth(svc):
    req = {"headers": {}, "body": {"title": "X"}}
    resp = svc.create_report(req)
    assert resp["status"] == 401


def test_get_report_success(svc, conn, editor_token, tenant):
    uid = conn.execute(
        "SELECT user_id FROM sessions WHERE token = ?", (editor_token,)
    ).fetchone()["user_id"]
    rid = make_report(conn, uid, tenant, "My Report")

    resp = svc.get_report(bearer(editor_token), rid)
    assert resp["status"] == 200
    assert resp["body"]["id"] == rid
    assert resp["body"]["title"] == "My Report"


def test_get_report_wrong_tenant(svc, conn, editor_token):
    """A report from another tenant is not visible."""
    from reportkit.fixtures import make_tenant, make_user
    other_tenant = make_tenant(conn, "OtherCo")
    other_user   = make_user(conn, other_tenant, role="editor")
    rid = make_report(conn, other_user, other_tenant, "Secret")

    resp = svc.get_report(bearer(editor_token), rid)
    assert resp["status"] == 404


def test_get_report_not_found(svc, editor_token):
    resp = svc.get_report(bearer(editor_token), "no-such-id")
    assert resp["status"] == 404


def test_list_reports_empty(svc, editor_token):
    resp = svc.list_reports(bearer(editor_token))
    assert resp["status"] == 200
    assert resp["body"]["reports"] == []


def test_list_reports_newest_first(svc, conn, editor_token, tenant):
    uid = conn.execute(
        "SELECT user_id FROM sessions WHERE token = ?", (editor_token,)
    ).fetchone()["user_id"]

    import time
    r1 = make_report(conn, uid, tenant, "First")
    time.sleep(0.01)
    r2 = make_report(conn, uid, tenant, "Second")

    resp = svc.list_reports(bearer(editor_token))
    ids = [r["id"] for r in resp["body"]["reports"]]
    assert ids.index(r2) < ids.index(r1)


def test_list_reports_only_own(svc, conn, editor_token, tenant):
    """Viewer's reports do not appear in editor's list."""
    from reportkit.fixtures import make_user, make_session
    other_uid = make_user(conn, tenant, role="viewer")
    make_report(conn, other_uid, tenant, "Other's report")

    resp = svc.list_reports(bearer(editor_token))
    assert resp["body"]["reports"] == []


def test_delete_report_admin(svc, conn, admin_token, tenant):
    uid = conn.execute(
        "SELECT user_id FROM sessions WHERE token = ?", (admin_token,)
    ).fetchone()["user_id"]
    rid = make_report(conn, uid, tenant, "To delete")

    resp = svc.delete_report(bearer(admin_token), rid)
    assert resp["status"] == 200
    assert resp["body"]["deleted"] is True

    # Confirm gone
    resp2 = svc.get_report(bearer(admin_token), rid)
    assert resp2["status"] == 404


def test_delete_report_editor_forbidden(svc, conn, editor_token, tenant):
    uid = conn.execute(
        "SELECT user_id FROM sessions WHERE token = ?", (editor_token,)
    ).fetchone()["user_id"]
    rid = make_report(conn, uid, tenant, "Protected")

    resp = svc.delete_report(bearer(editor_token), rid)
    assert resp["status"] == 403


def test_delete_report_not_found(svc, admin_token):
    resp = svc.delete_report(bearer(admin_token), "ghost-id")
    assert resp["status"] == 404
```

FILE: tests/test_search.py
```python
"""Tests for search_reports."""

import pytest
from reportkit.fixtures import make_report, bearer


def _uid(conn, token):
    return conn.execute(
        "SELECT user_id FROM sessions WHERE token = ?", (token,)
    ).fetchone()["user_id"]


def test_search_finds_matching(svc, conn, editor_token, tenant):
    uid = _uid(conn, editor_token)
    make_report(conn, uid, tenant, "Annual Budget")
    make_report(conn, uid, tenant, "Monthly Expenses")

    req = {**bearer(editor_token), "query": {"q": "Budget"}}
    resp = svc.search_reports(req)
    assert resp["status"] == 200
    titles = [r["title"] for r in resp["body"]["reports"]]
    assert "Annual Budget" in titles
    assert "Monthly Expenses" not in titles


def test_search_empty_query_returns_all(svc, conn, editor_token, tenant):
    uid = _uid(conn, editor_token)
    make_report(conn, uid, tenant, "Alpha")
    make_report(conn, uid, tenant, "Beta")

    req = {**bearer(editor_token), "query": {"q": ""}}
    resp = svc.search_reports(req)
    assert len(resp["body"]["reports"]) == 2


def test_search_sort_by_title(svc, conn, editor_token, tenant):
    uid = _uid(conn, editor_token)
    make_report(conn, uid, tenant, "Zebra")
    make_report(conn, uid, tenant, "Apple")

    req = {**bearer(editor_token), "query": {"q": "", "sort": "title"}}
    resp = svc.search_reports(req)
    titles = [r["title"] for r in resp["body"]["reports"]]
    assert titles == sorted(titles, reverse=True)


def test_search_invalid_sort_rejected(svc, editor_token):
    req = {**bearer(editor_token), "query": {"sort": "'; DROP TABLE reports; --"}}
    resp = svc.search_reports(req)
    assert resp["status"] == 422


def test_search_sort_default_is_created_at(svc, conn, editor_token, tenant):
    uid = _uid(conn, editor_token)
    import time
    make_report(conn, uid, tenant, "First")
    time.sleep(0.01)
    make_report(conn, uid, tenant, "Second")

    req = {**bearer(editor_token), "query": {}}
    resp = svc.search_reports(req)
    titles = [r["title"] for r in resp["body"]["reports"]]
    assert titles[0] == "Second"


def test_search_no_cross_tenant(svc, conn, editor_token):
    from reportkit.fixtures import make_tenant, make_user
    other = make_tenant(conn, "Other")
    ouid  = make_user(conn, other, role="editor")
    make_report(conn, ouid, other, "Secret Report")

    req = {**bearer(editor_token), "query": {"q": "Secret"}}
    resp = svc.search_reports(req)
    assert resp["body"]["reports"] == []
```

FILE: tests/test_attachments.py
```python
"""Tests for upload_attachment."""

import pytest
from reportkit.fixtures import make_report, bearer
from reportkit.scanner import MAX_ATTACHMENT_BYTES


def _uid(conn, token):
    return conn.execute(
        "SELECT user_id FROM sessions WHERE token = ?", (token,)
    ).fetchone()["user_id"]


def test_upload_success(svc, conn, editor_token, tenant):
    uid = _uid(conn, editor_token)
    rid = make_report(conn, uid, tenant)

    req = {
        **bearer(editor_token),
        "body":       {"report_id": rid, "filename": "data.csv", "mime_type": "text/csv"},
        "body_bytes": b"col1,col2\n1,2\n",
    }
    resp = svc.upload_attachment(req)
    assert resp["status"] == 201
    assert resp["body"]["filename"] == "data.csv"
    assert resp["body"]["size_bytes"] == len(b"col1,col2\n1,2\n")


def test_upload_viewer_forbidden(svc, conn, viewer_token, tenant):
    uid = _uid(conn, viewer_token)
    # Viewer cannot write, so make the report via SQL directly
    rid = make_report(conn, uid, tenant)

    req = {
        **bearer(viewer_token),
        "body":       {"report_id": rid, "filename": "x.txt"},
        "body_bytes": b"hello",
    }
    resp = svc.upload_attachment(req)
    assert resp["status"] == 403


def test_upload_too_large(svc, conn, editor_token, tenant):
    uid = _uid(conn, editor_token)
    rid = make_report(conn, uid, tenant)

    big = b"x" * (MAX_ATTACHMENT_BYTES + 1)
    req = {
        **bearer(editor_token),
        "body":       {"report_id": rid, "filename": "big.bin"},
        "body_bytes": big,
    }
    resp = svc.upload_attachment(req)
    assert resp["status"] == 413


def test_upload_disallowed_content(svc, conn, editor_token, tenant):
    uid = _uid(conn, editor_token)
    rid = make_report(conn, uid, tenant)

    payload = b"EICAR-STANDARD-ANTIVIRUS-TEST-FILE"
    req = {
        **bearer(editor_token),
        "body":       {"report_id": rid, "filename": "evil.txt"},
        "body_bytes": payload,
    }
    resp = svc.upload_attachment(req)
    assert resp["status"] == 422


def test_upload_missing_report_id(svc, editor_token):
    req = {
        **bearer(editor_token),
        "body":       {"filename": "oops.txt"},
        "body_bytes": b"data",
    }
    resp = svc.upload_attachment(req)
    assert resp["status"] == 422


def test_upload_report_wrong_tenant(svc, conn, editor_token):
    from reportkit.fixtures import make_tenant, make_user
    other = make_tenant(conn, "Other")
    ouid  = make_user(conn, other, role="editor")
    rid   = make_report(conn, ouid, other)

    req = {
        **bearer(editor_token),
        "body":       {"report_id": rid, "filename": "f.txt"},
        "body_bytes": b"data",
    }
    resp = svc.upload_attachment(req)
    assert resp["status"] == 404
```

FILE: tests/test_billing.py
```python
"""Tests for the billing webhook handler."""

import hashlib
import hmac
import json
import os
import uuid

import pytest

from reportkit.billing import handle_webhook
from reportkit.errors import WebhookSignatureError, ValidationError
from reportkit.db import open_db, apply_schema
from reportkit.fixtures import make_tenant


SECRET = "test-billing-secret"


@pytest.fixture(autouse=True)
def set_secret(monkeypatch):
    monkeypatch.setenv("BILLING_WEBHOOK_SECRET", SECRET)


@pytest.fixture()
def conn():
    db = open_db(":memory:")
    apply_schema(db)
    return db


def _signed_request(payload: dict, secret: str = SECRET) -> dict:
    raw = json.dumps(payload).encode()
    sig = hmac.new(secret.encode(), raw, hashlib.sha256).hexdigest()
    return {
        "body_bytes": raw,
        "headers":    {"X-Billing-Signature": sig},
    }


def test_invoice_paid_accepted(conn):
    tid = make_tenant(conn, "BillCo")
    event_id = str(uuid.uuid4())
    payload = {
        "id":           event_id,
        "type":         "invoice.paid",
        "tenant_id":    tid,
        "amount_cents": 4999,
    }
    resp = handle_webhook(conn, _signed_request(payload))
    assert resp["status"] == 200
    assert resp["body"]["processed"] is True


def test_invoice_paid_idempotent(conn):
    tid = make_tenant(conn, "BillCo2")
    event_id = str(uuid.uuid4())
    payload = {
        "id":           event_id,
        "type":         "invoice.paid",
        "tenant_id":    tid,
        "amount_cents": 100,
    }
    req = _signed_request(payload)
    handle_webhook(conn, req)
    resp2 = handle_webhook(conn, req)
    assert resp2["status"] == 200
    assert resp2["body"].get("duplicate") is True


def test_bad_signature_rejected(conn):
    payload = {"id": "x", "type": "invoice.paid", "tenant_id": "t", "amount_cents": 0}
    raw = json.dumps(payload).encode()
    req = {"body_bytes": raw, "headers": {"X-Billing-Signature": "deadbeef"}}
    with pytest.raises(WebhookSignatureError):
        handle_webhook(conn, req)


def test_missing_signature_rejected(conn):
    payload = {"id": "x", "type": "invoice.paid", "tenant_id": "t"}
    raw = json.dumps(payload).encode()
    req = {"body_bytes": raw, "headers": {}}
    with pytest.raises(WebhookSignatureError):
        handle_webhook(conn, req)


def test_unknown_event_type_acknowledged(conn):
    payload = {"id": str(uuid.uuid4()), "type": "invoice.voided", "tenant_id": "t"}
    resp = handle_webhook(conn, _signed_request(payload))
    assert resp["status"] == 200
    assert resp["body"]["processed"] is False


def test_missing_tenant_id_raises(conn):
    payload = {"id": str(uuid.uuid4()), "type": "invoice.paid", "amount_cents": 0}
    with pytest.raises(ValidationError):
        handle_webhook(conn, _signed_request(payload))
```

FILE: tests/test_ops.py
```python
"""Tests for internal ops-console functions."""

import pytest

from reportkit.ops import purge_tenant, set_role
from reportkit.errors import NotFoundError, ValidationError
from reportkit.db import open_db, apply_schema
from reportkit.fixtures import make_tenant, make_user, make_report


@pytest.fixture()
def conn():
    db = open_db(":memory:")
    apply_schema(db)
    return db


def test_purge_tenant_removes_reports(conn):
    tid = make_tenant(conn, "PurgeCo")
    uid = make_user(conn, tid, role="editor")
    make_report(conn, uid, tid, "R1")
    make_report(conn, uid, tid, "R2")

    count = purge_tenant(conn, actor_id="ops", tenant_id=tid)
    assert count == 2

    remaining = conn.execute(
        "SELECT COUNT(*) AS n FROM reports WHERE tenant_id = ?", (tid,)
    ).fetchone()["n"]
    assert remaining == 0


def test_purge_tenant_not_found(conn):
    with pytest.raises(NotFoundError):
        purge_tenant(conn, actor_id="ops", tenant_id="ghost")


def test_purge_tenant_zero_reports(conn):
    tid = make_tenant(conn, "EmptyCo")
    count = purge_tenant(conn, actor_id="ops", tenant_id=tid)
    assert count == 0


def test_set_role_success(conn):
    tid = make_tenant(conn)
    uid = make_user(conn, tid, role="viewer")

    set_role(conn, actor_id="ops", user_id=uid, new_role="admin")

    row = conn.execute("SELECT role FROM users WHERE id = ?", (uid,)).fetchone()
    assert row["role"] == "admin"


def test_set_role_invalid_role(conn):
    tid = make_tenant(conn)
    uid = make_user(conn, tid, role="viewer")
    with pytest.raises(ValidationError):
        set_role(conn, actor_id="ops", user_id=uid, new_role="superuser")


def test_set_role_user_not_found(conn):
    with pytest.raises(NotFoundError):
        set_role(conn, actor_id="ops", user_id="ghost", new_role="editor")
```

FILE: tests/test_events.py
```python
"""Tests for the event registry."""

import pytest
from reportkit.events import EventRegistry


def test_handler_called(capsys):
    reg = EventRegistry()
    received = []

    @reg.on("report.created")
    def h(payload):
        received.append(payload)

    reg.dispatch("report.created", {"id": "1"})
    assert received == [{"id": "1"}]


def test_multiple_handlers_all_called():
    reg = EventRegistry()
    log = []

    reg.register("x", lambda p: log.append("a"))
    reg.register("x", lambda p: log.append("b"))

    reg.dispatch("x", {})
    assert log == ["a", "b"]


def test_no_handlers_no_error():
    reg = EventRegistry()
    reg.dispatch("unknown.event", {})  # should not raise


def test_failing_handler_does_not_block_next(capsys):
    reg = EventRegistry()
    log = []

    def bad(p):
        raise RuntimeError("boom")

    def good(p):
        log.append("ok")

    reg.register("ev", bad)
    reg.register("ev", good)

    reg.dispatch("ev", {})
    assert log == ["ok"]
    captured = capsys.readouterr()
    assert "boom" in captured.err


def test_dispatch_report_created_and_deleted():
    """Smoke-test the two application events."""
    reg = EventRegistry()
    seen = []
    reg.on("report.created")(lambda p: seen.append(("created", p["report_id"])))
    reg.on("report.deleted")(lambda p: seen.append(("deleted", p["report_id"])))

    reg.dispatch("report.created", {"report_id": "r1"})
    reg.dispatch("report.deleted", {"report_id": "r1"})

    assert seen == [("created", "r1"), ("deleted", "r1")]
```

FILE: tests/test_scanner.py
```python
"""Tests for the attachment content scanner."""

import pytest
from reportkit.scanner import scan, MAX_ATTACHMENT_BYTES
from reportkit.errors import DisallowedContentError, AttachmentTooLargeError


def test_clean_file_passes():
    scan(b"hello, world", "hello.txt")  # should not raise


def test_eicar_rejected():
    with pytest.raises(DisallowedContentError):
        scan(b"EICAR-STANDARD-ANTIVIRUS-TEST-FILE", "test.txt")


def test_oversized_rejected():
    big = b"a" * (MAX_ATTACHMENT_BYTES + 1)
    with pytest.raises(AttachmentTooLargeError):
        scan(big, "big.bin")


def test_exactly_max_size_passes():
    ok = b"a" * MAX_ATTACHMENT_BYTES
    scan(ok, "max.bin")  # should not raise


def test_empty_file_passes():
    scan(b"", "empty.txt")
```

FILE: NOTES.md
```markdown
# reportkit — design notes

## Decisions

### No web framework
The service exposes a plain class (`ReportService`) whose methods accept and
return plain dicts. A thin adapter (WSGI, ASGI, Lambda handler) would sit in
front; that layer is intentionally out of scope so the core is framework-
agnostic and trivially testable without HTTP machinery.

### Authentication via view
The hot-path permission check (session → user → tenant) is expressed as
`v_caller_permissions`, a SQLite view that collapses the three-table join into
a single named query. `resolve_caller` queries the view by token; SQLite uses
`idx_sessions_user` + `idx_users_tenant` to satisfy it. Because roles change
infrequently (ops console only) there is no application-level cache — the view
is fast enough and avoids stale-cache bugs.

### Permissions are derived, not stored
The set of permissions for a role is a static mapping in `auth.py`
(`_ROLE_PERMISSIONS`). The view also materialises a `permissions_csv` column
for observability, but the authoritative check is done in Python against the
frozenset. Adding a new permission requires only a code change, not a data
migration.

### SQL injection prevention in `search_reports`
The `sort` parameter comes directly from the UI. It is validated against an
explicit allowlist (`_ALLOWED_SORT_COLUMNS`) before being interpolated into the
`ORDER BY` clause. All other user-supplied values use parameterised queries.

### Billing webhook idempotency
The provider retries deliveries it considers failed. We deduplicate on the
provider's `id` field via a `UNIQUE INDEX` on `billing_events.id`. A second
delivery of the same event hits an `IntegrityError` and returns 200 (so the
provider stops retrying) without double-applying the payment.

### Attachment scanning
`scanner.py` contains a structural placeholder that checks for the EICAR test
string. The interface (`scan(data, filename) → None | raise`) is intentionally
narrow: swap the body for a ClamAV socket call or a vendor HTTP call without
changing any call sites. Size is checked before scanning so we never load a
25 MB blob into a real scanner unnecessarily.

### `tenant.quota_exceeded` handler
The spec mentions this handler is wanted for a batch-import feed that was
retired last quarter. I did **not** implement it. The `EventRegistry` is ready
to accept `reg.on("tenant.quota_exceeded")(handler)` when the feed returns;
adding it is a one-function change. Shipping dead code that suspends tenants
carries more risk than the stub being absent.

### `purge_tenant` / `set_role` (ops console)
These live in `ops.py` and receive an explicit `db` connection rather than
going through the HTTP surface. The ops console runs inside the private network
and is trusted; it does not need bearer-token auth. `actor_id` is accepted for
future audit-log integration.

### Multi-tenancy isolation
Every query that touches reports or attachments includes `tenant_id = ?`
alongside the primary key filter. `get_report_or_404` enforces this; a caller
from tenant A cannot read, delete, or attach to a report owned by tenant B even
if they guess the UUID.

### SQLite in WAL mode
WAL mode allows concurrent readers without blocking a writer, which suits a
service where reads dominate. `PRAGMA foreign_keys = ON` is set both in the
schema and in `open_db` so it is active regardless of whether the schema has
already been applied.

## Trade-offs / deliberately not done

- **No connection pool.** A single `sqlite3.Connection` is passed into the
  service. A real deployment would use a pool or switch to PostgreSQL; the
  interface does not need to change.
- **No pagination.** `list_reports` and `search_reports` return all matching
  rows. Cursor-based pagination would be a straightforward addition.
- **No audit log table.** `actor_id` is threaded through ops functions as a
  placeholder. A production system would `INSERT` into an `audit_events` table.
- **Attachments stored as BLOBs.** For simplicity, file bytes go into SQLite.
  A production system would store them in object storage (S3/GCS) and keep only
  a reference in the database.
- **No HTTPS / TLS.** That is the responsibility of the infrastructure layer
  (reverse proxy, load balancer).
- **No rate limiting.** Would live in the adapter layer, not the service layer.
```