Warp/src/Server/init.luau

--!optimize 2
--!strict
--@EternityDev
local Server = {}

local Thread = require("./Util/Thread")
local Buffer = require("./Util/Buffer")
local Identifier = require("./Util/Identifier")
local Replication = require("./Replication")
local Xor = require("./Util/Xor")

local Players = game:GetService("Players")
local RunService = game:GetService("RunService")
local Event: RemoteEvent = script.Parent:WaitForChild("Event") :: RemoteEvent
local _repl: RemoteEvent = script.Parent:WaitForChild("_repl") :: RemoteEvent
local UnreliableEvent: UnreliableRemoteEvent = script.Parent:WaitForChild("UnreliableEvent") :: UnreliableRemoteEvent
local deltaT: number, cycle: number = 0, 1 / 61
local writer: Buffer.Writer = Buffer.createWriter()

type Connection = {
	Connected: boolean,
	Disconnect: (self: Connection) -> (),
}
type Event = {
	i: number,
	c: (Player, ...any?) -> ...any?,
}

local queueEvent: {
	[Player]: { { any } },
} = {}
local queueUnreliableEvent: {
	[Player]: { { any } },
} = {}
local eventListeners: { [number]: { Event } } = {}
local eventSchemas: { [number]: Buffer.SchemaType } = {}
local players_ready: { Player }, player_bytes: { [Player]: number } = {}, {}

local pendingInvokes: { [string]: thread } = {}
local invokeId = 0

--@namespaces { string }
--@optional
-- Register namespaces to ensure all of the namespaces is being registered earlier on the server to prevent any unexpected issues on the client.
Server.reg_namespaces = function(namespaces: { string })
	for _, name: string in namespaces do
		Identifier.get_id(name)
	end
end

--@remoteName string
--@schema Buffer.SchemaType
-- Define a schema for strict data packing on a specific event.
Server.useSchema = function(remoteName: string, schema: Buffer.SchemaType)
	eventSchemas[Identifier.get_id(remoteName)] = schema
end

--@remoteName string
--@fn function
-- Connect to an event to receive incoming data from clients.
Server.Connect = function(remoteName: string, fn: (Player, ...any?) -> ...any?): Connection
	local detail = {
		i = Identifier.get_id(remoteName),
		c = fn,
	}
	if not eventListeners[detail.i] then
		eventListeners[detail.i] = {}
	end
	table.insert(eventListeners[detail.i], detail)
	return {
		Connected = true,
		Disconnect = function(self: Connection)
			if not self.Connected then
				return
			end
			self.Connected = false
			local bucket = eventListeners[detail.i]
			if bucket then
				local idx = table.find(bucket, detail)
				if idx then
					table.remove(bucket, idx)
				end
			end
		end,
	} :: Connection
end

--@remoteName string
--@fn function
-- Similar to :Connect but automatically disconnects after the first firing.
Server.Once = function(remoteName: string, fn: (...any?) -> ()): Connection
	local connection
	connection = Server.Connect(remoteName, function(...: any?)
		if connection then
			connection:Disconnect()
		end
		fn(...)
	end)
	return connection
end

--@remoteName string
-- Wait for an event to be triggered. Yields the current thread.
Server.Wait = function(remoteName: string): (number, ...any?)
	local thread, t = coroutine.running(), os.clock()
	Server.Once(remoteName, function(...: any?)
		task.spawn(thread, os.clock() - t, ...)
	end)
	return coroutine.yield()
end

--@remoteName string
-- Disconnect all connections for a specific event.
Server.DisconnectAll = function(remoteName: string)
	local id = Identifier.get_id(remoteName)
	if not id then
		return
	end
	eventListeners[id] = nil
end

--@remoteName string
-- Disconnect all connections and remove the event.
Server.Destroy = Server.DisconnectAll

--@remoteName string
--@reliable boolean
--@player Player
-- Fire an event to a specific player.
Server.Fire = function(remoteName: string, reliable: boolean, player: Player, ...: any?)
	local targetQueue = reliable and queueEvent or queueUnreliableEvent
	if not targetQueue[player] then
		targetQueue[player] = {} :: any
	end
	table.insert(targetQueue[player], {
		Identifier.get_id(remoteName),
		{ ... } :: any,
	})
end

--@remoteName string
--@reliable boolean
-- Fire an event to all connected players.
Server.Fires = function(remoteName: string, reliable: boolean, ...: any?)
	for _, player: Player in players_ready do
		Server.Fire(remoteName, reliable, player, ...)
	end
end

--@remoteName string
--@reliable boolean
--@except { Player }
-- Fire an event to all players except specified ones.
Server.FireExcept = function(remoteName: string, reliable: boolean, except: { Player }, ...: any?)
	for _, player: Player in players_ready do
		if table.find(except, player) then
			continue
		end
		Server.Fire(remoteName, reliable, player, ...)
	end
end

--@remoteName string
--@player Player
--@timeout number?
-- Invoke a client with timeout support. Yields the current thread. Returns nil if timeout occurs.
Server.Invoke = function(remoteName: string, player: Player, timeout: number?, ...: any?): ...any?
	invokeId += 1
	local reqId, thread = `{invokeId}`, coroutine.running()

	pendingInvokes[reqId] = thread
	task.delay(timeout or 2, function()
		local pending = pendingInvokes[reqId]
		if not pending then
			return
		end
		task.spawn(pending, nil)
		pendingInvokes[reqId] = nil
	end)
	if not queueEvent[player] then
		queueEvent[player] = {} :: any
	end
	table.insert(queueEvent[player], {
		0,
		{ reqId :: any, { ... } :: any } :: any,
	})
	return coroutine.yield()
end

if RunService:IsServer() then
	local function processIncoming(player: Player, b: buffer, ref: { Instance }?, handleInvokes: boolean)
		if type(b) ~= "buffer" then
			return
		end
		if not RunService:IsStudio() then
			local bytes: number = (player_bytes[player] or 0) + math.max(buffer.len(b), 800)
			if bytes > 8e3 then
				return
			end
			player_bytes[player] = bytes
		end

		local decoded = handleInvokes and Xor.decodeServer(player, b) or b
		local contents = Buffer.readEvents(decoded, ref, eventSchemas)
		for _, content in contents do
			local remote = content[1]
			local d = content[2]
			if handleInvokes then
				if remote == 1 then
					local id = d[1]
					local results = d[2]
					local pending = pendingInvokes[id]
					if pending then
						task.spawn(pending :: any, table.unpack(results))
						pendingInvokes[id] = nil
					end
					continue
				end
				if remote == 0 then
					if not next(eventListeners) then
						continue
					end
					local remoteName = d[1]
					local id = d[2]
					local args = d[3]
					local connections = eventListeners[remoteName]
					if connections and #connections > 0 then
						Thread(function()
							local results = { connections[1].c(player, table.unpack(args)) }
							if not queueEvent[player] then
								queueEvent[player] = {} :: any
							end
							table.insert(queueEvent[player], {
								0,
								{ id, results } :: any,
							})
						end)
					end
					continue
				end
			end
			local connections = eventListeners[remote]
			if connections then
				for _, connection in connections do
					Thread(connection.c :: any, player, table.unpack(d))
				end
			end
		end
	end

	Event.OnServerEvent:Connect(function(player: Player, b: buffer, ref: { Instance }?)
		processIncoming(player, b, ref, true)
	end)

	UnreliableEvent.OnServerEvent:Connect(function(player: Player, b: buffer, ref: { Instance }?)
		processIncoming(player, b, ref, false)
	end)

	RunService.PostSimulation:Connect(function(d: number)
		deltaT += d
		if deltaT < cycle then
			return
		end
		deltaT = 0

		-- reliable
		for player: Player, content in queueEvent do
			if #content == 0 or player.Parent ~= Players then
				continue
			end
			Buffer.writeEvents(writer, content, eventSchemas)
			do
				local buf, ref = Buffer.buildWithRefs(writer)
				local encoded = Xor.encodeServer(player, buf)
				Buffer.reset(writer)
				if not ref or #ref == 0 then
					Event:FireClient(player, encoded)
				else
					Event:FireClient(player, encoded, ref)
				end
			end
			player_bytes[player] = 0
			table.clear(queueEvent[player])
		end
		-- unreliable
		for player: Player, content in queueUnreliableEvent do
			if #content == 0 or player.Parent ~= Players then
				continue
			end
			Buffer.writeEvents(writer, content, eventSchemas)
			do
				local buf, ref = Buffer.buildWithRefs(writer)
				Buffer.reset(writer)
				if not ref or #ref == 0 then
					UnreliableEvent:FireClient(player, buf)
				else
					UnreliableEvent:FireClient(player, buf, ref)
				end
			end
			player_bytes[player] = 0
			table.clear(queueUnreliableEvent[player])
		end
	end)

	local function onAdded(player: Player)
		if not table.find(players_ready, player) then
			table.insert(players_ready, player)
		end
		if not queueEvent[player] then
			queueEvent[player] = {} :: any
		end
		if not queueUnreliableEvent[player] then
			queueUnreliableEvent[player] = {} :: any
		end
	end
	Players.PlayerAdded:Connect(onAdded)
	Players.PlayerRemoving:Connect(function(player: Player)
		Replication.remove(player)
		table.remove(players_ready, table.find(players_ready, player))
		if queueEvent[player] then
			table.clear(queueEvent[player])
			queueEvent[player] = nil
		end
		if player_bytes[player] then
			player_bytes[player] = nil
		end
		if queueUnreliableEvent[player] then
			table.clear(queueUnreliableEvent[player])
			queueUnreliableEvent[player] = nil
		end
		Xor.remove(player)
	end)
	for _, player: Player in ipairs(Players:GetPlayers()) do
		onAdded(player)
	end
end

--[[
 @class Server
 @schema
 define a schema for your data and use a strict packing
]]
Server.Schema = Buffer.Schema

return Server :: typeof(Server)