Best way to proxy Cloudflare DO sse reponse through orpc?

[marbemac]

First off, I've been loving orpc - awesome work!

I've got a cloudflare durable object that returns an SSE stream at /stream-events. I was hoping to wrap the do and proxy the stream back through orpc, so that I can leverage a) my existing orpc client and b) my existing orpc middleware hooks for auth, db connections, etc.

Basically the below (question in comments). Thank you in advance for any tips!

const watchEvents = base.threads.watchEvents.use(requireAuth)
  // some input shape
  .input(z.object({ threadId: ThreadId.validator }))

  // some output shape
  .output(eventIterator(z.object({ type: z.string(), content: z.string() })))

  // the orpc handler
  .handler(async function* ({
    context,
    input,
    errors,
    signal,
  }) {
    // get DO instance somehow...
    const threadInstance = getMyDurableObjectInstance();

    try {
      const eventSource = new EventSource('http://do/stream-events', { fetcher: threadInstance });

      // OR could just make a fetch request, and do something w regular response
      // const response = await threadInstance.fetch('http://do/stream-events', {
      //   signal,
      //   headers: { accept: 'text/event-stream' },
      // });

      // QUESTION: how best to leverage eventSource (or plain http response if that is the better option)... 
      // can we just return/proxy eventSource or response back, or do we need to actually consume and re-emit the SSE events
      // here? if so, what's best way?
    } finally {
      console.log('Aborting SSE stream');
      // controller.abort();
    }
  });

[unnoq]

oRPC doesn't have a built-in way to forward SSE, but you can implement something like this.
(This is just a rough idea - you'll likely need to handle more edge cases in a real-world scenario.)

const watchEvents = base.threads.watchEvents.use(requireAuth)
  .input(z.object({ threadId: ThreadId.validator }))
  .output(eventIterator(z.object({ type: z.string(), content: z.string() })))
  .handler(async function* ({ context, input, errors, signal }) {
    // get DO instance somehow...
    const threadInstance = getMyDurableObjectInstance()

    try {
      const eventSource = new EventSource('http://do/stream-events', { fetcher: threadInstance })

      const pending: any[] = []
      let pendingResolve: ((value: any) => void) | undefined

      eventSource.onmessage = function (event) {
        if (pendingResolve) {
          pendingResolve(JSON.parse(event.data))
        }
        else {
          pending.push(JSON.parse(event.data))
        }
      }

      while (true) {
        if (pending.length > 0) {
          yield pending.shift()
        }
        else {
          yield await new Promise((resolve) => {
            pendingResolve = resolve
          })
        }
      }
    }
    finally {
      console.log('Aborting SSE stream')
      // controller.abort();
    }
  })