--[[

	unreadable.gg

	Replaces the executor's `decompile` with the unreadable.gg decompiler. Every call sends the
	script's bytecode over a shared WebSocket and yields until its source comes back; a member's
	big batch spreads over a few of them

	shared.UNREADABLE_KEY = "ugk_..." -- optional: decompile as your account, not the free tier
	loadstring(game:HttpGet("https://unreadable.gg/decompile.lua"))()

]]--

-- read once, at load: set them before the loadstring, run it again to change them
local KEY = shared.UNREADABLE_KEY
local OPTIONS = shared.UNREADABLE_OPTIONS or { preset = "balanced" }
local URL = "wss://api.unreadable.gg/v1/stream"

local HttpService = game:GetService("HttpService")
local EncodingService = game:GetService("EncodingService")

-- these mean the client or its key is wrong: connecting again will not help
local FATAL = {
	["protocol"] = true,
	["invalid-key"] = true,
	["unsupported-protocol"] = true
}

-- one stream's server object carries every job's bytes and falls behind past a few hundred big
-- jobs: a member's batch opens another stream, one at a time, once each open one carries
-- PER_STREAM, up to STREAMS (the server allows 16 an identity). a stream with nothing to do for
-- IDLE seconds closes, all but the last, which the server closes after its own idle time
local PER_STREAM = 256
local STREAMS = 8
local IDLE = 10
-- after the server refuses a stream (its identity's are all in use), how long to keep to the
-- streams that were open then, once its retryAfter is over
local CAP_HOLD = 60

-- the open streams, each { socket, ready, everReady, dead, running = { [id] = job }, busy,
-- attempts, retryAfter, error, idle }
local streams = {}

-- what the latest `ready` allows across every stream, and whether that is the free tier's
local inFlight = 0
local free = true

-- streams to keep to after a refusal (nil: no cap known), whether a new one must wait, and which
-- refusal set them: only the latest one's timers clear them
local cap = nil
local streamHold = false
local refusals = 0

-- once set, every call returns it: the server refused us for good, or we can't reach it at all
local fatal = nil

-- a refusal past a minute (capacity or allowance used up until midnight), as { source }: while it
-- lasts every call gets it at once, with no socket opened to be refused again
local standing = nil

-- jobs waiting for a slot and jobs by bytecode; a stream holds the ones it sent
local queue = {}
local pending = {}
local nextId = 0

-- the same bytecode always decompiles to the same source, so it is never paid for twice
local cache = {}

-- waits in progress; a new `ready` or connection starts a new generation and ends them all
local holds, generation = 0, 0

local function base64(bytes)
	return buffer.tostring(EncodingService:Base64Encode(buffer.fromstring(bytes)))
end

local function failure(code, message)
	return `-- unreadable.gg could not decompile this script\n-- {code}: {message}\n`
end

-- read once, at load, so no call yields on a connect that can never start
local found, connectSocket = pcall(function()
	return WebSocket.connect
end)

if not found or type(connectSocket) ~= "function" then
	fatal = failure("unsupported-executor", "This executor has no WebSocket.connect.")
end

-- hands a job its answer and wakes every thread waiting on it
local function finish(job, source, keep)
	pending[job.bytecode] = nil

	if keep then
		cache[job.bytecode] = source
	end

	for _, thread in job.threads do
		task.spawn(thread, source)
	end
end

local function failAll(source)
	for _, stream in streams do
		for _, job in stream.running do
			finish(job, source, false)
		end

		stream.running, stream.busy = {}, 0
	end

	for _, job in queue do
		finish(job, source, false)
	end

	queue = {}
end

-- a long refusal answers every waiting call now and every new one until its retryAfter is over;
-- jobs already sent keep their own answers
local function stand(source, seconds)
	local mine = { source = source }
	standing = mine

	task.delay(seconds, function()
		if standing == mine then
			standing = nil
		end
	end)

	for _, job in queue do
		finish(job, source, false)
	end

	queue = {}
end

-- jobs sent on every stream and not yet answered: never more than `ready` allows
local function flying()
	local count = 0

	for _, stream in streams do
		count += stream.busy
	end

	return count
end

local function drop(stream)
	stream.dead = true

	local index = table.find(streams, stream)

	if index then
		table.remove(streams, index)
	end
end

-- a stream that ends here, not by the server: it leaves the pool first, so its close is no news
local function shut(stream)
	local socket = stream.socket
	stream.socket, stream.ready = nil, false
	drop(stream)

	if socket then
		pcall(socket.Close, socket)
	end
end

-- a stream that has had nothing to do for IDLE seconds closes, unless it is the last
local function watchIdle(stream)
	stream.idle = (stream.idle or 0) + 1
	local mine = stream.idle

	task.delay(IDLE, function()
		if stream.idle == mine and stream.busy == 0 and not stream.dead and #streams > 1 then
			shut(stream)
		end
	end)
end

local connect, onClose

-- no stream can open now or soon: the free tier's one, the most there may be, or the cap the
-- server showed us. only then does a stream carry more than PER_STREAM
local function atMost()
	return free or #streams >= math.min(STREAMS, cap or STREAMS)
end

-- a new stream only when every open one has its `ready`, carries its load, and the identity has
-- streams to spare
local function canGrow()
	if atMost() or streamHold then
		return false
	end

	for _, stream in streams do
		if not stream.ready then
			return false
		end
	end

	return true
end

-- false when the socket is already gone (a close we haven't heard yet) or won't take this message
local function send(stream, job)
	-- a counter never repeats, so a late answer from an old socket can't be mistaken for a new job's
	nextId += 1
	job.id = tostring(nextId)
	job.parts = {}
	stream.running[job.id] = job
	stream.busy += 1
	stream.idle = (stream.idle or 0) + 1

	return pcall(stream.socket.Send, stream.socket, HttpService:JSONEncode({
		type = "job",
		id = job.id,
		options = if next(OPTIONS) then OPTIONS else nil,
		bytecode = base64(job.bytecode)
	}))
end

local function pump()
	-- the server asked us to wait, or refused us for good
	if holds > 0 or fatal then
		return
	end

	local count = flying()
	local broken = {}

	while #queue > 0 and count < inFlight do
		-- the fullest stream that still has room, so a light load keeps to one and the rest go idle
		local pick = nil

		for _, stream in streams do
			if stream.ready and stream.busy < PER_STREAM and (not pick or stream.busy > pick.busy) then
				pick = stream
			end
		end

		-- every stream carries its load: the rest wait for the next one, unless no more can open,
		-- when the emptiest takes them
		if not pick and atMost() then
			for _, stream in streams do
				if stream.ready and (not pick or stream.busy < pick.busy) then
					pick = stream
				end
			end
		end

		if not pick then
			break
		end

		local job = table.remove(queue, 1)
		local ok, err = send(pick, job)

		if ok then
			count += 1
		elseif not pcall(pick.socket.Send, pick.socket, '{"type":"ping"}') then
			-- the socket won't take a ping either: it closed under us. the stream goes once the loop
			-- is done, and its jobs go again, this one too, with no strike against it
			pick.ready = false
			table.insert(broken, pick)
		else
			-- the socket is open but won't take this message (a frame-size limit): the job leaves
			-- the stream and goes again, and the second time it gets its answer
			pick.running[job.id] = nil
			pick.busy -= 1

			if job.thrown then
				finish(job, failure("send-failed", tostring(err)), false)
			else
				job.thrown = true
				table.insert(queue, job)
			end

			if pick.busy == 0 then
				watchIdle(pick)
			end
		end
	end

	for _, stream in broken do
		local socket = stream.socket

		if socket then
			onClose(stream)
			pcall(socket.Close, socket)
		end
	end

	if #queue > 0 and flying() < inFlight and #streams > 0 and canGrow() then
		task.spawn(connect)
	end
end

local function waitFor(seconds)
	local mine = generation
	holds += 1

	task.delay(seconds, function()
		if mine == generation then
			holds -= 1
		end

		pump()
	end)
end

local function clearWaits()
	generation += 1
	holds = 0
end

local function onJobLine(stream, message)
	local job = stream.running[message.id]

	-- a large source arrives in pieces, in order, before its `done`
	if job and message.stage == "source" then
		table.insert(job.parts, message.text)
		return
	end

	-- progress lines (`decompiling`, `renaming`, `waking-namer`) need no answer
	if not job or (message.stage ~= "done" and message.stage ~= "error") then
		return
	end

	stream.running[message.id] = nil
	stream.busy -= 1

	if message.stage == "done" then
		finish(job, message.source or table.concat(job.parts), true)
	elseif message.retryAfter and message.retryAfter <= 60 then
		-- busy, rate-limited or unavailable for a moment: same job, a little later. a longer
		-- wait (capacity or allowance used up until midnight) is an answer, not a pause
		table.insert(queue, 1, job)
		waitFor(message.retryAfter)
	else
		local answer = failure(message.error.code, message.error.message)
		finish(job, answer, false)

		-- a long wait is the identity's, not this job's: the jobs after it would be refused alike
		if message.retryAfter then
			stand(answer, message.retryAfter)
		end
	end

	pump()

	if stream.busy == 0 then
		watchIdle(stream)
	end
end

local function anyReady(except)
	for _, stream in streams do
		if stream ~= except and stream.ready then
			return true
		end
	end

	return false
end

local function onMessage(stream, message)
	if message.type == "ping" then
		-- the server checks we are still here while jobs run
		pcall(stream.socket.Send, stream.socket, '{"type":"pong"}')
	elseif message.type == "ready" then
		inFlight, free = message.limits.inFlight, message.limits.perMinute ~= nil
		stream.ready, stream.everReady, stream.attempts = true, true, 0
		-- the limits changed (a membership, midnight): a long refusal may be over
		standing = nil

		-- the first stream's `ready` starts a new generation: waits from before it are over
		if not anyReady(stream) then
			clearWaits()
		end

		pump()

		-- a stream the others' backlog didn't need after all goes idle like any other
		if stream.busy == 0 then
			watchIdle(stream)
		end
	elseif message.type == "error" then
		-- the server closes the connection right after this
		stream.retryAfter, stream.error = message.retryAfter, message.error.code

		if FATAL[message.error.code] then
			fatal = failure(message.error.code, message.error.message)
		elseif message.retryAfter and message.retryAfter > 60 then
			-- a wait past a minute is an answer, not a pause, but only once no other stream is left
			-- to serve the calls (onClose)
			stream.refusal = failure(message.error.code, message.error.message)
		end
	elseif message.id then
		onJobLine(stream, message)
	end
end

local open

-- the server's retryAfter if it gave one, else 1, 2, 4 … 30 s with jitter; nothing once the
-- stream is gone, the client refused for good, or no work is left for it
local function openLater(stream)
	stream.attempts += 1

	local delay = stream.retryAfter or math.min(30, 2 ^ (stream.attempts - 1)) * (0.5 + math.random())
	stream.retryAfter = nil

	task.delay(delay, function()
		if stream.dead or fatal then
			return
		end

		if #queue == 0 then
			return drop(stream)
		end

		open(stream)
	end)
end

-- the identity's streams are all in use: keep to the ones open now for a while, and wait as asked
local function refused(stream)
	cap = math.max(1, #streams - 1)
	streamHold = true
	refusals += 1
	local mine = refusals

	-- an earlier refusal's timers leave a later one's hold and cap alone
	task.delay(stream.retryAfter or 30, function()
		if mine == refusals then
			streamHold = false
			pump()
		end
	end)

	task.delay(CAP_HOLD, function()
		if mine == refusals then
			cap = nil
			pump()
		end
	end)
end

function onClose(stream)
	stream.socket, stream.ready = nil, false

	-- jobs the socket never answered were not done: they go again, on this stream or another
	for _, job in stream.running do
		table.insert(queue, job)
	end

	stream.running, stream.busy = {}, 0

	if fatal then
		-- every other stream goes too: each call already has its answer
		for _, other in streams do
			if other.socket then
				pcall(other.socket.Close, other.socket)
			end

			other.socket, other.ready, other.dead = nil, false, true
		end

		failAll(fatal)
		streams = {}
		return
	end

	local refusal = stream.refusal
	stream.refusal = nil

	-- a long refusal with no other stream left: the calls have their answer, and so do the next
	if refusal and #streams <= 1 then
		drop(stream)
		return stand(refusal, stream.retryAfter)
	end

	-- with other streams serving, a long refusal only stops the pool growing for as long as it lasts
	if refusal or (not stream.everReady and stream.error == "too-many-at-once") then
		refused(stream)
	end

	-- with others still open, a closed stream goes: they take its jobs, and a new one opens only if
	-- the load asks for it. the last one connects again while there is work
	if #streams > 1 or #queue == 0 then
		drop(stream)
		pump()
		return
	end

	openLater(stream)
end

function open(stream)
	stream.error = nil

	local ok, newSocket = pcall(connectSocket, URL)

	if not ok then
		return openLater(stream)
	end

	stream.socket = newSocket

	newSocket.OnMessage:Connect(function(raw)
		-- a dead socket can still deliver a late message; it belongs to no job of ours now
		if stream.socket ~= newSocket then
			return
		end

		local decoded, message = pcall(HttpService.JSONDecode, HttpService, raw)

		-- a frame that isn't JSON may have been a job's answer: the stream goes as if closed, and
		-- its jobs go again
		if decoded and type(message) == "table" then
			onMessage(stream, message)
		else
			onClose(stream)
			pcall(newSocket.Close, newSocket)
		end
	end)

	newSocket.OnClose:Connect(function()
		if stream.socket == newSocket then
			onClose(stream)
		end
	end)

	-- the key goes here, never in the URL; a socket that can't take it is a drop
	if not pcall(newSocket.Send, newSocket, HttpService:JSONEncode({ type = "hello", protocol = 1, key = KEY })) then
		onClose(stream)
		pcall(newSocket.Close, newSocket)
	end
end

function connect()
	if fatal then
		return
	end

	local stream = { ready = false, everReady = false, dead = false, running = {}, busy = 0, attempts = 0 }
	table.insert(streams, stream)
	open(stream)
end

getgenv().decompile = function(target)
	if fatal then
		return fatal
	end

	-- executors pass the script itself; raw bytecode works too
	local ok, bytecode = true, target

	if typeof(target) == "Instance" then
		ok, bytecode = pcall(getscriptbytecode, target)
	end

	if not ok or type(bytecode) ~= "string" or bytecode == "" then
		return "-- unreadable.gg: this script has no bytecode\n"
	end

	if cache[bytecode] then
		return cache[bytecode]
	end

	-- the same script already on its way: wait for that answer
	local job = pending[bytecode]

	if not job and standing then
		return standing.source
	end

	if not job then
		job = { bytecode = bytecode, threads = {} }
		pending[bytecode] = job
		table.insert(queue, job)
	end

	table.insert(job.threads, coroutine.running())

	if #streams == 0 then
		task.spawn(connect)
	else
		task.spawn(pump)
	end

	-- done!
	return coroutine.yield()
end