feat: realtime UI updates via Pusher/Soketi + TanStack Query invalidation (Phase 1 & 2)

Agent-Logs-Url: https://github.com/techdiary-dev/techdiary.dev/sessions/b073d05d-7457-4406-b59a-cafb8dac4c4d

Co-authored-by: shoaibsharif <29075110+shoaibsharif@users.noreply.github.com>

Author: Copilot

package.json

@@ -60,6 +60,8 @@
     "modern-screenshot": "^4.6.8",
     "next": "^16.2.1",
     "pg": "^8.14.1",
+    "pusher": "^5.3.3",
+    "pusher-js": "^8.5.0",
     "react": "^19",
     "react-advanced-cropper": "^0.20.1",
     "react-dom": "^19",

src/app/api/pusher/auth/route.ts

@@ -0,0 +1,32 @@
+import { pusherServer } from "@/lib/pusher.server";
+import { authID } from "@/backend/services/session.actions";
+import { NextRequest, NextResponse } from "next/server";
+
+/**
+ * Pusher private-channel auth endpoint.
+ * Pusher-js POSTs `socket_id` and `channel_name` as application/x-www-form-urlencoded.
+ * We verify the user's session and only sign subscriptions to the caller's own channel.
+ */
+export async function POST(req: NextRequest) {
+  if (!pusherServer) {
+    return new NextResponse("Realtime not configured", { status: 503 });
+  }
+
+  const userId = await authID();
+  if (!userId) {
+    return new NextResponse("Unauthorized", { status: 401 });
+  }
+
+  const body = await req.text();
+  const params = new URLSearchParams(body);
+  const socketId = params.get("socket_id") ?? "";
+  const channel = params.get("channel_name") ?? "";
+
+  // Only allow subscribing to the caller's own user channel
+  if (channel !== `private-user.${userId}`) {
+    return new NextResponse("Forbidden", { status: 403 });
+  }
+
+  const authResponse = pusherServer.authorizeChannel(socketId, channel);
+  return NextResponse.json(authResponse);
+}

src/backend/services/comment.action.ts

@@ -10,6 +10,7 @@ import { and, eq, inArray } from "sqlkit";
 import { CommentPresentation } from "../models/domain-models";
 import { inngest } from "@/lib/inngest";
 import { assertCommentResourceExists } from "./notifications.payload";
+import { pusherServer } from "@/lib/pusher.server";
 
 const sql = String.raw;
 
@@ -68,6 +69,14 @@ export const createMyComment = async (
       console.error("[inngest] Failed to send notification event:", err);
     });
 
+  pusherServer
+    ?.trigger(
+      `resource.${resource_type}.${resource_id}`,
+      "comment.created",
+      { scope: "comments" },
+    )
+    .catch(() => {});
+
   return created?.rows?.[0];
 };
 
@@ -97,6 +106,14 @@ export const updateMyComment = async (
       data: { body: input.body, updated_at: new Date() },
     });
 
+    pusherServer
+      ?.trigger(
+        `resource.${existing.resource_type}.${existing.resource_id}`,
+        "comment.updated",
+        { scope: "comments" },
+      )
+      .catch(() => {});
+
     return { success: true as const, data: { id: input.id } };
   } catch (error) {
     return handleActionException(error);
@@ -149,6 +166,14 @@ export const deleteMyComment = async (
       where: inArray("id", ids),
     });
 
+    pusherServer
+      ?.trigger(
+        `resource.${root.resource_type}.${root.resource_id}`,
+        "comment.deleted",
+        { scope: "comments" },
+      )
+      .catch(() => {});
+
     return { success: true as const, data: { id: input.id } };
   } catch (error) {
     return handleActionException(error);

src/components/comment-section.tsx

@@ -18,7 +18,7 @@ import {
   Trash2,
 } from "lucide-react";
 import Link from "next/link";
-import React, { useMemo, useState } from "react";
+import React, { useMemo, useState, useEffect } from "react";
 import { useImmer } from "use-immer";
 import { useLoginPopup } from "./app-login-popup";
 import ResourceReaction from "./ResourceReaction";
@@ -38,6 +38,7 @@ import { Button } from "./ui/button";
 import { Skeleton } from "./ui/skeleton";
 import { Textarea } from "./ui/textarea";
 import getFileUrl from "@/utils/getFileUrl";
+import { getPusherClient } from "@/lib/pusher.client";
 
 const Context = React.createContext<
   { mutatingId?: string; setMutatingId: (id?: string) => void } | undefined
@@ -155,6 +156,28 @@ export const CommentSection = (props: {
     refetchOnReconnect: false,
   });
 
+  // Phase 2: subscribe to the resource's public Pusher channel and
+  // invalidate the comments query when any mutation arrives from another client.
+  useEffect(() => {
+    const pusher = getPusherClient();
+    if (!pusher) return;
+
+    const channelName = `resource.${props.resource_type}.${props.resource_id}`;
+    const channel = pusher.subscribe(channelName);
+    const invalidate = () => {
+      queryClient.invalidateQueries({
+        queryKey: ["comments", props.resource_id, props.resource_type],
+      });
+    };
+    const events = ["comment.created", "comment.updated", "comment.deleted"];
+    events.forEach((event) => channel.bind(event, invalidate));
+
+    return () => {
+      events.forEach((event) => channel.unbind(event, invalidate));
+      pusher.unsubscribe(channelName);
+    };
+  }, [props.resource_id, props.resource_type, queryClient]);
+
   const generated_comment_id = () => crypto.randomUUID();
   const listQueryKey = [
     "comments",

src/components/providers/CommonProviders.tsx

@@ -11,6 +11,7 @@ import { ThemeProvider } from "./theme-provider";
 import { AppConfirmProvider } from "../app-confirm";
 import { AppAlertProvider } from "../app-alert";
 import { AppLoginPopupProvider } from "../app-login-popup";
+import { RealtimeProvider } from "./RealtimeProvider";
 
 type Props = PropsWithChildren<{
   initialTheme?: ThemePreference;
@@ -34,7 +35,7 @@ const CommonProviders: React.FC<Props> = ({
                   initialTheme={initialTheme}
                   migrateThemeFromLocalStorage={migrateThemeFromLocalStorage}
                 >
-                  {children}
+                  <RealtimeProvider>{children}</RealtimeProvider>
                 </ThemeProvider>
               </AppLoginPopupProvider>
             </Suspense>

src/components/providers/RealtimeProvider.tsx

@@ -0,0 +1,48 @@
+"use client";
+
+import { getPusherClient } from "@/lib/pusher.client";
+import { useSession } from "@/store/session.atom";
+import { useQueryClient } from "@tanstack/react-query";
+import React, { PropsWithChildren, useEffect } from "react";
+
+/**
+ * Subscribes to the authenticated user's private Pusher channel
+ * (`private-user.{userId}`) and invalidates TanStack Query caches
+ * when realtime events arrive.
+ *
+ * Phase 1 — Notifications:
+ *   event `notification.new` → invalidate `my-notifications` and
+ *   `unread-notification-count`.
+ *
+ * This component is a no-op when Pusher is not configured
+ * (NEXT_PUBLIC_PUSHER_APP_KEY is absent) or when the user is not
+ * signed in.
+ */
+export function RealtimeProvider({ children }: PropsWithChildren) {
+  const session = useSession();
+  const queryClient = useQueryClient();
+
+  const userId = session?.user?.id;
+
+  useEffect(() => {
+    if (!userId) return;
+
+    const pusher = getPusherClient();
+    if (!pusher) return;
+
+    const channelName = `private-user.${userId}`;
+    const channel = pusher.subscribe(channelName);
+
+    channel.bind("notification.new", () => {
+      queryClient.invalidateQueries({ queryKey: ["my-notifications"] });
+      queryClient.invalidateQueries({ queryKey: ["unread-notification-count"] });
+    });
+
+    return () => {
+      channel.unbind_all();
+      pusher.unsubscribe(channelName);
+    };
+  }, [userId, queryClient]);
+
+  return <>{children}</>;
+}

src/env.ts

@@ -25,10 +25,26 @@ export const env = createEnv({
     // Inngest
     INNGEST_EVENT_KEY: z.string().optional(),
     INNGEST_SIGNING_KEY: z.string().optional(),
+
+    // Pusher / Soketi (server-side)
+    PUSHER_APP_ID: z.string().optional(),
+    PUSHER_APP_KEY: z.string().optional(),
+    PUSHER_APP_SECRET: z.string().optional(),
+    PUSHER_HOST: z.string().optional(),
+    PUSHER_PORT: z.string().optional(),
+    PUSHER_USE_TLS: z.string().optional(),
+    PUSHER_CLUSTER: z.string().optional(),
   },
   client: {
     NEXT_PUBLIC_MEILISEARCH_API_HOST: z.url(),
     NEXT_PUBLIC_MEILISEARCH_SEARCH_API_KEY: z.string(),
+
+    // Pusher / Soketi (client-side)
+    NEXT_PUBLIC_PUSHER_APP_KEY: z.string().optional(),
+    NEXT_PUBLIC_PUSHER_HOST: z.string().optional(),
+    NEXT_PUBLIC_PUSHER_PORT: z.string().optional(),
+    NEXT_PUBLIC_PUSHER_FORCE_TLS: z.string().optional(),
+    NEXT_PUBLIC_PUSHER_CLUSTER: z.string().optional(),
   },
   runtimeEnv: {
     NODE_ENV: process.env.NODE_ENV,
@@ -54,6 +70,20 @@ export const env = createEnv({
 
     INNGEST_EVENT_KEY: process.env.INNGEST_EVENT_KEY,
     INNGEST_SIGNING_KEY: process.env.INNGEST_SIGNING_KEY,
+
+    PUSHER_APP_ID: process.env.PUSHER_APP_ID,
+    PUSHER_APP_KEY: process.env.PUSHER_APP_KEY,
+    PUSHER_APP_SECRET: process.env.PUSHER_APP_SECRET,
+    PUSHER_HOST: process.env.PUSHER_HOST,
+    PUSHER_PORT: process.env.PUSHER_PORT,
+    PUSHER_USE_TLS: process.env.PUSHER_USE_TLS,
+    PUSHER_CLUSTER: process.env.PUSHER_CLUSTER,
+
+    NEXT_PUBLIC_PUSHER_APP_KEY: process.env.NEXT_PUBLIC_PUSHER_APP_KEY,
+    NEXT_PUBLIC_PUSHER_HOST: process.env.NEXT_PUBLIC_PUSHER_HOST,
+    NEXT_PUBLIC_PUSHER_PORT: process.env.NEXT_PUBLIC_PUSHER_PORT,
+    NEXT_PUBLIC_PUSHER_FORCE_TLS: process.env.NEXT_PUBLIC_PUSHER_FORCE_TLS,
+    NEXT_PUBLIC_PUSHER_CLUSTER: process.env.NEXT_PUBLIC_PUSHER_CLUSTER,
   },
   onValidationError(issues: readonly StandardSchemaV1.Issue[]) {
     console.error("❌ Invalid environment variables:", issues);

src/lib/inngest.ts

@@ -7,6 +7,7 @@ import {
 import { persistenceRepository } from "@/backend/persistence/persistence-repositories";
 import { ActionException } from "@/backend/services/RepositoryException";
 import { buildPersistableNotification } from "@/backend/services/notifications.payload";
+import { pusherServer } from "@/lib/pusher.server";
 
 const notificationPayloadSchema = z.object({
   article_id: z.string().optional(),
@@ -176,6 +177,18 @@ export const persistNotificationFn = inngest.createFunction(
       },
     ]);
 
+    // Broadcast a lightweight signal so the recipient's browser can invalidate
+    // its TanStack Query caches without polling.
+    if (pusherServer) {
+      await pusherServer
+        .trigger(`private-user.${data.recipient_id}`, "notification.new", {
+          scope: "notifications",
+        })
+        .catch(() => {
+          // Publishing is best-effort; never let a Pusher failure break notification delivery.
+        });
+    }
+
     return { success: true };
   },
 );

src/lib/pusher.client.ts

@@ -0,0 +1,38 @@
+import Pusher from "pusher-js";
+import { env } from "@/env";
+
+let _pusherClient: Pusher | null = null;
+
+/**
+ * Returns a shared Pusher/Soketi client instance.
+ * Returns null when the public app key is not configured.
+ */
+export function getPusherClient(): Pusher | null {
+  if (typeof window === "undefined") return null;
+
+  const key = env.NEXT_PUBLIC_PUSHER_APP_KEY;
+  if (!key) return null;
+
+  if (_pusherClient) return _pusherClient;
+
+  const forceTLS = env.NEXT_PUBLIC_PUSHER_FORCE_TLS !== "false";
+  const port = env.NEXT_PUBLIC_PUSHER_PORT
+    ? Number(env.NEXT_PUBLIC_PUSHER_PORT)
+    : undefined;
+
+  // pusher-js always requires cluster; use empty string as placeholder when
+  // connecting to a self-hosted Soketi/compatible broker via wsHost.
+  _pusherClient = new Pusher(key, {
+    cluster: env.NEXT_PUBLIC_PUSHER_CLUSTER ?? "mt1",
+    authEndpoint: "/api/pusher/auth",
+    ...(env.NEXT_PUBLIC_PUSHER_HOST && {
+      wsHost: env.NEXT_PUBLIC_PUSHER_HOST,
+      wsPort: port,
+      wssPort: port,
+      enabledTransports: ["ws", "wss"],
+    }),
+    forceTLS,
+  });
+
+  return _pusherClient;
+}

src/lib/pusher.server.ts

@@ -0,0 +1,29 @@
+import Pusher from "pusher";
+import { env } from "@/env";
+
+/**
+ * Lazy singleton for the server-side Pusher/Soketi client.
+ * Returns null when Pusher is not configured (no PUSHER_APP_ID etc.),
+ * so callers can gracefully skip publishing in non-realtime environments.
+ */
+function createPusherServer(): Pusher | null {
+  const { PUSHER_APP_ID, PUSHER_APP_KEY, PUSHER_APP_SECRET } = env;
+  if (!PUSHER_APP_ID || !PUSHER_APP_KEY || !PUSHER_APP_SECRET) {
+    return null;
+  }
+
+  const base = {
+    appId: PUSHER_APP_ID,
+    key: PUSHER_APP_KEY,
+    secret: PUSHER_APP_SECRET,
+    useTLS: env.PUSHER_USE_TLS !== "false",
+  };
+
+  if (env.PUSHER_HOST) {
+    return new Pusher({ ...base, host: env.PUSHER_HOST, port: env.PUSHER_PORT });
+  }
+
+  return new Pusher({ ...base, cluster: env.PUSHER_CLUSTER ?? "mt1" });
+}
+
+export const pusherServer = createPusherServer();