Merge pull request #98 from techdiary-dev/copilot/add-realtime-ui-updates

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

Author: kingRayhan

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/(home)/_components/ArticleFeed.tsx

@@ -85,7 +85,7 @@ const ArticleFeed = () => {
                   !article?.cover_image
                     ? extractImageUrlsFromMarkdown(article?.body ?? "")?.splice(
                         0,
-                        4
+                        4,
                       )
                     : []
                 }

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

@@ -0,0 +1,32 @@
+import { pusherServer } from "@/lib/pusher/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/article-cleanup-service.ts

@@ -1,5 +1,5 @@
 "use server";
-import { and, lt, lte, neq } from "sqlkit";
+import { and, lte, neq } from "sqlkit";
 import { persistenceRepository } from "../persistence/persistence-repositories";
 import { handleActionException } from "./RepositoryException";
 import { deleteArticleById } from "./search.service";

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 { publishMessage } from "@/lib/pusher/pusher.server";
 
 const sql = String.raw;
 
@@ -68,6 +69,12 @@ export const createMyComment = async (
       console.error("[inngest] Failed to send notification event:", err);
     });
 
+  void publishMessage(
+    `resource.${resource_type}.${resource_id}`,
+    "comment.created",
+    { scope: "comments" },
+  );
+
   return created?.rows?.[0];
 };
 
@@ -97,6 +104,12 @@ export const updateMyComment = async (
       data: { body: input.body, updated_at: new Date() },
     });
 
+    void publishMessage(
+      `resource.${existing.resource_type}.${existing.resource_id}`,
+      "comment.updated",
+      { scope: "comments" },
+    );
+
     return { success: true as const, data: { id: input.id } };
   } catch (error) {
     return handleActionException(error);
@@ -149,6 +162,12 @@ export const deleteMyComment = async (
       where: inArray("id", ids),
     });
 
+    void publishMessage(
+      `resource.${root.resource_type}.${root.resource_id}`,
+      "comment.deleted",
+      { scope: "comments" },
+    );
+
     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 { listenChannel } from "@/lib/pusher/pusher.client";
 
 const Context = React.createContext<
   { mutatingId?: string; setMutatingId: (id?: string) => void } | undefined
@@ -155,6 +156,22 @@ 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 channelName = `resource.${props.resource_type}.${props.resource_id}`;
+    const invalidate = () => {
+      queryClient.invalidateQueries({
+        queryKey: ["comments", props.resource_id, props.resource_type],
+      });
+    };
+    return listenChannel(channelName, {
+      "comment.created": invalidate,
+      "comment.updated": invalidate,
+      "comment.deleted": invalidate,
+    });
+  }, [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,40 @@
+"use client";
+
+import { listenChannel } from "@/lib/pusher/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 channelName = `private-user.${userId}`;
+    return listenChannel(channelName, "notification.new", () => {
+      queryClient.invalidateQueries({ queryKey: ["my-notifications"] });
+      queryClient.invalidateQueries({
+        queryKey: ["unread-notification-count"],
+      });
+    });
+  }, [userId, queryClient]);
+
+  return <>{children}</>;
+}

src/env.ts

@@ -23,10 +23,20 @@ export const env = createEnv({
     // Inngest
     INNGEST_EVENT_KEY: z.string().optional(),
     INNGEST_SIGNING_KEY: z.string().optional(),
+
+    // Pusher / Soketi (server-side)
+    PUSHER_WS_HOST: z.string().min(1),
+    PUSHER_APP_ID: z.string().min(1),
+    PUSHER_APP_KEY: z.string().min(1),
+    PUSHER_APP_SECRET: z.string().min(1),
   },
   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().min(1),
+    NEXT_PUBLIC_PUSHER_WS_HOST: z.string().min(1),
   },
   runtimeEnv: {
     NODE_ENV: process.env.NODE_ENV,
@@ -50,6 +60,14 @@ 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_WS_HOST: process.env.PUSHER_WS_HOST,
+
+    NEXT_PUBLIC_PUSHER_APP_KEY: process.env.NEXT_PUBLIC_PUSHER_APP_KEY,
+    NEXT_PUBLIC_PUSHER_WS_HOST: process.env.NEXT_PUBLIC_PUSHER_WS_HOST,
   },
   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 { publishMessage } from "@/lib/pusher/pusher.server";
 import { deleteExpiredArticles } from "@/backend/services/article-cleanup-service";
 
 const notificationPayloadSchema = z.object({
@@ -188,6 +189,12 @@ export const persistNotificationFn = inngest.createFunction(
       },
     ]);
 
+    // Broadcast a lightweight signal so the recipient's browser can invalidate
+    // its TanStack Query caches without polling.
+    await publishMessage(`private-user.${data.recipient_id}`, "notification.new", {
+      scope: "notifications",
+    });
+
     return { success: true };
   },
 );