diff --git a/dashboard/src/components/layout/nav-config.ts b/dashboard/src/components/layout/nav-config.ts index a1e5c56e..7b1d8dfe 100644 --- a/dashboard/src/components/layout/nav-config.ts +++ b/dashboard/src/components/layout/nav-config.ts @@ -8,6 +8,7 @@ import { LayoutDashboard, ListTree, type LucideIcon, + Radio, ScrollText, Server, Settings2, @@ -39,6 +40,7 @@ export const NAV: NavGroup[] = [ { to: "/metrics", label: "Metrics", icon: BarChart3 }, { to: "/logs", label: "Logs", icon: ScrollText }, { to: "/workflows", label: "Workflows", icon: GitBranch }, + { to: "/topics", label: "Topics", icon: Radio }, ], }, { diff --git a/dashboard/src/features/topics/api.ts b/dashboard/src/features/topics/api.ts new file mode 100644 index 00000000..b2d28140 --- /dev/null +++ b/dashboard/src/features/topics/api.ts @@ -0,0 +1,13 @@ +import { api } from "@/lib/api-client"; +import type { SubscriptionBacklog, TopicSummary } from "@/lib/api-types"; + +export function fetchTopics(signal?: AbortSignal): Promise { + return api.get("/api/topics", { signal }); +} + +export function fetchTopicDetail( + topic: string, + signal?: AbortSignal, +): Promise { + return api.get(`/api/topics/${encodeURIComponent(topic)}`, { signal }); +} diff --git a/dashboard/src/features/topics/components/index.ts b/dashboard/src/features/topics/components/index.ts new file mode 100644 index 00000000..ee96bc6e --- /dev/null +++ b/dashboard/src/features/topics/components/index.ts @@ -0,0 +1,2 @@ +export { SubscriptionsTable } from "./subscriptions-table"; +export { TopicsTable } from "./topics-table"; diff --git a/dashboard/src/features/topics/components/subscriptions-table.tsx b/dashboard/src/features/topics/components/subscriptions-table.tsx new file mode 100644 index 00000000..8cb188e8 --- /dev/null +++ b/dashboard/src/features/topics/components/subscriptions-table.tsx @@ -0,0 +1,148 @@ +import type { ColumnDef } from "@tanstack/react-table"; +import { Radio } from "lucide-react"; +import { useMemo } from "react"; +import { Badge, DataTable, EmptyState, ErrorState, TableSkeleton } from "@/components/ui"; +import type { SubscriptionBacklog } from "@/lib/api-types"; +import { formatCount } from "@/lib/number"; +import { formatDuration } from "@/lib/time"; + +interface SubscriptionsTableProps { + subscriptions: SubscriptionBacklog[] | undefined; + loading: boolean; + error: Error | null; + onRetry: () => void; +} + +const NUM_CELL = "block w-full text-right font-mono text-[0.82rem] tabular-nums"; + +export function SubscriptionsTable({ + subscriptions, + loading, + error, + onRetry, +}: SubscriptionsTableProps) { + const rows = useMemo( + () => + subscriptions + ? [...subscriptions].sort((a, b) => a.subscription.localeCompare(b.subscription)) + : [], + [subscriptions], + ); + + const columns = useMemo[]>( + () => [ + { + accessorKey: "subscription", + header: "Subscription", + cell: ({ getValue }) => ( + {getValue()} + ), + }, + { + accessorKey: "task_name", + header: "Task", + cell: ({ getValue }) => ( + + {getValue()} + + ), + }, + { + accessorKey: "queue", + header: "Queue", + cell: ({ getValue }) => ( + {getValue()} + ), + }, + { + accessorKey: "active", + header: "Status", + cell: ({ getValue }) => + getValue() ? ( + + Active + + ) : ( + + Paused + + ), + }, + { + accessorKey: "durable", + header: "Kind", + cell: ({ getValue }) => + getValue() ? ( + Durable + ) : ( + Ephemeral + ), + }, + { + accessorKey: "pending", + header: "Pending", + cell: ({ getValue }) => {formatCount(getValue())}, + }, + { + accessorKey: "running", + header: "Running", + cell: ({ getValue }) => ( + {formatCount(getValue())} + ), + }, + { + accessorKey: "dead", + header: "Dead", + cell: ({ getValue }) => { + const dead = getValue(); + return ( + 0 ? "text-danger" : "text-[var(--fg-muted)]"}`}> + {formatCount(dead)} + + ); + }, + }, + { + accessorKey: "oldest_pending_age_ms", + header: "Oldest pending", + cell: ({ getValue }) => ( + + {formatDuration(getValue())} + + ), + }, + ], + [], + ); + + if (error) { + return ( + + ); + } + + if (loading && rows.length === 0) { + return ( + + ); + } + + if (rows.length === 0) { + return ( + + ); + } + + return r.subscription} />; +} diff --git a/dashboard/src/features/topics/components/topics-table.tsx b/dashboard/src/features/topics/components/topics-table.tsx new file mode 100644 index 00000000..658c3363 --- /dev/null +++ b/dashboard/src/features/topics/components/topics-table.tsx @@ -0,0 +1,91 @@ +import { useNavigate } from "@tanstack/react-router"; +import type { ColumnDef } from "@tanstack/react-table"; +import { Radio } from "lucide-react"; +import { useMemo } from "react"; +import { DataTable, EmptyState, ErrorState, TableSkeleton } from "@/components/ui"; +import type { TopicSummary } from "@/lib/api-types"; +import { formatCount } from "@/lib/number"; + +interface TopicsTableProps { + topics: TopicSummary[] | undefined; + loading: boolean; + error: Error | null; + onRetry: () => void; +} + +const NUM_CELL = "block w-full text-right font-mono text-[0.82rem] tabular-nums"; + +export function TopicsTable({ topics, loading, error, onRetry }: TopicsTableProps) { + const navigate = useNavigate(); + + const rows = useMemo( + () => (topics ? [...topics].sort((a, b) => a.topic.localeCompare(b.topic)) : []), + [topics], + ); + + const columns = useMemo[]>( + () => [ + { + accessorKey: "topic", + header: "Topic", + cell: ({ getValue }) => ( + {getValue()} + ), + }, + { + accessorKey: "subscription_count", + header: "Subscriptions", + cell: ({ getValue }) => {formatCount(getValue())}, + }, + { + accessorKey: "backlog", + header: "Backlog", + cell: ({ getValue }) => ( + {formatCount(getValue())} + ), + }, + { + accessorKey: "dead", + header: "DLQ depth", + cell: ({ getValue }) => { + const dead = getValue(); + return ( + 0 ? "text-danger" : "text-[var(--fg-muted)]"}`}> + {formatCount(dead)} + + ); + }, + }, + ], + [], + ); + + if (error) { + return ( + + ); + } + + if (loading && rows.length === 0) { + return ; + } + + if (rows.length === 0) { + return ( + + ); + } + + return ( + r.topic} + onRowClick={(r) => navigate({ to: "/topics/$topic", params: { topic: r.topic } })} + /> + ); +} diff --git a/dashboard/src/features/topics/derived.test.ts b/dashboard/src/features/topics/derived.test.ts new file mode 100644 index 00000000..9fd9209c --- /dev/null +++ b/dashboard/src/features/topics/derived.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; +import type { TopicSummary } from "@/lib/api-types"; +import { topicTotals } from "./derived"; + +function topic(overrides: Partial): TopicSummary { + return { + topic: overrides.topic ?? "orders", + subscription_count: overrides.subscription_count ?? 1, + backlog: overrides.backlog ?? 0, + dead: overrides.dead ?? 0, + }; +} + +describe("topicTotals", () => { + it("returns zeroes for undefined (pre-load) input", () => { + expect(topicTotals(undefined)).toEqual({ topics: 0, backlog: 0, dead: 0 }); + }); + + it("returns zeroes for an empty list", () => { + expect(topicTotals([])).toEqual({ topics: 0, backlog: 0, dead: 0 }); + }); + + it("sums backlog and dead across topics and counts them", () => { + const totals = topicTotals([ + topic({ topic: "orders", backlog: 4, dead: 1 }), + topic({ topic: "audit", backlog: 1, dead: 0 }), + topic({ topic: "emails", backlog: 7, dead: 2 }), + ]); + expect(totals).toEqual({ topics: 3, backlog: 12, dead: 3 }); + }); +}); diff --git a/dashboard/src/features/topics/derived.ts b/dashboard/src/features/topics/derived.ts new file mode 100644 index 00000000..d75dddfe --- /dev/null +++ b/dashboard/src/features/topics/derived.ts @@ -0,0 +1,24 @@ +import type { TopicSummary } from "@/lib/api-types"; + +export interface TopicTotals { + topics: number; + backlog: number; + dead: number; +} + +/** + * Roll a topic list up into the summary counters shown in the page's stat + * cards. Mirrors the server-side per-topic aggregation one level higher so + * the header totals stay consistent with the table rows below them. + */ +export function topicTotals(topics: TopicSummary[] | undefined): TopicTotals { + if (!topics) return { topics: 0, backlog: 0, dead: 0 }; + return topics.reduce( + (acc, t) => ({ + topics: acc.topics + 1, + backlog: acc.backlog + t.backlog, + dead: acc.dead + t.dead, + }), + { topics: 0, backlog: 0, dead: 0 }, + ); +} diff --git a/dashboard/src/features/topics/hooks.ts b/dashboard/src/features/topics/hooks.ts new file mode 100644 index 00000000..5efa5d74 --- /dev/null +++ b/dashboard/src/features/topics/hooks.ts @@ -0,0 +1,33 @@ +import { queryOptions, useQuery } from "@tanstack/react-query"; +import { useRefreshInterval } from "@/providers"; +import { fetchTopicDetail, fetchTopics } from "./api"; + +const KEY = { + all: ["topics"] as const, + list: ["topics", "list"] as const, + detail: (topic: string) => ["topics", "detail", topic] as const, +}; + +export function topicsQuery() { + return queryOptions({ + queryKey: KEY.list, + queryFn: ({ signal }) => fetchTopics(signal), + }); +} + +export function topicDetailQuery(topic: string) { + return queryOptions({ + queryKey: KEY.detail(topic), + queryFn: ({ signal }) => fetchTopicDetail(topic, signal), + }); +} + +export function useTopics() { + const { intervalMs } = useRefreshInterval(); + return useQuery({ ...topicsQuery(), refetchInterval: intervalMs }); +} + +export function useTopicDetail(topic: string) { + const { intervalMs } = useRefreshInterval(); + return useQuery({ ...topicDetailQuery(topic), refetchInterval: intervalMs }); +} diff --git a/dashboard/src/features/topics/index.ts b/dashboard/src/features/topics/index.ts new file mode 100644 index 00000000..a2a8e387 --- /dev/null +++ b/dashboard/src/features/topics/index.ts @@ -0,0 +1,3 @@ +export * from "./components"; +export * from "./derived"; +export * from "./hooks"; diff --git a/dashboard/src/lib/api-types.ts b/dashboard/src/lib/api-types.ts index ea0104cc..b3978112 100644 --- a/dashboard/src/lib/api-types.ts +++ b/dashboard/src/lib/api-types.ts @@ -244,6 +244,29 @@ export interface WorkflowRun { created_at: number; } +export interface TopicSummary { + topic: string; + subscription_count: number; + /** Sum of pending + running across every subscription. */ + backlog: number; + /** Sum of dead-lettered deliveries across every subscription. */ + dead: number; +} + +export interface SubscriptionBacklog { + topic: string; + subscription: string; + task_name: string; + queue: string; + active: boolean; + durable: boolean; + pending: number; + running: number; + dead: number; + /** Age of the oldest pending delivery in ms; null when no pending backlog. */ + oldest_pending_age_ms: number | null; +} + export interface WorkflowNode { node_name: string; status: WorkflowNodeStatus; diff --git a/dashboard/src/routes/topics.tsx b/dashboard/src/routes/topics.tsx new file mode 100644 index 00000000..79680975 --- /dev/null +++ b/dashboard/src/routes/topics.tsx @@ -0,0 +1,51 @@ +import { createFileRoute } from "@tanstack/react-router"; +import { Layers, Radio, Skull } from "lucide-react"; +import { PageHeader } from "@/components/layout"; +import { StatCard } from "@/components/ui"; +import { TopicsTable, topicsQuery, topicTotals, useTopics } from "@/features/topics"; +import { formatCount } from "@/lib/number"; + +export const Route = createFileRoute("/topics")({ + loader: ({ context: { queryClient } }) => queryClient.ensureQueryData(topicsQuery()), + component: TopicsPage, +}); + +function TopicsPage() { + const topics = useTopics(); + const data = topics.data; + + const { topics: totalTopics, backlog: totalBacklog, dead: totalDead } = topicTotals(data); + + return ( +
+ +
+ } value={formatCount(totalTopics)} /> + } + value={formatCount(totalBacklog)} + hint="pending + running" + /> + } + value={formatCount(totalDead)} + hint="across subscribers" + /> +
+ topics.refetch()} + /> +
+ ); +} diff --git a/dashboard/src/routes/topics_.$topic.tsx b/dashboard/src/routes/topics_.$topic.tsx new file mode 100644 index 00000000..125d87a3 --- /dev/null +++ b/dashboard/src/routes/topics_.$topic.tsx @@ -0,0 +1,30 @@ +import { createFileRoute } from "@tanstack/react-router"; +import { PageHeader } from "@/components/layout"; +import { SubscriptionsTable, topicDetailQuery, useTopicDetail } from "@/features/topics"; + +export const Route = createFileRoute("/topics_/$topic")({ + loader: ({ context: { queryClient }, params: { topic } }) => + queryClient.ensureQueryData(topicDetailQuery(topic)), + component: TopicDetailPage, +}); + +function TopicDetailPage() { + const { topic } = Route.useParams(); + const detail = useTopicDetail(topic); + + return ( +
+ + detail.refetch()} + /> +
+ ); +} diff --git a/sdks/python/taskito/dashboard/handlers/topics.py b/sdks/python/taskito/dashboard/handlers/topics.py new file mode 100644 index 00000000..bcc8f270 --- /dev/null +++ b/sdks/python/taskito/dashboard/handlers/topics.py @@ -0,0 +1,47 @@ +"""Topic pub/sub route handlers.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from taskito.app import Queue + + +def _handle_topics(queue: Queue, qs: dict) -> list[dict[str, Any]]: + """Aggregate ``topic_stats()`` into one row per topic. + + Backlog folds pending + running across every subscription so operators + see a topic's total in-flight work at a glance; ``dead`` sums the DLQ + depth across subscribers. + """ + topics: dict[str, dict[str, Any]] = {} + for row in queue.topic_stats(): + entry = topics.setdefault( + row["topic"], + {"topic": row["topic"], "subscription_count": 0, "backlog": 0, "dead": 0}, + ) + entry["subscription_count"] += 1 + entry["backlog"] += row["pending"] + row["running"] + entry["dead"] += row["dead"] + return list(topics.values()) + + +def _handle_topic_detail(queue: Queue, qs: dict, topic: str) -> list[dict[str, Any]]: + """Per-subscription backlog rows for a single topic.""" + return queue.topic_stats(topic=topic) + + +def _handle_pause_subscription(queue: Queue, topic_and_name: tuple[str, str]) -> dict[str, bool]: + topic, name = topic_and_name + return {"paused": queue.pause_subscription(topic, name)} + + +def _handle_resume_subscription(queue: Queue, topic_and_name: tuple[str, str]) -> dict[str, bool]: + topic, name = topic_and_name + return {"active": queue.resume_subscription(topic, name)} + + +def _handle_unsubscribe(queue: Queue, topic_and_name: tuple[str, str]) -> dict[str, bool]: + topic, name = topic_and_name + return {"unsubscribed": queue.unsubscribe(topic, name)} diff --git a/sdks/python/taskito/dashboard/routes.py b/sdks/python/taskito/dashboard/routes.py index 84f06643..3a66459d 100644 --- a/sdks/python/taskito/dashboard/routes.py +++ b/sdks/python/taskito/dashboard/routes.py @@ -62,6 +62,13 @@ _handle_list_settings, _handle_set_setting, ) +from taskito.dashboard.handlers.topics import ( + _handle_pause_subscription, + _handle_resume_subscription, + _handle_topic_detail, + _handle_topics, + _handle_unsubscribe, +) from taskito.dashboard.handlers.webhook_deliveries import ( handle_get_delivery, handle_list_deliveries, @@ -144,6 +151,7 @@ def is_public_path(path: str) -> bool: "/api/queues": handle_list_queues, "/api/middleware": handle_list_middleware, "/api/workflows/runs": _handle_list_workflow_runs, + "/api/topics": _handle_topics, } # ── Parameterized GET routes: regex → handler(queue, qs, captured_id) ── @@ -169,6 +177,9 @@ def is_public_path(path: str) -> bool: (re.compile(r"^/api/workflows/runs/([^/]+)/dag$"), _handle_get_workflow_dag), (re.compile(r"^/api/workflows/runs/([^/]+)/children$"), _handle_get_workflow_children), (re.compile(r"^/api/workflows/runs/([^/]+)$"), _handle_get_workflow_run), + # Distinct ``/api/topics/`` prefix — neither shadows nor is shadowed by + # the patterns above. + (re.compile(r"^/api/topics/([^/]+)$"), _handle_topic_detail), ] # GET routes with 2 captured groups (handler signature: queue, qs, (g1, g2)) @@ -233,6 +244,14 @@ def is_public_path(path: str) -> bool: re.compile(r"^/api/webhooks/([^/]+)/deliveries/([^/]+)/replay$"), handle_replay_delivery, ), + ( + re.compile(r"^/api/topics/([^/]+)/subscriptions/([^/]+)/pause$"), + _handle_pause_subscription, + ), + ( + re.compile(r"^/api/topics/([^/]+)/subscriptions/([^/]+)/resume$"), + _handle_resume_subscription, + ), ] # ── Parameterized PUT routes: regex → handler(queue, body, captured_id) ── @@ -264,6 +283,14 @@ def is_public_path(path: str) -> bool: ), ] +# DELETE routes with 2 captured groups (handler signature: queue, (g1, g2)). +DELETE_PARAM2_ROUTES: list[tuple[re.Pattern, Any]] = [ + ( + re.compile(r"^/api/topics/([^/]+)/subscriptions/([^/]+)$"), + _handle_unsubscribe, + ), +] + def is_state_changing_method(method: str) -> bool: """POST/PUT/DELETE/PATCH all require a CSRF token.""" diff --git a/sdks/python/taskito/dashboard/server.py b/sdks/python/taskito/dashboard/server.py index 2519f7b9..aa3c69ce 100644 --- a/sdks/python/taskito/dashboard/server.py +++ b/sdks/python/taskito/dashboard/server.py @@ -46,6 +46,7 @@ from taskito.dashboard.routes import ( AUTH_CONTEXT_GET_PATHS, AUTH_CONTEXT_POST_PATHS, + DELETE_PARAM2_ROUTES, DELETE_PARAM_ROUTES, GET_CTX_ROUTES, GET_PARAM2_ROUTES, @@ -444,6 +445,15 @@ def _handle_delete(self) -> None: g1 = unquote(m.group(1)) self._dispatch_with_handler(param_handler, lambda h, g1=g1: h(queue, g1)) return + for pattern, param_handler in DELETE_PARAM2_ROUTES: + m = pattern.match(path) + if m: + g1, g2 = unquote(m.group(1)), unquote(m.group(2)) + self._dispatch_with_handler( + param_handler, + lambda h, g1=g1, g2=g2: h(queue, (g1, g2)), + ) + return self._json_response({"error": "Not found"}, status=404) # ── Auth gating ───────────────────────────────────────────── diff --git a/sdks/python/tests/dashboard/test_topics_api.py b/sdks/python/tests/dashboard/test_topics_api.py new file mode 100644 index 00000000..d2ed103f --- /dev/null +++ b/sdks/python/tests/dashboard/test_topics_api.py @@ -0,0 +1,150 @@ +"""HTTP route tests for ``/api/topics/*``. + +Boots a real dashboard server on a random port and drives it through the +standard AuthedClient + admin session pattern. The queue is seeded with a +few durable subscriptions and some published-but-unworked deliveries so the +aggregation and per-topic filtering can be asserted against known counts. +""" + +from __future__ import annotations + +import threading +from collections.abc import Generator +from pathlib import Path +from typing import Any + +import pytest + +from taskito import Queue +from taskito.dashboard._testing import AuthedClient, seed_admin_and_session + + +def _start_dashboard(queue: Queue) -> tuple[str, Any]: + from http.server import ThreadingHTTPServer + + from taskito.dashboard import _make_handler + + handler = _make_handler(queue, static_assets=None) + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return f"http://127.0.0.1:{port}", server + + +@pytest.fixture +def topics_dashboard(tmp_path: Path) -> Generator[tuple[AuthedClient, Queue]]: + db_path = str(tmp_path / "topics_dashboard.db") + q = Queue(db_path=db_path, workers=2) + + @q.subscriber("orders", name="fulfillment") + def fulfillment(order_id: int) -> None: + pass + + @q.subscriber("orders", name="analytics") + def analytics(order_id: int) -> None: + pass + + @q.subscriber("audit", name="audit_log") + def audit_log(event: str) -> None: + pass + + # Producer-only flush so publish() sees the durable subscriptions. + q.declare_subscriptions() + # Two orders publishes fan out to both order subscribers (4 jobs); one + # audit publish fans out to the single audit subscriber (1 job). No worker + # runs, so every delivery stays pending. + q.publish("orders", 1) + q.publish("orders", 2) + q.publish("audit", "login") + + url, server = _start_dashboard(q) + session = seed_admin_and_session(q) + client = AuthedClient(base=url, session=session) + try: + yield client, q + finally: + server.shutdown() + + +# ── List endpoint ───────────────────────────────────────────────────────── + + +def test_list_topics_aggregates_per_topic( + topics_dashboard: tuple[AuthedClient, Queue], +) -> None: + client, _ = topics_dashboard + data = client.get("/api/topics") + assert isinstance(data, list) + by_topic = {row["topic"]: row for row in data} + assert set(by_topic) == {"orders", "audit"} + + orders = by_topic["orders"] + assert orders["subscription_count"] == 2 + # 2 publishes x 2 subscribers = 4 pending, 0 running. + assert orders["backlog"] == 4 + assert orders["dead"] == 0 + + audit = by_topic["audit"] + assert audit["subscription_count"] == 1 + assert audit["backlog"] == 1 + assert audit["dead"] == 0 + + +# ── Detail endpoint ─────────────────────────────────────────────────────── + + +def test_topic_detail_filters_to_topic( + topics_dashboard: tuple[AuthedClient, Queue], +) -> None: + client, _ = topics_dashboard + rows = client.get("/api/topics/orders") + assert isinstance(rows, list) + assert len(rows) == 2 + assert {r["subscription"] for r in rows} == {"fulfillment", "analytics"} + for row in rows: + assert row["topic"] == "orders" + assert row["pending"] == 2 # one delivery per publish + assert row["running"] == 0 + assert row["active"] is True + assert row["durable"] is True + for key in ("task_name", "queue", "dead", "oldest_pending_age_ms"): + assert key in row + + +def test_topic_detail_unknown_topic_is_empty( + topics_dashboard: tuple[AuthedClient, Queue], +) -> None: + client, _ = topics_dashboard + assert client.get("/api/topics/does-not-exist") == [] + + +# ── Actions ─────────────────────────────────────────────────────────────── + + +def test_pause_and_resume_subscription( + topics_dashboard: tuple[AuthedClient, Queue], +) -> None: + client, _ = topics_dashboard + + paused = client.post("/api/topics/orders/subscriptions/fulfillment/pause") + assert paused == {"paused": True} + rows = {r["subscription"]: r for r in client.get("/api/topics/orders")} + assert rows["fulfillment"]["active"] is False + + resumed = client.post("/api/topics/orders/subscriptions/fulfillment/resume") + assert resumed == {"active": True} + rows = {r["subscription"]: r for r in client.get("/api/topics/orders")} + assert rows["fulfillment"]["active"] is True + + +def test_unsubscribe_removes_subscription( + topics_dashboard: tuple[AuthedClient, Queue], +) -> None: + client, _ = topics_dashboard + + result = client.delete("/api/topics/orders/subscriptions/analytics") + assert result == {"unsubscribed": True} + + rows = client.get("/api/topics/orders") + assert {r["subscription"] for r in rows} == {"fulfillment"}