Adding external POST messages to an SSE queue

[bitaccesscomau]

I'm attempting to use this repo to monitor status from an external API that POSTs regularly to a provided notification endpoint.

Basically a POST is provided to this external API with a notificationUrl parameter in the body. Any future updates from the external API are sent to this notificationURL until I want to terminate the connection (usually when seeing a {message: 'Success'} in the body from the external API). I have got the initial connection open to my SSE URL, and I seem to be receiving the messages to the endpoint from the external API, but can't seem to understand how to pass them forward to the client. Each update from the external API server is an individual POST and isn't "kept alive" or streaming if that makes sense. I looked at the code in https://github.com/tncrazvan/sveltekit-sse and there is a few sections of the library that it seems to be using that are a bit more complex/undocumented than is available in the readme of this repo. Are you able to point me in the right direction?

Thanks for contributing to the Svelte ecosystem with this library!

+server.ts

function delay(milliseconds: number) {
  return new Promise(function run(resolve) {
    setTimeout(resolve, milliseconds);
  });
}

export function POST({ request }) {
  return events({
    request,
    timeout: 5000,
    async start({ emit }) {
      // eslint-disable-next-line no-constant-condition
      while (true) {
        emit(
          'message',
          JSON.stringify(request.body)
        );
        await delay(1000);
      }
    },
    cancel(){
      console.log("Connection canceled.")
    }
  });
}

Viewer.svelte

<script lang="ts">

// POST request to external API here

let notificationUrl = 'xxx'
let externalAPI = 'api.test.com';
let notificationBody = {Url: `${notificationUrl}`};
let token = 'xxx'

		const response = await fetch(
			externalAPI,
			{
				method: 'POST',
				headers: {
					Accept: 'application/json',
					'Content-Type': 'application/json',
					Authorization: `Bearer ${token}`
				},
				body: JSON.stringify(notificationBody)
			}
		);

		if (response.status === 202) {
			// Accepted
			console.log('Accepted');
			start(notificationUrl);
		} else {
			// Failed
			console.log('Failed');
		}
	}

	let message;

	function start(url) {
		message = source(url, {
			beacon: 3000
		}).select('message');
	}

	function stop(url) {
		source(url).close();
		message = null;
	}

</script>

{#if !message}
	Updates will appear here.
{:else}
	{$message}
{/if}

[razshare]

From what I'm understanding, you want a third party to be able to send notifications to your client.

You need to cache your emitter and map it to something the third party can provide you with in the future, so you can retrieve it and emit the message to the original client.

Try this https://github.com/tncrazvan/sveltekit-sse-issue-34

You have 3 endpoints.

• /request-notification-broker will generate a key that both the client and the third party can use to read or write to the stream.
• /notifications/[key]/read the client will use this to read the notifications.
  The act of reading from the stream will also invalidate the key.
  Multiple clients should not be able to read from the same stream.
  Unless you actually want that to happen, ofc, in which case you need to add an extra layer of abstraction to make sure things are closed properly.
• /notifications/[key]/write the third party will use this to write notifications to your client

The design looks like this

Or even this

So the third party is instantly aware of the new key.

Note

This doesn't guarantee there's an actual open stream for that key yet.
You still need the client to actually open the stream, otherwise you'll be sending notifications into the void.

I'm saying "third party" a lot, but it could be any server.

---

Let me know if this solves your problem.

[bitaccesscomau]

Thank you so much for your detailed response, I'm quite new to web development in general but I understand this conceptually. I'll go off and digest the code and let you know.

[bitaccesscomau]

Thanks for your help on this, a bit of an update.

I've ported over your example code into my application to fit the use case. It seems to be working perfectly fine locally, taking test messages in from POSTs, etc.

This may be a bit outside of scope, but when deploying to Vercel, the standard timed test message behaviour works well until the events({...,timeout: 30000,}) expires. I've added a beacon on the source function which resets this timer as outlined in the readme, but this only works locally.

On Vercel, it will also not accept POSTs both with server-less functions or edge-functions due to a 401 error. I assume it's because when the backend adds the key to the Set, it is stored in memory on that server instance. Following POST requests may be handled by different servers, that do not have access to the same memory and so are accessing a blank Set. Let me know if I'm on the right track here.

What would be the best approach here to get this working? Is this also why the beacon updates are not refreshing the timeout? I'd like to avoid having to setup infrastructure just to track sessions, but it seems like with my limited experience, this might be the only way?

[razshare]

Hey @bitaccesscomau ,
I'm afraid you are correct.

Unfortunately real time applications don't work exactly well with edge servers or serverless solutions in general.
You're technically not paying for the "server", you're paying for the service of exchanging requests between server and client, so you don't get to directly dictate how the server behaves.

My Azure experience

I don't have much experience with Vercel, so I can't speak on that matter, but I do have experience with other cloud providers like Azure and their web app scaling systems.

They actually do have a solution to this exact problem.
They use a cookie to allow the user to hit multiple times the same instance, and you can actually configure that, which is what we did in a previous job at work, because we had a similar requirement to yours.

https://techcommunity.microsoft.com/t5/apps-on-azure-blog/configure-arraffinity-cookie-when-accessing-azure-app-service/ba-p/3842511

Vercel

I did a bit of research quickly and I haven't been able to find an equivalent for Vercel, though I did stumble upon this https://vercel.com/docs/edge-network/headers#x-vercel-id-req

However it's not a cookie, it's a header

and from what the Vercel documentation says it's not deterministic, it doesn't identify a specific server instance, it identifies a group of instances located in the same zone.

And I'm not even sure if you can send this back and expect Vercel to take it into account.

If you really want to make this work I think it will be pretty difficult.

Final notes

Websockets could be a better solution for this use case... but even then - if the stream connection goes down for a moment, you can't reconnect to the same instance, you would lose the stream state, which is yet another headache to deal with.

Suffice to say, statefull server solutions (like websockets and sse), don't work well with stateless server architecture, like edge servers.

[razshare]

I'm going to close this because the issue in itself seems to be resolved.
Feel free to open more issues if you need to.

[Jan-Koll]

@razshare Your example has helped me a lot, thank you very much. I guess that this use case is generally quite common for SSE, so maybe you could reference this example in the README.md, too?

[razshare]

Hello @Jan-Koll ,
many apologies, I had read you message but then forgot about it.
It's ok to open new issues with these kind of questions, I'm not that picky about these kind of things.

To answer your question, yes, I will be adding a proper example in the readme, I'll try to get it done this weekend.

[razshare]

Hello again @Jan-Koll ,
as promised, the readme now includes an FAQ section with this topic https://github.com/razshare/sveltekit-sse?tab=readme-ov-file#faq

[Jan-Koll]

Awesome! Thanks again for your work!