feat: share publish telemetry across services

pmoves/docs/NEXT_STEPS.md

@@ -2,7 +2,7 @@
 
 _Last updated: 2025-09-26 (geometry cache sync)_
 
-_Last updated: 2025-10-01_
+_Last updated: 2025-10-05_
 
 
 ## Immediate
@@ -13,15 +13,15 @@ _Last updated: 2025-10-01_
 - [ ] Activate the n8n approval poller and echo publisher workflows once secrets are loaded; document the activation + first successful run.
 - [ ] Confirm Jellyfin credentials (API key and optional user id) allow library enumeration; note any dependency gaps that require new guardrails.
 - [ ] Validate that enriched publisher metadata propagates into Agent Zero and Discord events; schedule a backfill for legacy records if fields are missing.
-- [ ] Hit the publisher `/metrics` endpoint and capture the turnaround/latency summary for the runbook.
-- [ ] Confirm Supabase `publisher_metrics_rollup` rows are created with engagement + cost payloads and link the ROI dashboard query.
+- [ ] Hit the publisher and publisher-discord `/metrics` endpoints and capture the turnaround/latency summary for the runbook.
+- [ ] Confirm Supabase `publisher_metrics_rollup` and `publisher_discord_metrics` rows are created with engagement + cost payloads and link the ROI dashboard query.
 - [ ] Record step-by-step evidence in `SESSION_IMPLEMENTATION_PLAN.md` while executing the operational reminders list.
 
 ### 2. Jellyfin Publisher Reliability
 - [x] Add a scheduled refresh or webhook trigger so Jellyfin libraries update after publisher runs; include cron/webhook settings in `services/publisher/README.md`.
 - [ ] Expand error/reporting hooks so failures surface with actionable messages (Jellyfin HTTP errors, dependency mismatches, asset gaps).
 - [ ] Backfill historic Jellyfin entries with enriched metadata and confirm downstream consumers (Agent Zero, Discord) render the new fields.
-- [ ] Plot baseline ROI visuals (turnaround vs engagement vs cost) using the Supabase rollup table and document interpretation guidance in the dashboard notes.
+- [ ] Plot baseline ROI visuals (turnaround vs engagement vs cost) using the Supabase rollup tables and incorporate the guidance captured in `TELEMETRY_ROI.md` into the dashboard notes.
 
 ### 3. Graph & Retrieval Enhancements (Kickoff M3)
 - [ ] Seed Neo4j with the brand alias dictionary (DARKXSIDE, POWERFULMOVES, plus pending community submissions) and record Cypher script locations (draft plan in `SESSION_IMPLEMENTATION_PLAN.md`).

pmoves/docs/ROADMAP.md

@@ -1,5 +1,5 @@
 # PMOVES v5 • ROADMAP
-_Last updated: 2025-10-01_
+_Last updated: 2025-10-05_
 
 ## Vision
 A production-ready, self-hostable orchestration mesh for creative + agent workloads across GPU boxes and Jetsons: **hybrid Hi‑RAG**, **Supabase Studio**, **n8n orchestration**, **Jellyfin publishing**, and **graph-aware retrieval**.
@@ -19,15 +19,15 @@ A production-ready, self-hostable orchestration mesh for creative + agent worklo
 | ✅ | ComfyUI ↔ MinIO Presign microservice | `services/presign/api.py` provides presigned PUT/GET/POST helpers for MinIO/S3. |
 | ✅ | Render Webhook (Comfy → Supabase Studio) | `services/render-webhook/webhook.py` inserts submissions into `studio_board` with optional auto-approval. |
 | 🚧 | Publisher (Jellyfin) | `services/publisher/publisher.py` consumes approval events and refreshes Jellyfin; optional dependency guards and envelope fallback landed, but richer metadata handling and error reporting are still pending. |
-| ✅ | Publisher telemetry & ROI rollups | `/metrics` endpoint plus Supabase rollups from `services/publisher/publisher.py` make turnaround/latency/cost telemetry queryable for dashboards. |
+| ✅ | Publisher telemetry & ROI rollups | `/metrics` feeds from `services/publisher/publisher.py` and `services/publisher-discord/main.py` expose turnaround/latency/cost telemetry, with Supabase rollups powering the ROI dashboards documented in `pmoves/docs/TELEMETRY_ROI.md`. |
 | ✅ | PDF/MinIO ingestion | `services/pdf-ingest/app.py` pulls PDFs from MinIO, extracts text, forwards chunks, and emits ingest events. |
 | ⏳ | n8n flows (Discord/webhooks) | `n8n/flows/*.json` only define placeholder workflows; Supabase pollers and Discord actions must be configured. |
 | 🚧 | Jellyfin library refresh hook + Discord rich cards | Jellyfin refresh occurs in the publisher, and `services/publisher-discord` formats embeds, but published-event wiring and asset deep links remain. Automation activation plan logged in `pmoves/docs/SESSION_IMPLEMENTATION_PLAN.md`. |
 
 **Outstanding to close M2:**
 - publisher metadata/envelope polish — namespace-aware filenames, dependency guards, and fallback envelopes merged; monitor adoption and backfill historic assets if needed
 - add published-event Discord embeds via `content.published.v1`; execution plan staged in `SESSION_IMPLEMENTATION_PLAN.md`
-- wire Supabase ROI dashboards to the new publisher telemetry rollups; document interpretation guidance alongside ROI reporting.
+- wire Supabase ROI dashboards to the new publisher telemetry rollups; document interpretation guidance alongside ROI reporting (**see `docs/TELEMETRY_ROI.md` for the latest walkthrough**).
 - build the Supabase→Discord automation inside the n8n exports and track discrete workflow validation steps in the implementation log
 - execute the Supabase → Agent Zero → Discord activation checklist (`pmoves/docs/SUPABASE_DISCORD_AUTOMATION.md`) and log the validation timestamp (see operational reminders captured in the implementation plan)
 

pmoves/docs/TELEMETRY_ROI.md

@@ -0,0 +1,64 @@
+# Telemetry & ROI Dashboards
+
+_Last updated: 2025-10-05_
+
+## Overview
+
+Publisher-facing services now expose real-time telemetry so operators can
+correlate turnaround, approval latency, engagement, and cost. Two lightweight
+HTTP endpoints provide quick snapshots while Supabase rollup tables persist the
+event-level detail required for historical dashboards.
+
+| Service | Endpoint | Purpose |
+| --- | --- | --- |
+| `services/publisher/publisher.py` | `http://<host>:9095/metrics` | Aggregated counts/averages for artifact downloads, Jellyfin refreshes, turnaround, approval latency, engagement, and cost inputs. |
+| `services/publisher-discord/main.py` | `http://<host>:<app-port>/metrics` | Discord delivery counters plus the same turnaround/approval telemetry derived from published events. |
+
+Both endpoints return JSON with `telemetry` blocks shaped like:
+
+```json
+{
+  "turnaround_samples": 12,
+  "avg_turnaround_seconds": 5400.25,
+  "approval_latency_samples": 12,
+  "avg_approval_latency_seconds": 900.5,
+  "engagement_totals": {"views": 1800, "ctr": 12.5},
+  "cost_totals": {"storage_gb": 6.2, "processing_minutes": 44.0}
+}
+```
+
+Use the metrics endpoints for smoke tests and runbooks; dashboards should query
+Supabase directly to avoid scraping service processes.
+
+## Supabase Rollups
+
+| Table | Source | Description |
+| --- | --- | --- |
+| `publisher_metrics_rollup` | `services/publisher/publisher.py` | One row per published artifact, keyed by `artifact_uri`, capturing turnaround/approval latency plus engagement and cost payloads forwarded from approvals. |
+| `publisher_discord_metrics` | `services/publisher-discord/main.py` | Mirrors publisher telemetry for Discord notifications and adds `webhook_success` + `channel` flags to monitor downstream delivery health. |
+
+Both tables are upserted using `services/common/supabase.py::upsert_row`. Use
+`PUBLISHER_METRICS_CONFLICT` and `DISCORD_METRICS_CONFLICT` environment
+variables to tune the `ON CONFLICT` keys when running migrations.
+
+## Interpreting ROI Dashboards
+
+1. **Turnaround vs. Approval Latency** – Plot `avg_turnaround_seconds` and
+   `avg_approval_latency_seconds` trends. Rising approval latency usually points
+   to reviewer bottlenecks; rising turnaround means ingest-to-publish automation
+   needs tuning.
+2. **Engagement Ratios** – Divide `engagement_totals` (views, CTR, likes) by the
+   corresponding `cost_totals` (processing minutes, storage GB, egress). High
+   engagement with low cost indicates good ROI; low engagement + high cost flags
+   candidates for pruning or repackaging.
+3. **Channel Health** – Join `publisher_metrics_rollup` and
+   `publisher_discord_metrics` on `artifact_uri` (or slug) to confirm Discord
+   notifications land for the same assets that hit Jellyfin. Investigate rows
+   where `webhook_success` is false to catch stale credentials or rate limits.
+4. **Namespace Drill-down** – Group by `namespace` to understand which creator
+   lanes drive the most engagement per unit of spend. Feed these insights into
+   the sprint prioritisation matrix.
+
+Document snapshot queries alongside dashboards so the next operator can run
+validations quickly (see `docs/NEXT_STEPS.md` for the checklist).
+

pmoves/services/common/telemetry.py

@@ -0,0 +1,228 @@
+"""Shared telemetry helpers for publisher-style services."""
+
+from __future__ import annotations
+
+import datetime as _dt
+from dataclasses import asdict, dataclass, field
+from typing import Any, Dict, Optional, Tuple
+
+
+@dataclass
+class PublisherMetrics:
+    """Aggregate metrics for publisher style services."""
+
+    downloads: int = 0
+    download_failures: int = 0
+    refresh_attempts: int = 0
+    refresh_success: int = 0
+    refresh_failures: int = 0
+    turnaround_samples: int = 0
+    total_turnaround_seconds: float = 0.0
+    max_turnaround_seconds: float = 0.0
+    approval_latency_samples: int = 0
+    total_approval_latency_seconds: float = 0.0
+    max_approval_latency_seconds: float = 0.0
+    engagement_events: int = 0
+    engagement_totals: Dict[str, float] = field(default_factory=dict)
+    cost_events: int = 0
+    cost_totals: Dict[str, float] = field(default_factory=dict)
+
+    def record_download_success(self) -> None:
+        self.downloads += 1
+
+    def record_download_failure(self) -> None:
+        self.download_failures += 1
+
+    def record_refresh_attempt(self) -> None:
+        self.refresh_attempts += 1
+
+    def record_refresh_success(self) -> None:
+        self.refresh_success += 1
+
+    def record_refresh_failure(self) -> None:
+        self.refresh_failures += 1
+
+    def record_turnaround(self, seconds: Optional[float]) -> None:
+        if seconds is None or seconds < 0:
+            return
+        self.turnaround_samples += 1
+        self.total_turnaround_seconds += seconds
+        if seconds > self.max_turnaround_seconds:
+            self.max_turnaround_seconds = seconds
+
+    def record_approval_latency(self, seconds: Optional[float]) -> None:
+        if seconds is None or seconds < 0:
+            return
+        self.approval_latency_samples += 1
+        self.total_approval_latency_seconds += seconds
+        if seconds > self.max_approval_latency_seconds:
+            self.max_approval_latency_seconds = seconds
+
+    def record_engagement(self, engagement: Dict[str, float]) -> None:
+        if not engagement:
+            return
+        self.engagement_events += 1
+        for key, value in engagement.items():
+            try:
+                numeric = float(value)
+            except (TypeError, ValueError):
+                continue
+            self.engagement_totals[key] = self.engagement_totals.get(key, 0.0) + numeric
+
+    def record_cost(self, cost: Dict[str, float]) -> None:
+        if not cost:
+            return
+        self.cost_events += 1
+        for key, value in cost.items():
+            try:
+                numeric = float(value)
+            except (TypeError, ValueError):
+                continue
+            self.cost_totals[key] = self.cost_totals.get(key, 0.0) + numeric
+
+    def summary(self) -> Dict[str, Any]:
+        data = asdict(self)
+        if self.turnaround_samples:
+            data["avg_turnaround_seconds"] = self.total_turnaround_seconds / self.turnaround_samples
+        if self.approval_latency_samples:
+            data["avg_approval_latency_seconds"] = (
+                self.total_approval_latency_seconds / self.approval_latency_samples
+            )
+        return data
+
+
+@dataclass
+class PublishTelemetry:
+    published_at: _dt.datetime
+    turnaround_seconds: Optional[float]
+    approval_latency_seconds: Optional[float]
+    engagement: Dict[str, float]
+    cost: Dict[str, float]
+
+    def to_meta(self) -> Dict[str, Any]:
+        meta: Dict[str, Any] = {
+            "published_at": self.published_at.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
+        }
+        if self.turnaround_seconds is not None:
+            meta["turnaround_seconds"] = self.turnaround_seconds
+        if self.approval_latency_seconds is not None:
+            meta["approval_to_publish_seconds"] = self.approval_latency_seconds
+        if self.engagement:
+            meta["engagement"] = self.engagement
+        if self.cost:
+            meta["cost"] = self.cost
+        return meta
+
+    def to_rollup_row(self, *, artifact_uri: str, namespace: str, slug: str) -> Dict[str, Any]:
+        return {
+            "artifact_uri": artifact_uri,
+            "namespace": namespace,
+            "slug": slug,
+            "published_at": self.published_at.isoformat(),
+            "turnaround_seconds": self.turnaround_seconds,
+            "approval_latency_seconds": self.approval_latency_seconds,
+            "engagement": self.engagement or None,
+            "cost": self.cost or None,
+        }
+
+
+def _parse_iso8601(value: Optional[Any]) -> Optional[_dt.datetime]:
+    if not value or not isinstance(value, str):
+        return None
+    value = value.strip()
+    if not value:
+        return None
+    try:
+        if value.endswith("Z"):
+            value = value[:-1] + "+00:00"
+        return _dt.datetime.fromisoformat(value)
+    except ValueError:
+        return None
+
+
+def _coerce_numeric(value: Any) -> Optional[float]:
+    try:
+        if value is None:
+            return None
+        numeric = float(value)
+        if numeric != numeric:  # NaN
+            return None
+        return numeric
+    except (TypeError, ValueError):
+        return None
+
+
+def _extract_first(meta: Dict[str, Any], keys: Tuple[str, ...]) -> Optional[str]:
+    for key in keys:
+        candidate = meta.get(key)
+        if isinstance(candidate, str) and candidate:
+            return candidate
+    return None
+
+
+def compute_publish_telemetry(
+    incoming_meta: Optional[Dict[str, Any]],
+    event_ts: Optional[str],
+    published_at: _dt.datetime,
+) -> PublishTelemetry:
+    meta = incoming_meta or {}
+    start_keys = (
+        "ingest_started_at",
+        "submitted_at",
+        "created_at",
+        "capture_completed_at",
+    )
+    approval_keys = (
+        "approval_granted_at",
+        "approved_at",
+        "approval_completed_at",
+    )
+
+    start_ts = _parse_iso8601(_extract_first(meta, start_keys))
+    approval_ts = _parse_iso8601(_extract_first(meta, approval_keys))
+    event_timestamp = _parse_iso8601(event_ts)
+
+    turnaround_seconds: Optional[float] = None
+    if start_ts is not None:
+        turnaround_seconds = (published_at - start_ts).total_seconds()
+
+    approval_latency_seconds: Optional[float] = None
+    reference_ts = approval_ts or event_timestamp
+    if reference_ts is not None:
+        approval_latency_seconds = (published_at - reference_ts).total_seconds()
+
+    engagement: Dict[str, float] = {}
+    for key in ("engagement", "analytics", "metrics"):
+        candidate = meta.get(key)
+        if isinstance(candidate, dict):
+            for metric_key, value in candidate.items():
+                numeric = _coerce_numeric(value)
+                if numeric is None:
+                    continue
+                engagement[metric_key] = engagement.get(metric_key, 0.0) + numeric
+
+    cost: Dict[str, float] = {}
+    for key in ("cost", "spend", "usage"):
+        candidate = meta.get(key)
+        if isinstance(candidate, dict):
+            for cost_key, value in candidate.items():
+                numeric = _coerce_numeric(value)
+                if numeric is None:
+                    continue
+                cost[cost_key] = cost.get(cost_key, 0.0) + numeric
+
+    return PublishTelemetry(
+        published_at=published_at,
+        turnaround_seconds=turnaround_seconds,
+        approval_latency_seconds=approval_latency_seconds,
+        engagement=engagement,
+        cost=cost,
+    )
+
+
+__all__ = [
+    "PublisherMetrics",
+    "PublishTelemetry",
+    "compute_publish_telemetry",
+]
+