From 2d1facf59dd0cdca074fa637518cbe3472b53736 Mon Sep 17 00:00:00 2001
From: ASTROWwwW <loulouyt113@gmail.com>
Date: Wed, 24 Jun 2026 21:38:16 +0200
Subject: [PATCH] feat(esx_lib): add a pub/sub multicast module

Adds xLib.pubsub for server-driven topic multicast: a producer resource
subscribes players to a topic server-side and publishes data that is pushed
only to the subscribed players. Clients listen with xLib.pubsub.on and never
subscribe or publish themselves.

The subscription registry is a single shared instance in the esx_lib server VM,
reached cross-resource through exports; the client side is a per-VM listener
over one net event. Subscriptions are cleaned up on playerDropped, so a reused
serverId never inherits them, and on the owning resource stopping.

Also adds xLib.triggerClientEvent, which packs the payload once when sending to
many players; publish uses it so a broadcast to N subscribers serialises once.
---
 [core]/esx_lib/imports/pubsub/client.lua      |  79 +++++++
 [core]/esx_lib/imports/pubsub/server.lua      |  47 ++++
 [core]/esx_lib/resource/pubsub/server.lua     | 207 ++++++++++++++++++
 .../resource/triggerClientEvent/server.lua    |  21 ++
 4 files changed, 354 insertions(+)
 create mode 100644 [core]/esx_lib/imports/pubsub/client.lua
 create mode 100644 [core]/esx_lib/imports/pubsub/server.lua
 create mode 100644 [core]/esx_lib/resource/pubsub/server.lua
 create mode 100644 [core]/esx_lib/resource/triggerClientEvent/server.lua

diff --git a/[core]/esx_lib/imports/pubsub/client.lua b/[core]/esx_lib/imports/pubsub/client.lua
new file mode 100644
index 00000000..7fd26b65
--- /dev/null
+++ b/[core]/esx_lib/imports/pubsub/client.lua
@@ -0,0 +1,79 @@
+-- Per-VM client listener for xLib pub/sub. Routes incoming topic data to the
+-- handlers registered with xLib.pubsub.on. on() only listens; the player must be
+-- subscribed server-side to actually receive anything.
+
+local PUBSUB_EVENT <const> = '__xLib_pubsub' -- must match resource/pubsub/server.lua
+
+-- topic -> array of handlers
+local handlers = {}
+
+local pubsub = {}
+
+---Register a handler for a topic. Multiple handlers per topic are allowed.
+---@param topic string
+---@param handler fun(data: any, topic: string)
+function pubsub.on(topic, handler)
+    if type(topic) ~= 'string' or type(handler) ~= 'function' then
+        return
+    end
+
+    local list = handlers[topic]
+    if not list then
+        list = {}
+        handlers[topic] = list
+    end
+    list[#list + 1] = handler
+end
+
+---Remove handlers for a topic. Without a handler argument, removes all of them.
+---@param topic string
+---@param handler? function
+function pubsub.off(topic, handler)
+    if not handler then
+        handlers[topic] = nil
+        return
+    end
+
+    local list = handlers[topic]
+    if not list then
+        return
+    end
+
+    for i = #list, 1, -1 do
+        if list[i] == handler then
+            table.remove(list, i)
+        end
+    end
+
+    if list[1] == nil then
+        handlers[topic] = nil
+    end
+end
+
+RegisterNetEvent(PUBSUB_EVENT, function(topic, data)
+    local list = handlers[topic]
+    if not list then
+        return
+    end
+
+    -- fast path: a topic almost always has a single handler, so skip the copy
+    local n = #list
+    if n == 1 then
+        list[1](data, topic)
+        return
+    end
+
+    -- snapshot so a handler that calls off() mid-dispatch cannot shift the list
+    -- and skip a sibling that has not run yet
+    local snapshot = {}
+    for i = 1, n do
+        snapshot[i] = list[i]
+    end
+
+    for i = 1, n do
+        snapshot[i](data, topic)
+    end
+end)
+
+xLib.pubsub = pubsub
+return pubsub
diff --git a/[core]/esx_lib/imports/pubsub/server.lua b/[core]/esx_lib/imports/pubsub/server.lua
new file mode 100644
index 00000000..585c21dd
--- /dev/null
+++ b/[core]/esx_lib/imports/pubsub/server.lua
@@ -0,0 +1,47 @@
+-- Per-VM wrapper exposing xLib.pubsub.* on the server; forwards to the shared
+-- registry (resource/pubsub/server.lua) through its exports.
+
+local pubsub = {}
+
+---@param src number serverId of the player to add
+---@param topic string
+---@return boolean added
+function pubsub.subscribe(src, topic)
+    return xLib.pubsub_subscribe(src, topic)
+end
+
+---@param src number
+---@param topic string
+---@return boolean removed
+function pubsub.unsubscribe(src, topic)
+    return xLib.pubsub_unsubscribe(src, topic)
+end
+
+---@param topic string
+---@param data any server-trusted data only
+---@return integer count number of players notified
+function pubsub.publish(topic, data)
+    return xLib.pubsub_publish(topic, data)
+end
+
+---@param topic string
+---@return number[] serverIds
+function pubsub.subscribers(topic)
+    return xLib.pubsub_subscribers(topic)
+end
+
+---@param src number
+---@param topic string
+---@return boolean
+function pubsub.isSubscribed(src, topic)
+    return xLib.pubsub_isSubscribed(src, topic)
+end
+
+---@param topic string
+---@return integer count
+function pubsub.clear(topic)
+    return xLib.pubsub_clear(topic)
+end
+
+xLib.pubsub = pubsub
+return pubsub
diff --git a/[core]/esx_lib/resource/pubsub/server.lua b/[core]/esx_lib/resource/pubsub/server.lua
new file mode 100644
index 00000000..65c725cd
--- /dev/null
+++ b/[core]/esx_lib/resource/pubsub/server.lua
@@ -0,0 +1,207 @@
+-- xLib pub/sub: a producer subscribes players to a topic server-side and
+-- publishes data pushed only to them. Clients only listen, never subscribe.
+-- The registry is shared here (esx_lib server VM). Namespace topics by resource
+-- (e.g. "esx_shops:247:stock") so two resources cannot clash on a bare name.
+
+---@diagnostic disable: duplicate-set-field
+
+local PUBSUB_EVENT <const> = '__xLib_pubsub' -- must match imports/pubsub/client.lua
+local TOPIC_PATTERN <const> = '^[%w_:%.%-]+$'
+local MAX_TOPIC_LEN <const> = 200
+local resourceName <const> = GetCurrentResourceName()
+
+-- topic -> set of subscribed serverIds: subs[topic][src] = true
+local subs = {}
+-- serverId -> set of topics: memberOf[src][topic] = true (O(degree) cleanup on drop)
+local memberOf = {}
+-- topic -> resource that created it, purged on that resource's onResourceStop
+local ownerOf = {}
+
+local function isValidTopic(topic)
+    return type(topic) == 'string'
+        and #topic > 0
+        and #topic <= MAX_TOPIC_LEN
+        and topic:match(TOPIC_PATTERN) ~= nil
+end
+
+local function isLivePlayer(src)
+    -- a connected player has a name; rejects 0, stale or recycled serverIds
+    return type(src) == 'number' and src > 0 and GetPlayerName(src) ~= nil
+end
+
+local function detach(src, topic)
+    local set = subs[topic]
+    if set then
+        set[src] = nil
+        if next(set) == nil then
+            subs[topic] = nil
+            ownerOf[topic] = nil
+        end
+    end
+
+    local topics = memberOf[src]
+    if topics then
+        topics[topic] = nil
+        if next(topics) == nil then
+            memberOf[src] = nil
+        end
+    end
+end
+
+---Subscribe a player to a topic. Called server-side by the producer resource.
+---@param src number serverId of the player to add
+---@param topic string
+---@return boolean added false if already subscribed or input is invalid
+function xLib.pubsub_subscribe(src, topic)
+    if not isLivePlayer(src) or not isValidTopic(topic) then
+        return false
+    end
+
+    local set = subs[topic]
+    if set and set[src] then
+        return false -- idempotent: already subscribed
+    end
+
+    if not set then
+        set = {}
+        subs[topic] = set
+        ownerOf[topic] = GetInvokingResource() or resourceName
+    end
+    set[src] = true
+
+    local topics = memberOf[src]
+    if not topics then
+        topics = {}
+        memberOf[src] = topics
+    end
+    topics[topic] = true
+
+    return true
+end
+
+---Unsubscribe a player from a topic.
+---@param src number
+---@param topic string
+---@return boolean removed false if it was not subscribed
+function xLib.pubsub_unsubscribe(src, topic)
+    local set = subs[topic]
+    if not set or not set[src] then
+        return false -- idempotent
+    end
+    detach(src, topic)
+    return true
+end
+
+---Push data to every player subscribed to a topic. Server-trusted data only.
+---@param topic string
+---@param data any
+---@return integer count number of players notified
+function xLib.pubsub_publish(topic, data)
+    -- no isValidTopic() here on purpose: an invalid topic never created a set,
+    -- so the lookup below already returns 0 for it without a regex on the hot path
+    local set = subs[topic]
+    if not set then
+        return 0
+    end
+
+    local targets, count = {}, 0
+    for src in pairs(set) do
+        count = count + 1
+        targets[count] = src
+    end
+
+    -- packs the payload once for the whole set instead of once per client
+    xLib.triggerClientEvent(PUBSUB_EVENT, targets, topic, data)
+    return count
+end
+
+---@param topic string
+---@return number[] serverIds a fresh copy, empty if the topic has no subscribers
+function xLib.pubsub_subscribers(topic)
+    local set = subs[topic]
+    if not set then
+        return {}
+    end
+
+    local list, i = {}, 0
+    for src in pairs(set) do
+        i = i + 1
+        list[i] = src
+    end
+    return list
+end
+
+---@param src number
+---@param topic string
+---@return boolean
+function xLib.pubsub_isSubscribed(src, topic)
+    local set = subs[topic]
+    return set ~= nil and set[src] == true
+end
+
+---Drop a whole topic and all of its subscribers.
+---@param topic string
+---@return integer count number of subscribers that were removed
+function xLib.pubsub_clear(topic)
+    local set = subs[topic]
+    if not set then
+        return 0
+    end
+
+    local count = 0
+    for src in pairs(set) do
+        local topics = memberOf[src]
+        if topics then
+            topics[topic] = nil
+            if next(topics) == nil then
+                memberOf[src] = nil
+            end
+        end
+        count = count + 1
+    end
+
+    subs[topic] = nil
+    ownerOf[topic] = nil
+    return count
+end
+
+-- Remove a leaving player from every topic. serverIds are reused by the server,
+-- so skipping this would leak a topic's data to whoever inherits the slot.
+AddEventHandler('playerDropped', function()
+    local src = source
+    local topics = memberOf[src]
+    if not topics then
+        return
+    end
+
+    for topic in pairs(topics) do
+        local set = subs[topic]
+        if set then
+            set[src] = nil
+            if next(set) == nil then
+                subs[topic] = nil
+                ownerOf[topic] = nil
+            end
+        end
+    end
+    memberOf[src] = nil
+end)
+
+-- Purge the topics a stopped producer owned, so a producer restart never leaves
+-- orphan subscriptions that keep receiving pushes.
+AddEventHandler('onResourceStop', function(stopped)
+    if stopped == resourceName then
+        return
+    end
+
+    local orphaned = {}
+    for topic, owner in pairs(ownerOf) do
+        if owner == stopped then
+            orphaned[#orphaned + 1] = topic
+        end
+    end
+
+    for i = 1, #orphaned do
+        xLib.pubsub_clear(orphaned[i])
+    end
+end)
diff --git a/[core]/esx_lib/resource/triggerClientEvent/server.lua b/[core]/esx_lib/resource/triggerClientEvent/server.lua
new file mode 100644
index 00000000..8f5d0412
--- /dev/null
+++ b/[core]/esx_lib/resource/triggerClientEvent/server.lua
@@ -0,0 +1,21 @@
+-- Triggers an event for one or more clients. For an array of players the payload
+-- is packed once instead of being re-serialised per client.
+
+---@diagnostic disable: duplicate-set-field
+
+local pack = msgpack.pack_args
+
+---@param eventName string
+---@param targets number | number[] a single serverId, or an array of them
+---@param ... any
+function xLib.triggerClientEvent(eventName, targets, ...)
+    if type(targets) == 'number' then
+        return TriggerClientEvent(eventName, targets, ...)
+    end
+
+    local payload = pack(...)
+    local length = #payload
+    for i = 1, #targets do
+        TriggerClientEventInternal(eventName, targets[i], payload, length)
+    end
+end