How to Structure a Single SSE Endpoint for Multiple Users With Per-User State

[SOG-web]

I’m currently implementing an SSE (Server-Sent Events) workflow using the HTTP library, and I’m trying to design a pattern for handling many users subscribing to a single SSE endpoint, while still being able to treat each subscriber individually.

What I want to achieve

I want to expose one SSE endpoint, for example:

/tournaments/updates

But this single endpoint should be able to:

1. Accept multiple users simultaneously
2. Identify each user separately
3. Maintain per-user state or metadata (e.g., what tournament they are interested in)
4. Only send each user the events relevant to their context

Example use-case

Imagine a game platform where many users subscribe to the tournament updates SSE.
There can be hundreds of tournaments, but I don’t want to create an SSE endpoint per tournament.

Instead:

• All users connect to one SSE endpoint
• Each user sends (for example) a query parameter or header such as ?tournament_id=abc123
• The server receives the SSE connection
• The SSE handler stores some per-user info (user ID, tournament ID, permissions, etc.)
• The library should allow me to route or target updates to the correct subscriber(s)

My question

What is the recommended way to structure this with your SSE implementation?

• How do I keep per-connection state inside an SSE endpoint?
• How do I distinguish users and send each of them only the events they subscribed to?
• Is there a canonical pattern for fan-out to multiple users from a single endpoint using this library?

Any guidance or example code would help a lot.
Thanks so much

[SOG-web]

this is how am currently trying to use it, am thinking thread will be too heavy

const std = @import("std");

const httpz = @import("httpz");
const xev = @import("xev");

const RequestContext = @import("../../../../app/app.zig").RequestContext;
const responses = @import("../../../common/responses.zig");

var sse_manager: ?*SSEManager = null;

pub fn initSSE(allocator: std.mem.Allocator) !void {
    sse_manager = try SSEManager.init(allocator);
}

pub fn deinitSSE() void {
    sse_manager.deinit();
    sse_manager = null;
}

// Multi-producer single-consumer
fn handler(ctx: *RequestContext, req: *httpz.Request, res: *httpz.Response) !void {
    if (ctx.user == null) {
        return responses.sendError(res, 401, "Unauthorized");
    }

    const user = ctx.user.?;

    const topics = req.param("topics");
    if (topics == null) {
        return responses.sendError(res, 400, "Missing topics parameter");
    }
    var topics_iter = std.mem.splitSequence(u8, topics.?, ",");
    const topics_list = std.ArrayList([]const u8){};
    defer topics_list.deinit(ctx.allocator);
    while (topics_iter.next()) |topic| {
        try topics_list.append(ctx.allocator, topic);
    }

    const stream = try res.startEventStreamSync();

    const subscriber = Subscriber{
        .stream = stream,
        .user_id = user.id,
        .topics = topics_list,
    };

    try sse_manager.addSubscriber(stream.handle, subscriber);
}

const Subscriber = struct {
    stream: std.net.Stream,
    user_id: []const u8,
    topics: std.ArrayList([]const u8),
};

pub const SSEManager = struct {
    allocator: std.mem.Allocator,
    loop: xev.Loop,
    async_handle: xev.Async,
    async_completion: xev.Completion,
    event_queue: EventQueue,
    subscribers: std.AutoHashMap(std.posix.fd_t, Subscriber),
    topic_index: std.StringHashMap(std.ArrayList(std.posix.fd_t)),
    mutex: std.Thread.Mutex = .{},

    pub fn init(allocator: std.mem.Allocator) !*SSEManager {
        const self = try allocator.create(SSEManager);
        errdefer allocator.destroy(self);

        var loop = try xev.Loop.init(.{});
        errdefer loop.deinit();

        var async_handle = try xev.Async.init();
        errdefer async_handle.deinit();

        self.* = .{
            .allocator = allocator,
            .loop = loop,
            .async_handle = async_handle,
            .async_completion = undefined,
            .event_queue = EventQueue.init(allocator),
            .subscribers = std.AutoHashMap(std.posix.fd_t, Subscriber).init(allocator),
            .topic_index = std.StringHashMap(std.ArrayList(std.posix.fd_t)).init(allocator),
        };

        self.async_handle.wait(
            &self.loop, // loop pointer
            &self.async_completion, // completion pointer
            SSEManager, // The TYPE, not a value (comptime!)
            self, // Pointer to the SSEManager instance
            asyncCallback, // Your callback function name
        );

        return self;
    }

    pub fn deinit(self: *SSEManager) void {
        self.loop.deinit();
        self.async_handle.deinit();
        self.event_queue.deinit();
        self.subscribers.deinit();
        self.topic_index.deinit();
    }

    fn asyncCallback(
        userdata: ?*SSEManager, // Your SSEManager pointer
        loop: *xev.Loop,
        completion: *xev.Completion,
        result: xev.Async.WaitError!void,
    ) xev.CallbackAction {
        _ = loop;
        _ = completion;
        _ = result catch return .disarm; // Handle error

        const self = userdata orelse return .disarm;
        self.mutex.lock();
        defer self.mutex.unlock();

        const events = self.event_queue.popAll() catch |err| {
            std.log.err("Failed to pop events from queue: {s}", .{@errorName(err)});
            return .disarm;
        };

        for (events) |event| {
            const subscribers = self.topic_index.get(event.topic) orelse continue;
            for (subscribers.items) |fd| {
                const subscriber = self.subscribers.get(fd) orelse continue;
                const formatted = std.fmt.allocPrint(
                    self.allocator,
                    "event: {s}\ndata: {s}\n\n",
                    .{ event.event_type, event.data },
                ) catch continue;
                defer self.allocator.free(formatted);
                subscriber.stream.writeAll(formatted) catch continue;
            }
        }

        return .rearm; // Keep listening for more notifications
    }

    pub fn addSubscriber(self: *SSEManager, fd: std.posix.fd_t, subscriber: Subscriber) !void {
        self.mutex.lock();
        defer self.mutex.unlock();
        try self.subscribers.put(fd, subscriber);

        for (subscriber.topics.items) |topic| {
            const entry = try self.topic_index.getOrPut(topic);
            if (!entry.found_existing) {
                entry.value_ptr.* = std.ArrayList(std.posix.fd_t){};
            }
            try entry.value_ptr.append(self.allocator, fd);
        }
    }

    pub fn removeSubscriber(self: *SSEManager, fd: std.posix.fd_t) !void {
        self.mutex.lock();
        defer self.mutex.unlock();
        _ = self.subscribers.remove(fd);
    }

    pub fn publish(self: *SSEManager, topic: []const u8, event_type: []const u8, data: []const u8) !void {
        try self.event_queue.push(.{
            .topic = topic,
            .event_type = event_type,
            .data = data,
        });
        try self.async_handle.notify();
    }

    pub fn run(self: *SSEManager) void {
        self.loop.run(.until_done) catch {};
    }
};

/// Thread-safe event queue for SSE broadcasting
pub const EventQueue = struct {
    mutex: std.Thread.Mutex = .{},
    items: std.ArrayList(Event),
    allocator: std.mem.Allocator,

    pub fn init(allocator: std.mem.Allocator) EventQueue {
        return .{
            .items = std.ArrayList(Event){},
            .allocator = allocator,
        };
    }

    /// Push an event to the queue (thread-safe, can be called from any thread)
    pub fn push(self: *EventQueue, event: Event) !void {
        self.mutex.lock();
        defer self.mutex.unlock();
        try self.items.append(self.allocator, event);
    }

    /// Pop a single event from the queue
    pub fn pop(self: *EventQueue) ?Event {
        self.mutex.lock();
        defer self.mutex.unlock();
        if (self.items.items.len == 0) return null;
        return self.items.orderedRemove(0);
    }

    /// Pop all events and return as owned slice (caller frees)
    pub fn popAll(self: *EventQueue) ![]Event {
        self.mutex.lock();
        defer self.mutex.unlock();
        return try self.items.toOwnedSlice(self.allocator);
    }

    pub fn deinit(self: *EventQueue) void {
        self.items.deinit(self.allocator);
    }
};

[karlseguin]

That's not really an http.zig thing, but at a high level, this looks ok.

The only things that stood out to me was the coarseness of the locks (which does have the benefit of keeping it straightforward and deadlock-free, I think) and some concern over the fact that FDs can be re-used and cleanup code largely seems absent.

Quick win: move formatted outside of the inner loop.

Anything extra really starts to be application/context specific. If you're only using xev for the notification, a ResetEvent/Condition might be a simpler alternative. I've seen systems like this become bottlenecked on the publish side, some having a pool of writers that can take an event and write it could be nice, but that adds a lot of threading complexity (e.g. two workers could try to write to the same socket at the same time)...and how you solve that is even more application specific (like, maybe every subscriber is assigned to 1 specific writing thread..but that means every writer has to handle every event (but only for a % of subscribers) etc, etc, etc..).

There's no end to how complicated this can get, so I wouldn't worry too much about it until it becomes an issue.

[SOG-web]

thanks so much for the response, I have added some of your suggestions here is the updated version, but I also explored one other pathway,

fn asyncCallback(
        userdata: ?*SSEManager, // Your SSEManager pointer
        loop: *xev.Loop,
        completion: *xev.Completion,
        result: xev.Async.WaitError!void,
    ) xev.CallbackAction {
        _ = loop;
        _ = completion;
        _ = result catch return .disarm; // Handle error

        const self = userdata orelse return .disarm;

        // No need to lock here - EventQueue has its own internal mutex
        const events = self.event_queue.popAll() catch |err| {
            std.log.err("Failed to pop events from queue: {s}", .{@errorName(err)});
            return .disarm;
        };

        var dead_fds = std.ArrayList(std.posix.fd_t){};
        defer dead_fds.deinit(self.allocator);

        for (events) |event| {
            // Get handler for this event type, or use default formatter
            const maybe_handler = self.event_registry.getHandler(event.handler_name);

            // Format the message ONCE before iterating subscribers
            const formatted_message: ?[]const u8 = if (maybe_handler) |handler|
                handler.formatter(event, self.allocator)
            else if (self.event_registry.getDefaultFormatter()) |default_fmt|
                default_fmt(event, self.allocator)
            else
                null;

            if (formatted_message == null) continue;
            defer self.allocator.free(formatted_message.?);

            // Get the sender function (use default if not specified)
            const sender = if (maybe_handler) |handler| handler.sender else defaultSender;

            self.mutex.lock();
            const subscribers = self.topic_index.get(event.topic);

            if (subscribers) |subs| {
                for (subs.items) |fd| {
                    const subscriber = self.subscribers.get(fd) orelse continue;

                    // Send pre-formatted message to each subscriber
                    const success = sender(&subscriber, formatted_message.?);
                    //TODO: consider a retry mechanism in the future
                    if (!success) {
                        dead_fds.append(self.allocator, fd) catch {};
                    }
                }
            }

            for (dead_fds.items) |fd| {
                self.removeSubscriberInternal(fd);
            }
            dead_fds.clearRetainingCapacity();
            self.mutex.unlock();
        }

        // Free owned strings in each event, then free the events slice
        for (events) |event| {
            self.allocator.free(event.topic);
            self.allocator.free(event.event_type);
            self.allocator.free(event.data);
            self.allocator.free(event.handler_name);
        }
        self.allocator.free(events);

        return .rearm; // Keep listening for more notifications
    }

a cloning approach to reduce lock time

fn asyncCallback(
        userdata: ?*SSEManager, // Your SSEManager pointer
        loop: *xev.Loop,
        completion: *xev.Completion,
        result: xev.Async.WaitError!void,
    ) xev.CallbackAction {
        _ = loop;
        _ = completion;
        _ = result catch return .disarm; // Handle error

        const self = userdata orelse return .disarm;

        // No need to lock here - EventQueue has its own internal mutex
        const events = self.event_queue.popAll() catch |err| {
            std.log.err("Failed to pop events from queue: {s}", .{@errorName(err)});
            return .disarm;
        };

        var dead_fds = std.ArrayList(std.posix.fd_t){};
        defer dead_fds.deinit(self.allocator);

        for (events) |event| {
            // Get handler for this event type, or use default formatter
            const maybe_handler = self.event_registry.getHandler(event.handler_name);

            // Format the message ONCE before iterating subscribers
            const formatted_message: ?[]const u8 = if (maybe_handler) |handler|
                handler.formatter(event, self.allocator)
            else if (self.event_registry.getDefaultFormatter()) |default_fmt|
                default_fmt(event, self.allocator)
            else
                null;

            if (formatted_message == null) continue;
            defer self.allocator.free(formatted_message.?);

            // Get the sender function (use default if not specified)
            const sender = if (maybe_handler) |handler| handler.sender else defaultSender;

            // Clone subscriber list while holding lock briefly
            self.mutex.lock();
            const fd_list = self.topic_index.get(event.topic);
            const subscriber_fds = if (fd_list) |list|
                self.allocator.dupe(std.posix.fd_t, list.items) catch null
            else
                null;
            self.mutex.unlock();

            const fds = subscriber_fds orelse continue;
            defer self.allocator.free(fds);

            // Send OUTSIDE the lock - HTTP threads won't block during slow I/O
            for (fds) |fd| {
                // Brief lock to get subscriber data
                self.mutex.lock();
                const maybe_subscriber = self.subscribers.get(fd);
                self.mutex.unlock();

                const subscriber = maybe_subscriber orelse continue;

                // Send pre-formatted message (no lock held during network I/O)
                const success = sender(&subscriber, formatted_message.?);
                //TODO: consider a retry mechanism in the future
                if (!success) {
                    dead_fds.append(self.allocator, fd) catch {};
                }
            }

            // Lock briefly to clean up dead subscribers
            if (dead_fds.items.len > 0) {
                self.mutex.lock();
                for (dead_fds.items) |fd| {
                    self.removeSubscriberInternal(fd);
                }
                self.mutex.unlock();
            }
            dead_fds.clearRetainingCapacity();
        }

        // Free owned strings in each event, then free the events slice
        for (events) |event| {
            self.allocator.free(event.topic);
            self.allocator.free(event.event_type);
            self.allocator.free(event.data);
            self.allocator.free(event.handler_name);
        }
        self.allocator.free(events);

        return .rearm; // Keep listening for more notifications
    }

am also thinking of adding a thread pool writer

  const success = sender(&subscriber, formatted_message.?); // wraping this with a thread pool approach 

[karlseguin]

There's an endless list of possible implementation / tweaks / optimization. Since, from this snippet, I believe asyncCallback can't be called concurrently (it can't be called until it's re-armed again), I'd be tempted to use an ArenaAllocator:

const self = userdata orelse return .disarm;
const arena = self.arena.allocator();
defer self.arena.retain(.{.retain_with_limit = MAX_MEMORY_YOURE_WILL_TO_RETAIN_BETWEEN_CALLS});?

This could drastically reduce the amount of allocation/de-allocation you're doing.

[zigster64]

@SOG-web

https://github.com/zigster64/datastar.http.zig

This is a full stack lib that sits on top of http.zig using SSE and zig 0.15, and it has a built in pub/sub mechanism that does what you describe.

Feel free to pull out any code you might find useful for your stuff. In particular look at the pubsub engine

Of note :

• the endpoint that serves a long lived SSE stream uses res.disown() to detach the socket connection from the handler. So the handler can exit cleanly, freeing up a thread, whilst keeping the socket open
• on publish(topic), the pubsub engine maintains a list of open connections subscribed by topic, so it can quickly find all the connections to broadcast to
• you don’t need long-running handler processes for this, as the publish() mechanism invokes a new callback func for each subscriber, passing the open connection handle.

This approach can handle over 40k concurrent SSE streams on moderate hardware, all updating constantly.

There are some example apps to pick through

Example_22 has a complex setup where all users subscribe to updates on the 1 topic, but they also have session based preferences about how the broadcast data should be displayed … so the publish() call customises the output to match the session prefs. It’s a bit complicated, but useful to demonstrate a worse case scenario .

Hope that helps

[lalinsky]

I really don't want to spam this with yet another project link, but I think something like this would REALLY benefit from coroutines. The state management gets much easier.

https://github.com/lalinsky/dusty/blob/main/examples/sse.zig

[zigster64]

I really don't want to spam this with yet another project link, but I think something like this would REALLY benefit from coroutines. The state management gets much easier.

https://github.com/lalinsky/dusty/blob/main/examples/sse.zig

Totally agree with that - actor model approach is a really good fit for handling long running SSE streams. Starts to really shine as soon as the handler gets a little bit more complicated… wanting to select on a range of different events, timeouts, applying weird logic, etc

And if you are doing actor model, then coroutines are just the thing.

With the work I’m doing on zig 0.16, I’m using the actor model for the reasons above. Sitting on Io.Threaded for now because other than what you have done there with libxev, ain’t much other choice yet :)

Surprisingly, using threads isn’t as bad as I thought - you can tune the stack size to keep memory usage under control, and whilst waiting for the next msg, each thread is put to sleep awaiting a thread condition … so no real stress holding a few thousand threads open. Not ideal of course, but not terrible either. Still outperforms Go by 2-3x even, which wasn’t expected at all.

Gotta have a play with Dusty yet .. I will need to support integrating with that at some stage too for my other web app libs. Nice work.