refactor: update Pusher integration and environment variable validation

- Replaced Pusher client and server imports with a unified publishMessage function for better abstraction.
- Updated environment variable validation to require non-empty values for Pusher-related keys.
- Removed deprecated Pusher client and server files to streamline the codebase.

Author: kingRayhan

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

@@ -1,4 +1,4 @@
-import { pusherServer } from "@/lib/pusher.server";
+import { pusherServer } from "@/lib/pusher/pusher.server";
 import { authID } from "@/backend/services/session.actions";
 import { NextRequest, NextResponse } from "next/server";
 

src/backend/services/comment.action.ts

@@ -10,7 +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";
+import { publishMessage } from "@/lib/pusher/pusher.server";
 
 const sql = String.raw;
 
@@ -69,13 +69,11 @@ 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(() => {});
+  void publishMessage(
+    `resource.${resource_type}.${resource_id}`,
+    "comment.created",
+    { scope: "comments" },
+  );
 
   return created?.rows?.[0];
 };
@@ -106,13 +104,11 @@ 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(() => {});
+    void publishMessage(
+      `resource.${existing.resource_type}.${existing.resource_id}`,
+      "comment.updated",
+      { scope: "comments" },
+    );
 
     return { success: true as const, data: { id: input.id } };
   } catch (error) {
@@ -166,13 +162,11 @@ export const deleteMyComment = async (
       where: inArray("id", ids),
     });
 
-    pusherServer
-      ?.trigger(
-        `resource.${root.resource_type}.${root.resource_id}`,
-        "comment.deleted",
-        { scope: "comments" },
-      )
-      .catch(() => {});
+    void publishMessage(
+      `resource.${root.resource_type}.${root.resource_id}`,
+      "comment.deleted",
+      { scope: "comments" },
+    );
 
     return { success: true as const, data: { id: input.id } };
   } catch (error) {

src/components/comment-section.tsx

@@ -38,7 +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";
+import { listenChannel } from "@/lib/pusher/pusher.client";
 
 const Context = React.createContext<
   { mutatingId?: string; setMutatingId: (id?: string) => void } | undefined
@@ -159,23 +159,17 @@ export const CommentSection = (props: {
   // 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);
-    };
+    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();

src/components/providers/RealtimeProvider.tsx

@@ -1,6 +1,6 @@
 "use client";
 
-import { getPusherClient } from "@/lib/pusher.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";
@@ -27,24 +27,13 @@ export function RealtimeProvider({ children }: PropsWithChildren) {
   useEffect(() => {
     if (!userId) return;
 
-    const pusher = getPusherClient();
-    console.log("pusher", pusher);
-    if (!pusher) return;
-
     const channelName = `private-user.${userId}`;
-    const channel = pusher.subscribe(channelName);
-
-    channel.bind("notification.new", () => {
+    return listenChannel(channelName, "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,24 +25,18 @@ export const env = createEnv({
     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(),
+    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().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(),
+    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,
@@ -70,16 +64,10 @@ export const env = createEnv({
     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,
+    PUSHER_WS_HOST: process.env.PUSHER_WS_HOST,
 
     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,
+    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,7 +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";
+import { publishMessage } from "@/lib/pusher/pusher.server";
 import { deleteExpiredArticles } from "@/backend/services/article-cleanup-service";
 
 const notificationPayloadSchema = z.object({
@@ -191,15 +191,9 @@ 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.
-        });
-    }
+    await publishMessage(`private-user.${data.recipient_id}`, "notification.new", {
+      scope: "notifications",
+    });
 
     return { success: true };
   },

src/lib/pusher.client.ts

@@ -0,0 +1,65 @@
+import Pusher from "pusher-js";
+import { env } from "@/env";
+
+let _pusherClient: Pusher | null = null;
+
+function getPusherClient(): Pusher | null {
+  if (typeof window === "undefined") return null;
+
+  if (_pusherClient) return _pusherClient;
+
+  const key = env.NEXT_PUBLIC_PUSHER_APP_KEY;
+  if (!key) return null;
+
+  // 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: "mt1",
+    authEndpoint: "/api/socket/auth",
+    wsHost: env.NEXT_PUBLIC_PUSHER_WS_HOST,
+    enabledTransports: ["ws", "wss"],
+  });
+
+  return _pusherClient;
+}
+
+type EventHandlers = Record<string, () => void>;
+
+export function listenChannel(
+  channel: string,
+  handlers: EventHandlers,
+): () => void;
+export function listenChannel(
+  channel: string,
+  event: string,
+  handler: () => void,
+): () => void;
+export function listenChannel(
+  channel: string,
+  eventOrHandlers: string | EventHandlers,
+  handler?: () => void,
+): () => void {
+  const handlers: EventHandlers =
+    typeof eventOrHandlers === "string" && handler !== undefined
+      ? { [eventOrHandlers]: handler }
+      : (eventOrHandlers as EventHandlers);
+
+  const pusher = getPusherClient();
+  if (!pusher) {
+    return () => {};
+  }
+
+  const ch = pusher.subscribe(channel);
+  for (const [event, fn] of Object.entries(handlers)) {
+    ch.bind(event, fn);
+  }
+
+  return () => {
+    for (const [event, fn] of Object.entries(handlers)) {
+      ch.unbind(event, fn);
+    }
+    pusher.unsubscribe(channel);
+  };
+}
+
+/** Server-side publish: `publishMessage` in `./pusher.server` (cannot run in the browser). */

src/lib/pusher.server.ts

@@ -0,0 +1,46 @@
+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;
+  }
+
+  return new Pusher({
+    host: env.PUSHER_WS_HOST,
+    appId: PUSHER_APP_ID,
+    key: PUSHER_APP_KEY,
+    secret: PUSHER_APP_SECRET,
+  });
+}
+
+export const pusherServer = createPusherServer();
+
+/**
+ * Best-effort publish. No-ops when Pusher is not configured; swallows errors
+ * so callers (e.g. Inngest) are not broken by broker failures.
+ */
+export async function publishMessage(
+  channel: string,
+  event: "comment.created" | "comment.updated" | "comment.deleted",
+  data: Record<string, unknown> = {},
+): Promise<void> {
+  console.log(`
+    [pusher] Publishing message to channel ${channel} with event ${event} and data ${JSON.stringify(data)}
+  `);
+
+  pusherServer
+    ?.trigger(channel, event, data)
+    .then((data) => {
+      console.log("[pusher] Published message successfully");
+    })
+    .catch((err) => {
+      console.error("[pusher] Failed to publish message:", JSON.stringify(err));
+    });
+}