tRPC Channel

README.md

tRPC Channel

apps\nextjs\src\app\_components\chat.tsx

export function ChatPage() {
  const { data } = api.post.onAdd.useQuery({ chatId: "123" });
  useSubscribe(data?.newMessage, (newData) => console.log("received", newData.content));
  const addNew = api.post.addNew.useMutation();

  return (
    <div>
      <Button onClick={() => addNew.mutate({ chatId: "123", content: "new message" })}>
        Add
      </Button>
    </div>
  );
}

packages\api\src\router\chat.ts

interface NewMessageData {
  content: string;
}

const newMessage = (
  chatId: string,
  data: null | NewMessageData = null,
): MetaEvent<NewMessageData> => ({
  channelName: `chat_${chatId}_messages`,
  eventName: "new",
  data,
});

export const chatRouter = {
  addNew: publicProcedure
    .input(z.object({ chatId: z.string(), content: z.string() }))
    .mutation(async ({ ctx, input }) => {
      const { chatId, content } = input;
      await ctx.publish(newMessage(chatId, { content }));
      return 'done';
    }),
  onAdd: publicProcedure
    .input(z.object({ chatId: z.string() }))
    .query(({ input }) => {
      const { chatId } = input;
      return {
        currentMessages: ["test"],
        newMessage: newMessage(chatId),
      };
    }),
} satisfies TRPCRouterRecord;

realtime\react.ts

import { createContext, use, useState } from "react";
import type { MetaEvent } from "./types";

export interface Realtime {
  subscribe<T>(
    channel: string,
    event: string,
    listener: (message: T) => void,
  ): () => void;
}

export const RealtimeContext = createContext<Realtime | undefined>(undefined);

export function useRealtime() {
  const realtime = use(RealtimeContext);
  if (!realtime) throw new Error("no realtime");
  return realtime;
}

export function useSubscribe<C extends MetaEvent>(
  metaEvent: C | undefined,
  listener: (data: Exclude<C["data"], null>) => void,
): void {
  const realtime = useRealtime();
  const listenerRef = useRef(listener);
  listenerRef.current = listener;
  const channelName = metaEvent?.channelName,
    eventName = metaEvent?.eventName;
  useEffect(() => {
    if (!channelName || !eventName) return;
    return realtime.subscribe(channelName, eventName, (data: any) =>
      listenerRef.current(data),
    );
  }, [channelName, eventName, realtime]);
}

realtime\server.ts

import Pusher from "pusher";
import type { MetaEvent } from "./types";

const pusher = new Pusher(config);
async function publish(event: MetaEvent) {
  return pusher.trigger(event.channelName, event.eventName, event.data);
}

export const createTRPCContext = async (opts: {
  headers: Headers;
  session: Session | null;
}) => ({ session, db, publish });

realtime\type.ts

export interface MetaEvent<Data = unknown> {
  channelName: string;
  eventName: string;
  data: null | Data;
}