feat: support Pusher/Ably/WSaaS for subscriptions

Describe the feature you'd like to request

The current implementation of subscriptions makes it hard to use in serverless environments as you need a dedicated websocket server to run for extended periods of time. I believe it would be nice to have an alternative means for subscriptions which can be routed via services like Pusher, which would support NextJS or other serverless environments better.

Describe the solution you'd like to see

A couple of flaws with the current implementation (wrt serverless):

The unsubscribe function is currently a callback returned by the subscribe function in observable. This means that the execution run must exist for the duration of the subscription, which is not particularly ideal for long-running subscriptions in serverless environments
There is no real API around submitting events from the server to the subscription, and instead it exists simply as a custom event emitter that the developer must implement themselves

Note that these are only flaws because serverless + WSaaS operates with different assumptions than normal subscriptions would.

A possible API for implementing the above ideas:

onAdd: t.procedure.subscription(({ ctx }) => {
    return channel<Post>(ctx.session.user.id);
  }),
add: t.procedure
    .input(
      z.object({
        id: z.string().uuid().optional(),
        text: z.string().min(1),
      }),
    )
    .mutation(async (opts) => {
      const post = { ...opts.input }; /* [..] add to db */
      await emitToChannel(opts.ctx.session.user.id, post);
      return post;
    }),

This API would be able to wrap services like Pusher as well as barebones websockets.

What this might look like under the hood is:

onAdd exposes 2 API endpoints, one for subscribing and one for unsubscribing. Each one handles their respective operations on the underlying channel implementation. For example, a subscribe call might create a Pusher channel called trpc:onAdd:1234 (where 1234 is the user ID passed into the function), and unsubscribe would delete said channel.
- If the channel already exists, the user could be added to that channel's members list, and vice versa while unsubscribing.
emitToChannel would route the data to the channel with the respective user id, serializing the message with the existing protocols.

This code is obviously quite rough and would need tweaking, but the resulting API might make subscriptions nicer to work with in a way thats a bit more agnostic to the backing implementation than the current design.

Describe alternate solutions

An alternative might be a new procedure type that mimics subscriptions but is designed around the new APIs, which would effectively be the above implementation again (or something similar) but maybe as t.procedure.channel instead. The obvious upside is backwards compatibility is much easier, but the downside is that it is conceptually quite similar to subscriptions.

Additional information

I wouldn't mind trying to write a PR for this, but it would take me quite some time to do because I am not familiar with the codebase or best practices of this repo, and I am also not sure what the best approaches for this particular problem would be.

👨‍👧‍👦 Contributing

[X] 🙋‍♂️ Yes, I'd be down to file a PR implementing this feature!

Funding

You can sponsor this specific effort via a Polar.sh pledge below
We receive the pledge once the issue is completed & verified

My current solution focuses on having a unified type for emitting and subscribing to events. Please check out this gist.

In router:

import newMessage from "./newMessage";

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 })); // <- the event payload type is enforced.
      return 'done';
    }),
  onAdd: publicProcedure
    .input(z.object({ chatId: z.string() }))
    .query(({ input }) => {
      return {
        currentMessages: ["test"],
        newMessage: newMessage(input.chatId),
      };
    }),
} satisfies TRPCRouterRecord;

In frontend app

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

I'm still looking for a more elegant approach. Let me know what you think; I might turn this into a package with proper documentation.

trpc / trpc