GuildHelper = GuildHelper or {}

local addon = GuildHelper

addon.Transport = addon.Transport or {}
addon.Transport.incomingSnapshots = addon.Transport.incomingSnapshots or {}
addon.Transport.outboundQueue = addon.Transport.outboundQueue or {}
addon.Transport.maintenanceTicker = addon.Transport.maintenanceTicker or nil
addon.Transport.sendTicker = addon.Transport.sendTicker or nil
addon.Transport.inboundTaskQueue = addon.Transport.inboundTaskQueue or {}
addon.Transport.taskTicker = addon.Transport.taskTicker or nil
addon.Transport.pushSequence = addon.Transport.pushSequence or 0

local MAX_SYNC_MESSAGE_BYTES = 240
local SEND_INTERVAL_SECONDS = 0.05
local MAINTENANCE_INTERVAL_SECONDS = 5
local BUFFER_TIMEOUT_SECONDS = 180
local MAX_OUTBOUND_QUEUE = 12000
local INBOUND_PROCESS_INTERVAL_SECONDS = 0.05
local MAX_INBOUND_TASK_QUEUE = 64

local function getEncodedEnvelopeSize(envelope)
    return #(addon.SyncProtocol:Encode(envelope) or "")
end

local function encodedPayloadFits(op, sessionId, tableName, payload, seq, total)
    local envelopeOverhead = getEncodedEnvelopeSize({
        op = op,
        sessionId = sessionId,
        tableName = tableName,
        seq = seq or 0,
        total = total or 0,
        payload = "",
    })

    return envelopeOverhead + #(payload or "") <= MAX_SYNC_MESSAGE_BYTES
end

local function buildChunkPlan(op, sessionId, tableName, payload)
    local payloadLength = #(payload or "")
    local chunkSize = math.max(1, MAX_SYNC_MESSAGE_BYTES - getEncodedEnvelopeSize({
        op = op,
        sessionId = sessionId,
        tableName = tableName,
        seq = 1,
        total = 1,
        payload = "",
    }))

    while true do
        local total = math.max(1, math.ceil(payloadLength / chunkSize))
        local nextChunkSize = math.max(1, MAX_SYNC_MESSAGE_BYTES - getEncodedEnvelopeSize({
            op = op,
            sessionId = sessionId,
            tableName = tableName,
            seq = total,
            total = total,
            payload = "",
        }))

        if nextChunkSize == chunkSize then
            return chunkSize, total
        end

        chunkSize = nextChunkSize
    end
end

local function sendAddonMessage(prefix, payload, channel, peer)
    if C_ChatInfo and C_ChatInfo.SendAddonMessage then
        local ok, result = pcall(C_ChatInfo.SendAddonMessage, prefix, payload, channel, peer)

        if ok then
            return result
        end

        local parameterSets = {
            { prefix = prefix, message = payload, chatType = channel, target = peer },
            { prefix = prefix, text = payload, chatType = channel, target = peer },
            { prefix = prefix, message = payload, distribution = channel, target = peer },
            { prefix = prefix, text = payload, distribution = channel, target = peer },
        }

        for _, params in ipairs(parameterSets) do
            local tableOk, tableResult = pcall(C_ChatInfo.SendAddonMessage, params)

            if tableOk then
                return tableResult
            end
        end

        return nil
    end

    if SendAddonMessage then
        return SendAddonMessage(prefix, payload, channel, peer)
    end

    return nil
end

local function sendChunked(op, peer, tableName, sessionId, payload)
    local chunkSize, total = buildChunkPlan(op, sessionId, tableName, payload)

    if (#addon.Transport.outboundQueue + total) > MAX_OUTBOUND_QUEUE then
        return false, "outbound sync queue is full"
    end

    for index = 1, total do
        local chunk = payload:sub(((index - 1) * chunkSize) + 1, index * chunkSize)

        local ok, errorMessage = addon.Transport:Send(peer, "WHISPER", {
            op = op,
            sessionId = sessionId,
            tableName = tableName,
            seq = index,
            total = total,
            payload = chunk,
        })

        if not ok then
            return false, errorMessage
        end
    end

    return true
end

local function nextPushSessionId(peer, tableName)
    addon.Transport.pushSequence = (addon.Transport.pushSequence or 0) + 1
    return string.format("push:%s:%s:%d:%d", peer or "unknown", tableName or "table", time(), addon.Transport.pushSequence)
end

local function flushOneMessage()
    if addon.IsSyncSuspended and addon:IsSyncSuspended() then
        return
    end

    if #addon.Transport.outboundQueue == 0 then
        return
    end

    local entry = table.remove(addon.Transport.outboundQueue, 1)
    sendAddonMessage(entry.prefix, entry.payload, entry.channel, entry.peer)
    
    local session = addon.SessionManager:GetSession(entry.sessionId)
    
    if session
        and session.status == "active"
        and session.metadata
        and session.metadata.direction == "incoming"
        and entry.op == "SNAPSHOT_RES" then
        local hasMoreQueuedChunks = false
        
        for _, queuedEntry in ipairs(addon.Transport.outboundQueue) do
            if queuedEntry.sessionId == entry.sessionId then
                hasMoreQueuedChunks = true
                break
            end
        end
        
        if not hasMoreQueuedChunks then
            addon.SessionManager:CompleteSession(entry.sessionId)
        end
    end
end

local function applyPeerStatus(sender, payload)
    local status = addon.Reconciler:DecodeStatus(payload)

    if not status then
        return false, "status payload was invalid"
    end

    addon.Reconciler:RecordPeerStatus(sender, status)

    if addon.MainFrame and addon.MainFrame.MarkBackgroundRefreshPending then
        addon.MainFrame:MarkBackgroundRefreshPending()
    end

    return true
end

local function processOneInboundTask()
    if addon.IsSyncSuspended and addon:IsSyncSuspended() then
        return
    end

    if #addon.Transport.inboundTaskQueue == 0 then
        return
    end

    local task = table.remove(addon.Transport.inboundTaskQueue, 1)
    local session = addon.SessionManager:GetSession(task.sessionId)
    local taskScopeName = task and task.op and ("sync:task-" .. string.lower(tostring(task.op))) or nil
    local taskScopeStartedAt = taskScopeName and addon.BeginPerfScope and addon:BeginPerfScope(taskScopeName) or nil
    local function finishTaskScope()
        if taskScopeStartedAt ~= nil and addon.EndPerfScope then
            addon:EndPerfScope(taskScopeName, taskScopeStartedAt)
        end
    end

    if session and session.status == "cancelled" then
        finishTaskScope()
        return
    end

    local payload = table.concat(task.chunks)

    if task.op == "STATUS_RES" then
        local ok, errorMessage = applyPeerStatus(task.sender, payload)

        if ok then
            addon.SessionManager:CompleteSession(task.sessionId)
            finishTaskScope()
            return
        end

        addon.SessionManager:FailSession(task.sessionId, errorMessage or "failed to apply peer status")
        finishTaskScope()
        return
    end

    if task.op == "ROSTER_JOURNAL_RES" then
        local rosterPayload, errorMessage = addon.Reconciler:DeserializeRosterJournalPayload(payload)

        if not rosterPayload then
            addon.SessionManager:FailSession(task.sessionId, errorMessage)
            finishTaskScope()
            return
        end

        addon.Reconciler:ApplyRosterJournalPayload(rosterPayload, task.sender, {
            sessionId = task.sessionId,
        }, function(ok, applyError)
            finishTaskScope()

            if ok then
                addon.SessionManager:CompleteSession(task.sessionId)
                return
            end

            local currentSession = addon.SessionManager:GetSession(task.sessionId)
            if currentSession and currentSession.status == "cancelled" then
                return
            end

            addon.SessionManager:FailSession(task.sessionId, applyError or "failed to apply roster journal payload")
        end)

        return
    end

    if task.op == "MANIFEST_RES" then
        local manifest, errorMessage = addon.Reconciler:DeserializeManifest(payload)

        if not manifest then
            addon.SessionManager:FailSession(task.sessionId, errorMessage)
            finishTaskScope()
            return
        end

        local requestedIds = addon.Reconciler:GetDeltaRequestIds(task.tableName, manifest, task.sender)
        addon.SessionManager:AddRequestedFromPeer(task.sessionId, #requestedIds)

        if #requestedIds == 0 then
            if session then
                addon.SessionManager:CompleteSession(task.sessionId)
                addon.Reconciler:Notify("debug", task.tableName .. " already matched requested peer manifest.")
                finishTaskScope()
                return
            end

            local emptyRequestPayload = addon.Json:Encode({})
            local ok, sendError = addon.Transport:SendRecordRequest(task.sender, task.tableName, task.sessionId, emptyRequestPayload)

            if not ok and addon.Reconciler and addon.Reconciler.Notify then
                addon.Reconciler:Notify("debug", string.format("Failed to acknowledge no-op %s push from %s: %s", task.tableName or "table", task.sender or "unknown", sendError or "unknown error"))
            end

            finishTaskScope()

            return
        end

        local requestedPayload = addon.Json:Encode(requestedIds)
        local ok, sendError = addon.Transport:SendRecordRequest(task.sender, task.tableName, task.sessionId, requestedPayload)

        if not ok then
            addon.SessionManager:FailSession(task.sessionId, sendError)
            finishTaskScope()
            return
        end

        finishTaskScope()

        return
    end

    if task.op == "RECORD_REQ" then
        local ids, errorMessage = addon.Reconciler:DeserializeRequestedIds(payload)

        if not ids then
            addon.SessionManager:FailSession(task.sessionId, errorMessage)
            finishTaskScope()
            return
        end

        addon.SessionManager:AddRequestedByPeer(task.sessionId, #ids)

        local recordsPayload, serializeError = addon.Reconciler:SerializeRequestedRecords(task.tableName, ids)

        if not recordsPayload then
            addon.SessionManager:FailSession(task.sessionId, serializeError)
            finishTaskScope()
            return
        end

        local ok, queueError = sendChunked("RECORD_RES", task.sender, task.tableName, task.sessionId, recordsPayload)

        if not ok then
            addon.SessionManager:FailSession(task.sessionId, queueError)
            finishTaskScope()
            return
        end

        if session and session.status == "active" and session.peer == task.sender and session.tableName == task.tableName then
            addon.SessionManager:CompleteSession(task.sessionId)
        end

        finishTaskScope()

        return
    end

    local snapshot, errorMessage = addon.Reconciler:DeserializeTable(payload)

    if not snapshot then
        addon.SessionManager:FailSession(task.sessionId, errorMessage)
        finishTaskScope()
        return
    end

    local options = {
        sessionId = task.sessionId,
    }

    if type(task.metadata) == "table" then
        for key, value in pairs(task.metadata) do
            options[key] = value
        end
    end

    addon.Reconciler:ApplySnapshotAsync(task.tableName, snapshot, task.sender, options, function(ok, applyError)
        finishTaskScope()

        if ok then
            addon.SessionManager:CompleteSession(task.sessionId)
            return
        end

        local currentSession = addon.SessionManager:GetSession(task.sessionId)
        if currentSession and currentSession.status == "cancelled" then
            return
        end

        addon.SessionManager:FailSession(task.sessionId, applyError or "failed to apply incoming snapshot")
    end)
end

function addon.Transport:EnsurePump()
    if (addon.IsSyncSuspended and addon:IsSyncSuspended()) or self.sendTicker or not (C_Timer and C_Timer.NewTicker) then
        return
    end

    self.sendTicker = C_Timer.NewTicker(SEND_INTERVAL_SECONDS, function()
        local perfScopeName = "sync:send-tick"
        local perfStartedAt = addon.BeginPerfScope and addon:BeginPerfScope(perfScopeName) or nil

        if #addon.Transport.outboundQueue == 0 then
            if addon.Transport.sendTicker then
                addon.Transport.sendTicker:Cancel()
                addon.Transport.sendTicker = nil
            end

            if perfStartedAt ~= nil and addon.EndPerfScope then
                addon:EndPerfScope(perfScopeName, perfStartedAt)
            end
            return
        end

        flushOneMessage()

        if perfStartedAt ~= nil and addon.EndPerfScope then
            addon:EndPerfScope(perfScopeName, perfStartedAt)
        end
    end)
end

function addon.Transport:EnsureTaskPump()
    if (addon.IsSyncSuspended and addon:IsSyncSuspended()) or self.taskTicker or not (C_Timer and C_Timer.NewTicker) then
        return
    end

    self.taskTicker = C_Timer.NewTicker(INBOUND_PROCESS_INTERVAL_SECONDS, function()
        local perfScopeName = "sync:task-tick"
        local perfStartedAt = addon.BeginPerfScope and addon:BeginPerfScope(perfScopeName) or nil

        if #addon.Transport.inboundTaskQueue == 0 then
            if addon.Transport.taskTicker then
                addon.Transport.taskTicker:Cancel()
                addon.Transport.taskTicker = nil
            end

            if perfStartedAt ~= nil and addon.EndPerfScope then
                addon:EndPerfScope(perfScopeName, perfStartedAt)
            end
            return
        end

        processOneInboundTask()

        if perfStartedAt ~= nil and addon.EndPerfScope then
            addon:EndPerfScope(perfScopeName, perfStartedAt)
        end
    end)
end

function addon.Transport:CleanupStaleBuffers(now)
    local currentTime = now or time()
    local cleared = 0

    for sessionId, buffer in pairs(self.incomingSnapshots) do
        local lastSeenAt = buffer.lastSeenAt or buffer.startedAt or currentTime

        if (currentTime - lastSeenAt) > BUFFER_TIMEOUT_SECONDS then
            self.incomingSnapshots[sessionId] = nil
            addon.SessionManager:FailSession(sessionId, "incoming snapshot timed out")
            cleared = cleared + 1
        end
    end

    return cleared
end

function addon.Transport:EnsureMaintenance()
    if (addon.IsSyncSuspended and addon:IsSyncSuspended()) or self.maintenanceTicker or not (C_Timer and C_Timer.NewTicker) then
        return
    end

    self.maintenanceTicker = C_Timer.NewTicker(MAINTENANCE_INTERVAL_SECONDS, function()
        local now = time()
        addon.SessionManager:CleanupExpired(now)
        addon.Transport:CleanupStaleBuffers(now)
    end)
end

function addon.Transport:BuildDiagnostics()
    local diagnostics = {
        outboundQueueDepth = #self.outboundQueue,
        inboundBufferCount = 0,
        inboundBufferedChunks = 0,
        largestInboundBuffer = 0,
        inboundTaskQueueDepth = #self.inboundTaskQueue,
        sendPumpActive = self.sendTicker and true or false,
        maintenanceActive = self.maintenanceTicker and true or false,
        inboundTaskPumpActive = self.taskTicker and true or false,
    }

    for _, buffer in pairs(self.incomingSnapshots) do
        diagnostics.inboundBufferCount = diagnostics.inboundBufferCount + 1

        local chunkCount = 0
        for _ in pairs(buffer.chunks or {}) do
            chunkCount = chunkCount + 1
        end

        diagnostics.inboundBufferedChunks = diagnostics.inboundBufferedChunks + chunkCount
        diagnostics.largestInboundBuffer = math.max(diagnostics.largestInboundBuffer, chunkCount)
    end

    return diagnostics
end

function addon.Transport:CancelSession(sessionId)
    local nextOutboundQueue = {}

    for _, entry in ipairs(self.outboundQueue) do
        if entry.sessionId ~= sessionId then
            nextOutboundQueue[#nextOutboundQueue + 1] = entry
        end
    end

    self.outboundQueue = nextOutboundQueue
    self.incomingSnapshots[sessionId] = nil

    local nextInboundTasks = {}
    for _, task in ipairs(self.inboundTaskQueue) do
        if task.sessionId ~= sessionId then
            nextInboundTasks[#nextInboundTasks + 1] = task
        end
    end

    self.inboundTaskQueue = nextInboundTasks
end

function addon.Transport:CancelAllSessions(reason)
    local cancelled = 0
    local message = reason or "sync cancelled by user"

    for sessionId, session in pairs(addon.SessionManager.activeSessions or {}) do
        if session.status == "active" then
            self:CancelSession(sessionId)
            addon.SessionManager:CancelSession(sessionId, message)
            cancelled = cancelled + 1
        end
    end

    return cancelled
end

function addon.Transport:Pause()
    if self.sendTicker then
        self.sendTicker:Cancel()
        self.sendTicker = nil
    end

    if self.taskTicker then
        self.taskTicker:Cancel()
        self.taskTicker = nil
    end

    if self.maintenanceTicker then
        self.maintenanceTicker:Cancel()
        self.maintenanceTicker = nil
    end
end

function addon.Transport:Resume()
    self:EnsureMaintenance()

    if #self.outboundQueue > 0 then
        self:EnsurePump()
    end

    if #self.inboundTaskQueue > 0 then
        self:EnsureTaskPump()
    end
end

function addon.Transport:Send(peer, channel, envelope)
    if addon.IsSyncSuspended and addon:IsSyncSuspended() then
        return false, "GH3 sync is paused right now"
    end

    if not ((C_ChatInfo and C_ChatInfo.SendAddonMessage) or SendAddonMessage) then
        return false, "addon messaging unavailable"
    end

    if (channel or "WHISPER") == "WHISPER" then
        local isValidPeer, normalizedPeer = addon.Store:IsLikelyPlayerFullName(peer, GetRealmName() or "Unknown")

        if not isValidPeer then
            return false, "invalid whisper peer"
        end

        peer = normalizedPeer
    end

    local prefix = addon.Store:GetSync().prefix
    local payload = addon.SyncProtocol:Encode(envelope)

    if #payload > MAX_SYNC_MESSAGE_BYTES then
        return false, string.format("sync payload too large (%d bytes)", #payload)
    end

    if #self.outboundQueue >= MAX_OUTBOUND_QUEUE then
        return false, "outbound sync queue is full"
    end

    self.outboundQueue[#self.outboundQueue + 1] = {
        prefix = prefix,
        payload = payload,
        channel = channel or "WHISPER",
        peer = peer,
        sessionId = envelope.sessionId,
        op = envelope.op,
        tableName = envelope.tableName,
    }

    addon.SessionManager:RecordOutboundEnvelope(envelope.sessionId, envelope.op, #payload)

    self:EnsureMaintenance()
    self:EnsurePump()
    return true
end

function addon.Transport:EnvelopeFits(envelope)
    return getEncodedEnvelopeSize(envelope) <= MAX_SYNC_MESSAGE_BYTES
end

function addon.Transport:SendSnapshotPush(peer, tableName, payload)
    local sessionId = nextPushSessionId(peer, tableName)
    return sendChunked("SNAPSHOT_PUSH", peer, tableName, sessionId, payload or "{}")
end

function addon.Transport:SendManifestPush(peer, tableName, sessionId, payload)
    local resolvedSessionId = sessionId or nextPushSessionId(peer, tableName)
    local ok, errorMessage = sendChunked("MANIFEST_RES", peer, tableName, resolvedSessionId, payload or "{}")

    if not ok then
        return false, errorMessage
    end

    return true, resolvedSessionId
end

function addon.Transport:SendRecordRequest(peer, tableName, sessionId, payload)
    local resolvedSessionId = sessionId or nextPushSessionId(peer, tableName)
    local ok, errorMessage = sendChunked("RECORD_REQ", peer, tableName, resolvedSessionId, payload or "[]")

    if not ok then
        return false, errorMessage
    end

    return true, resolvedSessionId
end

function addon.Transport:HandleIncoming(prefix, message, channel, sender)
    if addon.IsSyncSuspended and addon:IsSyncSuspended() then
        return
    end

    if prefix ~= addon.Store:GetSync().prefix then
        return
    end

    self:EnsureMaintenance()

    local envelope = addon.SyncProtocol:Decode(message)

    if not envelope then
        addon:Debug("Ignored invalid sync payload from", sender)
        return
    end

    if sender == addon.Store:GetPlayerName() then
        return
    end

    addon.SessionManager:RecordInboundEnvelope(envelope.sessionId, envelope.op, #message)

    if envelope.op ~= "STATUS_REQ" and envelope.op ~= "STATUS_RES" and envelope.op ~= "FED_QUERY" and envelope.op ~= "FED_STATUS" and envelope.op ~= "FED_REQUEST" and envelope.op ~= "FED_APPROVE" then
        addon.Reconciler:MarkPeerSeen(sender)
    end

    local manualOnly = addon.Reconciler:IsManualOnlySync()

    if manualOnly and (envelope.op == "STATUS_REQ" or envelope.op == "SNAPSHOT_REQ" or envelope.op == "MANIFEST_REQ" or envelope.op == "RECORD_REQ" or envelope.op == "SYNC_BACK_REQ" or envelope.op == "ROSTER_JOURNAL_REQ") then
        local matchingSession = envelope.sessionId ~= "" and addon.SessionManager:GetSession(envelope.sessionId) or nil

        if not (envelope.op == "RECORD_REQ" and matchingSession and matchingSession.status == "active") then
            return
        end
    end

    if envelope.op == "FED_QUERY" then
        if addon.FederationService and addon.FederationService.HandleHandshakeQuery then
            addon.FederationService:HandleHandshakeQuery(sender)

            if addon.MainFrame and addon.MainFrame.MarkBackgroundRefreshPending then
                addon.MainFrame:MarkBackgroundRefreshPending()
            end
        end
        return
    end

    if envelope.op == "FED_STATUS" then
        local payload = addon.Json:Decode(envelope.payload)

        if addon.FederationService and addon.FederationService.HandleHandshakeStatus then
            addon.FederationService:HandleHandshakeStatus(sender, payload)

            if addon.MainFrame and addon.MainFrame.MarkBackgroundRefreshPending then
                addon.MainFrame:MarkBackgroundRefreshPending()
            end
        end
        return
    end

    if envelope.op == "FED_REQUEST" then
        local payload = addon.Json:Decode(envelope.payload)

        if addon.FederationService and addon.FederationService.HandleLinkRequest then
            addon.FederationService:HandleLinkRequest(sender, payload)

            if addon.MainFrame and addon.MainFrame.MarkBackgroundRefreshPending then
                addon.MainFrame:MarkBackgroundRefreshPending()
            end
        end
        return
    end

    if envelope.op == "FED_APPROVE" then
        local payload = addon.Json:Decode(envelope.payload)

        if addon.FederationService and addon.FederationService.HandleLinkApproval then
            addon.FederationService:HandleLinkApproval(sender, payload)

            if addon.MainFrame and addon.MainFrame.MarkBackgroundRefreshPending then
                addon.MainFrame:MarkBackgroundRefreshPending()
            end
        end
        return
    end

    if envelope.op == "STATUS_REQ" then
        local statusReqPerfScopeName = "transport:status-req"
        local statusReqPerfStartedAt = addon.BeginPerfScope and addon:BeginPerfScope(statusReqPerfScopeName) or nil
        local function finishStatusReqPerf()
            if statusReqPerfStartedAt ~= nil and addon.EndPerfScope then
                addon:EndPerfScope(statusReqPerfScopeName, statusReqPerfStartedAt)
            end
        end

        local payloadPerfStartedAt = addon.BeginPerfScope and addon:BeginPerfScope("transport:status-req-payload") or nil
        local payload = addon.Reconciler.GetCachedEncodedStatus
            and addon.Reconciler:GetCachedEncodedStatus(15)
            or addon.Reconciler:EncodeStatus(addon.Reconciler:BuildLocalStatus())

        if payloadPerfStartedAt ~= nil and addon.EndPerfScope then
            addon:EndPerfScope("transport:status-req-payload", payloadPerfStartedAt)
        end

        local responseEnvelope = {
            op = "STATUS_RES",
            sessionId = envelope.sessionId,
            tableName = "all",
            payload = payload,
        }

        local sendPerfStartedAt = addon.BeginPerfScope and addon:BeginPerfScope("transport:status-req-send") or nil
        local ok, errorMessage

        if encodedPayloadFits(responseEnvelope.op, responseEnvelope.sessionId, responseEnvelope.tableName, responseEnvelope.payload, responseEnvelope.seq, responseEnvelope.total) then
            ok, errorMessage = self:Send(sender, "WHISPER", responseEnvelope)
        else
            ok, errorMessage = sendChunked("STATUS_RES", sender, "all", envelope.sessionId, responseEnvelope.payload)
        end

        if sendPerfStartedAt ~= nil and addon.EndPerfScope then
            addon:EndPerfScope("transport:status-req-send", sendPerfStartedAt)
        end

        if not ok and addon.Reconciler and addon.Reconciler.Notify then
            addon.Reconciler:Notify("debug", string.format("Failed status response to %s: %s", sender or "unknown", errorMessage or "failed"))
        end

        finishStatusReqPerf()
        return
    end

    if envelope.op == "STATUS_RES" then
        if (tonumber(envelope.total) or 0) > 1 then
            local buffer = self.incomingSnapshots[envelope.sessionId]

            if not buffer then
                buffer = {
                    sender = sender,
                    tableName = envelope.tableName,
                    total = envelope.total,
                    op = envelope.op,
                    chunks = {},
                    startedAt = time(),
                }
                self.incomingSnapshots[envelope.sessionId] = buffer
                addon.SessionManager:SetTotal(envelope.sessionId, envelope.total)
            end

            buffer.total = envelope.total
            buffer.op = envelope.op
            buffer.chunks[envelope.seq] = envelope.payload
            buffer.lastSeenAt = time()
            addon.SessionManager:Advance(envelope.sessionId)
            addon.SessionManager:Touch(envelope.sessionId)

            for index = 1, buffer.total do
                if not buffer.chunks[index] then
                    return
                end
            end

            if #self.inboundTaskQueue >= MAX_INBOUND_TASK_QUEUE then
                addon.SessionManager:FailSession(envelope.sessionId, "inbound sync task queue is full")
                self.incomingSnapshots[envelope.sessionId] = nil
                return
            end

            self.inboundTaskQueue[#self.inboundTaskQueue + 1] = {
                sessionId = envelope.sessionId,
                sender = sender,
                tableName = buffer.tableName,
                op = buffer.op,
                chunks = buffer.chunks,
            }
            self.incomingSnapshots[envelope.sessionId] = nil
            self:EnsureTaskPump()
            return
        end

        applyPeerStatus(sender, envelope.payload)

        return
    end

    if envelope.op == "SYNC_BACK_REQ" then
        local ok, message = addon.Reconciler:RequestSnapshot(sender, envelope.tableName)

        if not ok and addon.Reconciler and addon.Reconciler.Notify then
            addon.Reconciler:Notify("debug", string.format("Ignored sync-back request for %s from %s: %s", envelope.tableName or "table", sender or "unknown", message or "failed"))
        end

        return
    end

    if envelope.op == "SNAPSHOT_REQ" then
        local canSync, permissionError = addon.Reconciler:CanSyncPeerTable(sender, envelope.tableName)

        if not canSync then
            if addon.Reconciler and addon.Reconciler.Notify then
                addon.Reconciler:Notify("debug", string.format("Ignored snapshot request for %s from %s: %s", envelope.tableName or "table", sender or "unknown", permissionError or "not permitted"))
            end
            return
        end

        addon.SessionManager:RegisterSession(envelope.sessionId, sender, envelope.tableName, 0, {
            direction = "incoming",
            resolution = "serve-snapshot",
        })

        local payload, errorMessage = addon.Reconciler:SerializeTable(envelope.tableName)

        if not payload then
            addon.SessionManager:FailSession(envelope.sessionId, errorMessage)
            return
        end

        local ok, queueError = sendChunked("SNAPSHOT_RES", sender, envelope.tableName, envelope.sessionId, payload)
        if not ok then
            addon.SessionManager:FailSession(envelope.sessionId, queueError)
        end
        return
    end

    if envelope.op == "MANIFEST_REQ" then
        local canSync, permissionError = addon.Reconciler:CanSyncPeerTable(sender, envelope.tableName)

        if not canSync then
            if addon.Reconciler and addon.Reconciler.Notify then
                addon.Reconciler:Notify("debug", string.format("Ignored manifest request for %s from %s: %s", envelope.tableName or "table", sender or "unknown", permissionError or "not permitted"))
            end
            return
        end

        addon.SessionManager:RegisterSession(envelope.sessionId, sender, envelope.tableName, 0, {
            direction = "incoming",
            resolution = "serve-delta",
        })

        local payload, errorMessage = addon.Reconciler:SerializeManifest(envelope.tableName)

        if not payload then
            addon.SessionManager:FailSession(envelope.sessionId, errorMessage)
            return
        end

        local ok, queueError = sendChunked("MANIFEST_RES", sender, envelope.tableName, envelope.sessionId, payload)
        if not ok then
            addon.SessionManager:FailSession(envelope.sessionId, queueError)
        end
        return
    end

    if envelope.op == "ROSTER_JOURNAL_REQ" then
        local canSync, permissionError = addon.Reconciler:CanSyncPeerTable(sender, envelope.tableName)

        if not canSync then
            if addon.Reconciler and addon.Reconciler.Notify then
                addon.Reconciler:Notify("debug", string.format("Ignored roster journal request for %s from %s: %s", envelope.tableName or "table", sender or "unknown", permissionError or "not permitted"))
            end
            return
        end

        addon.SessionManager:RegisterSession(envelope.sessionId, sender, envelope.tableName, 0, {
            direction = "incoming",
            resolution = "serve-roster-journal",
        })

        local requestPayload, requestError = addon.Reconciler:DeserializeRosterJournalRequest(envelope.payload)

        if not requestPayload then
            addon.SessionManager:FailSession(envelope.sessionId, requestError)
            return
        end

        local responsePayload, responseError = addon.Reconciler:BuildRosterJournalResponse(sender, requestPayload)

        if not responsePayload then
            addon.SessionManager:FailSession(envelope.sessionId, responseError)
            return
        end

        local ok, queueError = sendChunked("ROSTER_JOURNAL_RES", sender, envelope.tableName, envelope.sessionId, responsePayload)

        if not ok then
            addon.SessionManager:FailSession(envelope.sessionId, queueError)
        end

        return
    end

    if envelope.op == "MANIFEST_RES" or envelope.op == "RECORD_REQ" or envelope.op == "RECORD_RES" or envelope.op == "SNAPSHOT_RES" or envelope.op == "SNAPSHOT_PUSH" or envelope.op == "ROSTER_JOURNAL_RES" or envelope.op == "STATUS_RES" then
        local buffer = self.incomingSnapshots[envelope.sessionId]
        local session = addon.SessionManager:GetSession(envelope.sessionId)

        if not buffer then
            buffer = {
                sender = sender,
                tableName = envelope.tableName,
                total = envelope.total,
                op = envelope.op,
                chunks = {},
                startedAt = time(),
            }
            self.incomingSnapshots[envelope.sessionId] = buffer
            addon.SessionManager:SetTotal(envelope.sessionId, envelope.total)
        end

        buffer.total = envelope.total
        buffer.op = envelope.op
        buffer.chunks[envelope.seq] = envelope.payload
        buffer.lastSeenAt = time()
        addon.SessionManager:Advance(envelope.sessionId)
        addon.SessionManager:Touch(envelope.sessionId)

        for index = 1, buffer.total do
            if not buffer.chunks[index] then
                return
            end
        end
        if #self.inboundTaskQueue >= MAX_INBOUND_TASK_QUEUE then
            addon.SessionManager:FailSession(envelope.sessionId, "inbound sync task queue is full")
            self.incomingSnapshots[envelope.sessionId] = nil
            return
        end

        self.inboundTaskQueue[#self.inboundTaskQueue + 1] = {
            sessionId = envelope.sessionId,
            sender = sender,
            tableName = buffer.tableName,
            op = buffer.op,
            chunks = buffer.chunks,
            metadata = session and session.metadata or nil,
        }
        self.incomingSnapshots[envelope.sessionId] = nil
        self:EnsureTaskPump()
        return
    end

    addon:Debug("Received sync", envelope.op, "for", envelope.tableName, "from", sender, "via", channel)
end
