feat: add topic pub/sub dashboard view

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 },
     ],
   },
   {

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<TopicSummary[]> {
+  return api.get<TopicSummary[]>("/api/topics", { signal });
+}
+
+export function fetchTopicDetail(
+  topic: string,
+  signal?: AbortSignal,
+): Promise<SubscriptionBacklog[]> {
+  return api.get<SubscriptionBacklog[]>(`/api/topics/${encodeURIComponent(topic)}`, { signal });
+}

dashboard/src/features/topics/components/index.ts

@@ -0,0 +1,2 @@
+export { SubscriptionsTable } from "./subscriptions-table";
+export { TopicsTable } from "./topics-table";

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<SubscriptionBacklog[]>(
+    () =>
+      subscriptions
+        ? [...subscriptions].sort((a, b) => a.subscription.localeCompare(b.subscription))
+        : [],
+    [subscriptions],
+  );
+
+  const columns = useMemo<ColumnDef<SubscriptionBacklog>[]>(
+    () => [
+      {
+        accessorKey: "subscription",
+        header: "Subscription",
+        cell: ({ getValue }) => (
+          <span className="font-medium text-[var(--fg)]">{getValue<string>()}</span>
+        ),
+      },
+      {
+        accessorKey: "task_name",
+        header: "Task",
+        cell: ({ getValue }) => (
+          <span className="font-mono text-[0.8rem] text-[var(--fg-muted)]">
+            {getValue<string>()}
+          </span>
+        ),
+      },
+      {
+        accessorKey: "queue",
+        header: "Queue",
+        cell: ({ getValue }) => (
+          <span className="text-[var(--fg-muted)]">{getValue<string>()}</span>
+        ),
+      },
+      {
+        accessorKey: "active",
+        header: "Status",
+        cell: ({ getValue }) =>
+          getValue<boolean>() ? (
+            <Badge tone="success" dot>
+              Active
+            </Badge>
+          ) : (
+            <Badge tone="warning" dot>
+              Paused
+            </Badge>
+          ),
+      },
+      {
+        accessorKey: "durable",
+        header: "Kind",
+        cell: ({ getValue }) =>
+          getValue<boolean>() ? (
+            <Badge tone="neutral">Durable</Badge>
+          ) : (
+            <Badge tone="info">Ephemeral</Badge>
+          ),
+      },
+      {
+        accessorKey: "pending",
+        header: "Pending",
+        cell: ({ getValue }) => <span className={NUM_CELL}>{formatCount(getValue<number>())}</span>,
+      },
+      {
+        accessorKey: "running",
+        header: "Running",
+        cell: ({ getValue }) => (
+          <span className={`${NUM_CELL} text-info`}>{formatCount(getValue<number>())}</span>
+        ),
+      },
+      {
+        accessorKey: "dead",
+        header: "Dead",
+        cell: ({ getValue }) => {
+          const dead = getValue<number>();
+          return (
+            <span className={`${NUM_CELL} ${dead > 0 ? "text-danger" : "text-[var(--fg-muted)]"}`}>
+              {formatCount(dead)}
+            </span>
+          );
+        },
+      },
+      {
+        accessorKey: "oldest_pending_age_ms",
+        header: "Oldest pending",
+        cell: ({ getValue }) => (
+          <span className={`${NUM_CELL} text-[var(--fg-muted)]`}>
+            {formatDuration(getValue<number | null>())}
+          </span>
+        ),
+      },
+    ],
+    [],
+  );
+
+  if (error) {
+    return (
+      <ErrorState
+        title="Couldn't load subscriptions"
+        description={error.message}
+        onRetry={onRetry}
+      />
+    );
+  }
+
+  if (loading && rows.length === 0) {
+    return (
+      <TableSkeleton
+        rows={4}
+        columns={["w-32", "w-40", "w-20", "w-16", "w-16", "w-16", "w-16", "w-16", "w-20"]}
+      />
+    );
+  }
+
+  if (rows.length === 0) {
+    return (
+      <EmptyState
+        icon={Radio}
+        title="No subscriptions"
+        description="This topic has no registered subscriptions."
+      />
+    );
+  }
+
+  return <DataTable columns={columns} data={rows} rowKey={(r) => r.subscription} />;
+}

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<TopicSummary[]>(
+    () => (topics ? [...topics].sort((a, b) => a.topic.localeCompare(b.topic)) : []),
+    [topics],
+  );
+
+  const columns = useMemo<ColumnDef<TopicSummary>[]>(
+    () => [
+      {
+        accessorKey: "topic",
+        header: "Topic",
+        cell: ({ getValue }) => (
+          <span className="font-medium text-[var(--fg)]">{getValue<string>()}</span>
+        ),
+      },
+      {
+        accessorKey: "subscription_count",
+        header: "Subscriptions",
+        cell: ({ getValue }) => <span className={NUM_CELL}>{formatCount(getValue<number>())}</span>,
+      },
+      {
+        accessorKey: "backlog",
+        header: "Backlog",
+        cell: ({ getValue }) => (
+          <span className={`${NUM_CELL} text-info`}>{formatCount(getValue<number>())}</span>
+        ),
+      },
+      {
+        accessorKey: "dead",
+        header: "DLQ depth",
+        cell: ({ getValue }) => {
+          const dead = getValue<number>();
+          return (
+            <span className={`${NUM_CELL} ${dead > 0 ? "text-danger" : "text-[var(--fg-muted)]"}`}>
+              {formatCount(dead)}
+            </span>
+          );
+        },
+      },
+    ],
+    [],
+  );
+
+  if (error) {
+    return (
+      <ErrorState title="Couldn't load topics" description={error.message} onRetry={onRetry} />
+    );
+  }
+
+  if (loading && rows.length === 0) {
+    return <TableSkeleton rows={5} columns={["w-40", "w-24", "w-20", "w-20"]} />;
+  }
+
+  if (rows.length === 0) {
+    return (
+      <EmptyState
+        icon={Radio}
+        title="No topics"
+        description="Topics appear here once a subscriber is registered and a message is published."
+      />
+    );
+  }
+
+  return (
+    <DataTable
+      columns={columns}
+      data={rows}
+      rowKey={(r) => r.topic}
+      onRowClick={(r) => navigate({ to: "/topics/$topic", params: { topic: r.topic } })}
+    />
+  );
+}

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>): 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 });
+  });
+});

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<TopicTotals>(
+    (acc, t) => ({
+      topics: acc.topics + 1,
+      backlog: acc.backlog + t.backlog,
+      dead: acc.dead + t.dead,
+    }),
+    { topics: 0, backlog: 0, dead: 0 },
+  );
+}

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 });
+}

dashboard/src/features/topics/index.ts

@@ -0,0 +1,3 @@
+export * from "./components";
+export * from "./derived";
+export * from "./hooks";

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;