local addonName, TOAST = ...

-- CS1 transport and orchestration. All writes go through crew_state.lua and
-- every TOAST send goes through comms_queue.lua.

local COMMUNITY_NAME = "TIDE and Crew"
local TIDE_GUILD_NAME = "tide"
local TRANSFER_TTL = 90
local CLAIM_WINDOW = 2
local ACK_TIMEOUT = 12
local BROKER_BACKOFF = { 300, 900, 1800 }
local BRIDGE_SUCCESS_COOLDOWN = 15 * 60
local BRIDGE_EXHAUSTED_COOLDOWN = 30 * 60
local MAX_CHUNKS = 80
local CHUNK_PAYLOAD = 170

local communityClubID
local incoming = {}
local claims = {}
local outgoing = {}
local expectedCache = {}
local syncRequests = {}
local bridge
local bridgeBlockedReason
local bridgeBlockLogged = false
local harvest = {}
local pendingGuildRequest = false
local lastGuildStateRequestAt = 0

C_ChatInfo.RegisterAddonMessagePrefix("TOAST")
C_ChatInfo.RegisterAddonMessagePrefix("LibKS")

local function Now()
    return GetServerTime()
end

local function IsSecret(value)
    return type(issecretvalue) == "function" and issecretvalue(value) == true
end

local function NewRequestId()
    return string.format("%X%X", Now() % 0xFFFF, math.random(0, 0xFFFF))
end

local function StableFetchRequestId(identity)
    local hash = math.floor(Now() / 300) % 0xFFFF
    for index = 1, #identity do hash = (hash * 33 + identity:byte(index)) % 0xFFFFFF end
    return string.format("F%X", hash)
end

local function SenderIdentity(sender)
    if type(sender) ~= "string" then return nil end
    local name, realm = sender:match("^([^%-]+)%-(.+)$")
    return name or sender, realm
end

local function FullName(name, realm)
    return TOAST.FormatCharacterTarget(name, realm)
end

local function ResolveMember(name, realm)
    return TOAST.CrewState and TOAST.CrewState.ResolveMember(name, realm) or nil
end

local function ResolveSender(sender)
    local name, realm = SenderIdentity(sender)
    return ResolveMember(name, realm)
end

local function RankKind(member)
    if not member then return nil end

    -- members.lua carries the roster identity, while the authoritative
    -- TIDE/CPT marker is currently the generated leaderboard's Discord
    -- prefix.  The latter is deliberately independent of a linked Discord
    -- ID; it is bot-exported roster metadata.
    local function FromPrefix(value)
        if type(value) ~= "string" or value == "" then return nil end
        local parser = TOAST.ParseDiscordPrefix
        local parsed = parser and parser(value) or nil
        if parsed and (parsed.label == "TIDE" or parsed.label == "CPT") then
            return parsed.label
        end
        local upper = value:upper()
        if upper:find("TIDE", 1, true) or value:find("ᴛɪᴅᴇ", 1, true) then return "TIDE" end
        if upper:find("CPT", 1, true) or value:find("ᴄᴘᴛ", 1, true) then return "CPT" end
    end

    local direct = FromPrefix(member.prefix) or FromPrefix(member.rank) or FromPrefix(member.discord_nickname)
    if direct then return direct end

    local targetName = tostring(member.name or ""):lower()
    local targetRealm = tostring(member.realm or ""):gsub("[%s%-']", ""):lower()
    for _, entry in ipairs(TOAST.leaderboard or {}) do
        local entryName = tostring(entry.name or ""):lower()
        local entryRealm = tostring(entry.realm or ""):gsub("[%s%-']", ""):lower()
        if entryName == targetName and entryRealm == targetRealm then
            return FromPrefix(entry.discord_nickname)
        end
    end
    return nil
end

local function IsTIDE(member)
    return RankKind(member) ~= nil
end

-- Fetching is deliberately narrower than accepting a package. Any known crew
-- member may request or contribute data, but an automatic external fetch only
-- spends throttle on a character we can actually show in a leaderboard crew.
local function IsOnLeaderboard(member)
    if not member then return false end
    local targetName = tostring(member.name or ""):lower()
    local targetRealm = tostring(member.realm or ""):gsub("[%s%-']", ""):lower()
    if targetName == "" or targetRealm == "" then return false end
    for _, entry in ipairs(TOAST.leaderboard or {}) do
        local entryName = tostring(entry.name or ""):lower()
        local entryRealm = tostring(entry.realm or ""):gsub("[%s%-']", ""):lower()
        if entryName == targetName and entryRealm == targetRealm then return true end
        -- /crew is the authoritative alt relationship: a character of a
        -- leaderboard member shares that member's Discord ID in members.lua.
        local leaderboardMember = ResolveMember(entry.name, entry.realm)
        if member.discord_id and leaderboardMember and leaderboardMember.discord_id
            and member.discord_id == leaderboardMember.discord_id then
            return true
        end
    end
    return false
end

local function PlayerMember()
    return ResolveMember(UnitName("player"), GetRealmName())
end

-- Is a persisted TOAST_DB.me actually a previously confirmed identity, rather
-- than a half-written or outdated SavedVariables leftover?  discord_id is the
-- minimum: it is the only field that ties this installation to a real crew
-- record, and every consumer keys off it (see ShowCrewPopup, /crew, trends).
-- The rest of the table is display sugar that may legitimately be absent.
-- discord_id itself is an opaque equality key everywhere in the addon, so
-- accept either scalar form but never an empty or non-scalar one.
local function HasConfirmedIdentity(me)
    if type(me) ~= "table" then return false end
    local id = me.discord_id
    if type(id) == "number" then return true end
    return type(id) == "string" and id ~= ""
end

local function PlayerHasTideIdentity()
    return IsTIDE(PlayerMember())
end

local function LocalGuildName()
    local guildName = GetGuildInfo and GetGuildInfo("player")
    return type(guildName) == "string" and guildName or nil
end

local function PlayerCanUseTideGuildTransport()
    if not PlayerHasTideIdentity() or not IsInGuild or not IsInGuild() then return false end
    local guildName = LocalGuildName()
    return guildName and guildName:lower() == TIDE_GUILD_NAME or false
end

local function CurrentSeason()
    return TOAST.GetCurrentMythicPlusSeasonID and TOAST.GetCurrentMythicPlusSeasonID() or nil
end

local function IsLivePresence(presence)
    return presence == Enum.ClubMemberPresence.Online
        or presence == Enum.ClubMemberPresence.Away
        or presence == Enum.ClubMemberPresence.Busy
end

local function GetCommunityClubId()
    if communityClubID then return communityClubID end
    if InCombatLockdown() then return nil end

    -- C_Club can return secret values during dungeon/combat UI transitions,
    -- even after InCombatLockdown() becomes false. Keep the entire read and
    -- iteration inside pcall so a temporary protected result simply means
    -- "not readable yet", never a Lua error.
    local ok, foundClubID = pcall(function()
        local clubs = C_Club.GetSubscribedClubs()
        if type(clubs) ~= "table" then return nil end
        for _, club in ipairs(clubs) do
            if club.name == COMMUNITY_NAME and type(club.clubId) == "number" then
                return club.clubId
            end
        end
        return nil
    end)
    if ok and type(foundClubID) == "number" then
        communityClubID = foundClubID
        return communityClubID
    end
end

local function ScanClub(clubId, result, source)
    if not clubId then return false end

    -- Build a temporary result first. If one Club value becomes protected
    -- midway through the scan, do not retain a misleading partial roster.
    local collected = {}
    local ok, scanned = pcall(function()
        local ids = C_Club.GetClubMembers(clubId)
        if type(ids) ~= "table" then return false end
        for _, memberId in ipairs(ids) do
            local info = C_Club.GetMemberInfo(clubId, memberId)
            if info and info.name and IsLivePresence(info.presence) then
                local name, realm = SenderIdentity(info.name)
                local member = ResolveMember(name, realm)
                if member then
                    -- members.lua authorizes the crew identity, but C_Club owns
                    -- the live whisper address. The roster entry may not carry a
                    -- realm for every character; rebuilding a target from it can
                    -- turn an online cross-realm member into a bare-name whisper.
                    -- Dedupe key: the roster row's canonical identity, built by
                    -- the SAME helper the crew index uses. Never join the name
                    -- to a raw realm_slug here -- "tarren-mill" would then be
                    -- compared against live "TarrenMill"/"Tarren Mill" keys and
                    -- never match. Identity and whisper targeting stay separate.
                    local identity = TOAST.CrewState.MemberIdentity(member)
                    local target = FullName(name, realm)
                    if identity and target and target ~= "" then
                        collected[identity] = {
                            member = member,
                            target = target,
                            info = info,
                        }
                    end
                end
            end
        end
        return true
    end)
    if not ok or not scanned then return false end

    for identity, item in pairs(collected) do
        local existing = result[identity] or {}
        existing.member, existing.target, existing.info = item.member, item.target, item.info
        existing[source] = true
        result[identity] = existing
    end
    return true
end

local function OnlineRoster()
    if InCombatLockdown() then return nil end
    local result, scanned = {}, false
    scanned = ScanClub(GetCommunityClubId(), result, "inCommunity") or scanned
    -- This is only a shared-guild observation. It becomes Tide transport
    -- evidence solely after PlayerCanUseTideGuildTransport() confirms that
    -- the local character is actually in guild Tide.
    local ok, guildClubID = pcall(function() return C_Club.GetGuildClubId() end)
    if ok and type(guildClubID) == "number" then
        scanned = ScanClub(guildClubID, result, "inGuild") or scanned
    end
    return scanned and result or nil
end

function TOAST.IsInCommunity()
    if InCombatLockdown() then return TOAST.inCommunity or false end
    return GetCommunityClubId() ~= nil
end

-- Keyed on the shared identity key and carrying the realm, so a consumer never
-- has to resolve a bare character name back to a roster entry -- that step
-- silently dropped same-name characters on different realms.
function TOAST.GetOnlineMembers()
    local roster = OnlineRoster()
    if not roster then return nil end
    local result = {}
    for _, item in pairs(roster) do
        local identity = TOAST.CrewState.IdentityKey(item.member.name, item.member.realm)
        if identity then
            result[identity] = { name = item.member.name, realm = item.member.realm }
        end
    end
    return result
end

local function CandidateList(kind)
    local roster = OnlineRoster()
    if not roster then return nil end
    local own = FullName(UnitName("player"), GetRealmName()):lower()
    local result = {}
    for _, item in pairs(roster) do
        local eligible = kind == "broker" and IsTIDE(item.member) and IsOnLeaderboard(item.member)
            or kind == "external" and not item.inGuild and IsOnLeaderboard(item.member)
        if item.target:lower() ~= own and eligible then
            result[#result + 1] = {
                target = item.target,
                capable = TOAST.GetCrewCapability
                    and TOAST.GetCrewCapability(item.member.name, item.member.realm) ~= nil,
            }
        end
    end
    table.sort(result, function(a, b)
        if a.capable ~= b.capable then return a.capable end
        return a.target:lower() < b.target:lower()
    end)
    local targets = {}
    for _, candidate in ipairs(result) do targets[#targets + 1] = candidate.target end
    return targets
end

local function RecognizePlayer()
    -- Roster identity belongs to the active character, not to whichever
    -- character happens to have a leaderboard row.  In particular, a crew
    -- alt may be the first character logged in after an addon update.
    local playerName, playerRealm = UnitName("player"), GetRealmName()
    local member = ResolveMember(playerName, playerRealm)

    -- TOAST.me is the narrow claim: the character logged in right now is a
    -- verified roster member.  Clear it first so a character swap can never
    -- leave a stale one behind.
    TOAST.me = nil
    if not member then
        -- Roster external.  A TOAST_DB.me that was genuinely confirmed earlier
        -- deliberately survives: it is the remembered Discord identity of this
        -- installation (display and local account context), never proof that
        -- this character is crew.  Every capability gate resolves the live
        -- character instead -- see PlayerMember() and TOAST.GetActiveCharacter().
        --
        -- Surviving is a privilege of a valid record only.  A corrupt or
        -- pre-discord_id table has nothing to remember, and this is the last
        -- point that can drop it: on a crew character the block below rewrites
        -- it wholesale, so leaving it here would strand it for good.
        if not HasConfirmedIdentity(TOAST_DB.me) then
            TOAST_DB.me = nil
        end
        return
    end

    local exactIndex, exactEntry
    local crewIndex, crewEntry
    for index, entry in ipairs(TOAST.leaderboard or {}) do
        local entryMember = ResolveMember(entry.name, entry.realm)
        if entryMember == member then
            exactIndex, exactEntry = index, entry
            break
        end
        if not crewEntry and member.discord_id and entryMember
            and entryMember.discord_id == member.discord_id then
            crewIndex, crewEntry = index, entry
        end
    end

    -- An exact row is preferred.  Otherwise, an alt inherits only its
    -- crew's leaderboard context. Character-specific data must never leak
    -- from the crew's leaderboard character into the active alt.
    local rank, entry = exactIndex or crewIndex, exactEntry or crewEntry
    TOAST_DB.me = {
        -- Use the live unit identity for the Status Report. A roster record
        -- proves membership; it does not decide which character is logged in.
        name = playerName,
        realm = playerRealm,
        discord = entry and entry.discord_nickname or nil,
        discord_id = member.discord_id,
        seamen_rank = member.rank,
        rank = rank,
        rio = entry and entry.rio or nil,
        classToken = select(2, UnitClass("player")),
        role = member.role or (exactEntry and exactEntry.role) or nil,
        resi = exactEntry and exactEntry.resi or nil,
        class_ = member.class_ or (exactEntry and exactEntry.class_) or nil,
    }
    TOAST.me = TOAST_DB.me
end

local function ReadOwnedKeystone()
    if type(C_MythicPlus) ~= "table"
       or type(C_MythicPlus.GetOwnedKeystoneChallengeMapID) ~= "function"
       or type(C_MythicPlus.GetOwnedKeystoneLevel) ~= "function" then
        return 0, 0
    end

    local mapOK, mapID = pcall(C_MythicPlus.GetOwnedKeystoneChallengeMapID)
    local levelOK, level = pcall(C_MythicPlus.GetOwnedKeystoneLevel)
    if not mapOK or not levelOK or IsSecret(mapID) or IsSecret(level)
       or type(mapID) ~= "number" or type(level) ~= "number" then
        return 0, 0
    end
    if mapID <= 0 or level <= 0 then return 0, 0 end
    return mapID, level
end

local function CurrentState()
    if InCombatLockdown() then return nil end
    -- The single point where this client turns its own character into crew
    -- data (the SELF merges below plus the outgoing S payload).  A roster
    -- external character contributes neither, so refuse at the source rather
    -- than relying on every caller's transport gate to be unreachable.
    if not PlayerMember() then return nil end
    local name, realm = UnitName("player"), GetRealmName()
    local season = CurrentSeason()
    local score = TOAST.RefreshOwnScore and TOAST.RefreshOwnScore()
    score = score and score.score
    local mapID, level = ReadOwnedKeystone()
    if not season or type(score) ~= "number" then return nil end
    TOAST.MergeCrewKey(name, realm, mapID, level, "SELF", Now())
    TOAST.MergeCrewScore(name, realm, season, score, "SELF", Now(), season)
    return season, math.floor(score), mapID, level
end

local function Queue(priority, fields, channel, target, callback)
    if channel == "GUILD" and not PlayerCanUseTideGuildTransport() then
        if TOAST.LogCommsTrace then
            TOAST.LogCommsTrace("OUT", channel, fields, "Tide", "blocked: no Tide guild transport")
        end
        return false
    end
    return TOAST.QueueAddonMessage(priority, table.concat(fields, "|"), channel, target, callback)
end

local function SendSelfState(requestId, channel, target)
    local season, score, mapID, level = CurrentState()
    if not season then return false end
    local priority = requestId and requestId ~= "-" and TOAST.AddonMessagePriority.STATE_RESPONSE
        or TOAST.AddonMessagePriority.SELF_STATE
    return Queue(priority,
        { "CS1", "S", requestId or "-", TOAST.VERSION, season, score, mapID, level },
        channel, target)
end

-- Central keystone observer.
--
-- Every signal that can mean "my key changed" schedules the SAME check instead
-- of broadcasting on its own: CHALLENGE_MODE_MAPS_UPDATE, the dungeon
-- completion, the keystone item appearing or being rerolled/downgraded in the
-- bags, and the deferred check after combat ends. The check reads the owned
-- keystone once and only publishes when the value actually moved, so a burst
-- of BAG_UPDATE_DELAYED costs two API reads and nothing on the wire.
--
-- Blizzard's Mythic+ API does not necessarily reflect a new key by the time
-- CHALLENGE_MODE_COMPLETED fires, so an event that *expects* a change arms a
-- retry ladder: the observer keeps re-reading on a widening delay until the
-- key moves or the ladder runs out. That replaces the old single fixed
-- two-second timer, which silently lost the update whenever it fired early.
local KEY_CHECK_DEBOUNCE = 1.5
local KEY_RETRY_DELAYS = { 2, 4, 8, 15, 30 }
local FORCE_PUBLISH_INTERVAL = 60

local keyObserver = {
    known = false,      -- has a successful read happened this session?
    mapID = 0,
    level = 0,
    expecting = false,  -- an event told us a change is coming: run the ladder
    retry = 0,
    force = false,      -- publish even without a key change (score may differ)
    deferred = false,   -- combat blocked a check; PLAYER_REGEN_ENABLED resumes
    dueAt = 0,
    lastPublishAt = 0,
}
local keyTimer
local ScheduleKeyCheck

-- The single place this client puts its own current state on the wire.
-- SendSelfState -> CurrentState() performs the SELF key and score merges, so
-- the local crew state (and any open dashboard, via NotifyChanged) is updated
-- even on a client with no guild transport, where nothing is sent.
local function PublishOwnState()
    if InCombatLockdown() then
        keyObserver.deferred = true
        return false
    end
    local mapID, level = ReadOwnedKeystone()
    keyObserver.known, keyObserver.mapID, keyObserver.level = true, mapID, level
    keyObserver.lastPublishAt = GetTime()
    if PlayerCanUseTideGuildTransport() then
        return SendSelfState("-", "GUILD")
    end
    local merged = CurrentState() ~= nil
    -- Nothing reaches the wire here, so trace the observer locally instead --
    -- otherwise a client without guild transport shows no evidence at all that
    -- its key check ran.
    if TOAST.LogCommsTrace then
        TOAST.LogCommsTrace("LOCAL", "LOCAL", { "CS1", "S" }, "self",
            merged and "key check: merged locally" or "key check: no state")
    end
    return merged
end

local function RunKeyCheck()
    if InCombatLockdown() then
        keyObserver.deferred = true
        return
    end
    keyObserver.deferred = false

    local mapID, level = ReadOwnedKeystone()
    local changed = not keyObserver.known
        or mapID ~= keyObserver.mapID or level ~= keyObserver.level

    if changed then
        keyObserver.expecting, keyObserver.retry, keyObserver.force = false, 0, false
        PublishOwnState()
        return
    end

    -- No key change. A completed dungeon still moved the score, so honour a
    -- forced publish -- rate-limited, because CHALLENGE_MODE_MAPS_UPDATE fires
    -- every time the player opens the Mythic+ UI.
    if keyObserver.force and GetTime() - keyObserver.lastPublishAt >= FORCE_PUBLISH_INTERVAL then
        keyObserver.force = false
        PublishOwnState()
    end

    if keyObserver.expecting then
        keyObserver.retry = keyObserver.retry + 1
        local delay = KEY_RETRY_DELAYS[keyObserver.retry]
        if delay then
            ScheduleKeyCheck(delay)
        else
            keyObserver.expecting, keyObserver.retry, keyObserver.force = false, 0, false
        end
    end
end

-- Deduplicated scheduling: one pending check at a time. A later signal that
-- would fire no sooner than the pending one is simply absorbed; an earlier one
-- replaces it, so a burst of events collapses into a single read.
ScheduleKeyCheck = function(delay, opts)
    delay = delay or KEY_CHECK_DEBOUNCE
    opts = opts or {}
    if opts.expectChange then
        keyObserver.expecting, keyObserver.retry = true, 0
    end
    if opts.force then keyObserver.force = true end

    local dueAt = GetTime() + delay
    if keyTimer then
        if keyObserver.dueAt <= dueAt then return end
        keyTimer:Cancel()
    end
    keyObserver.dueAt = dueAt
    keyTimer = C_Timer.NewTimer(delay, function()
        keyTimer = nil
        RunKeyCheck()
    end)
end

local function SendUpdatePolicy(channel, target)
    local policy = TOAST.ShouldAnnounceUpdatePolicy and TOAST.ShouldAnnounceUpdatePolicy()
    if not policy then return false end
    return Queue(TOAST.AddonMessagePriority.CONTROL,
        { "CS1", "U", policy.policyId, policy.minimumVersion, "1" }, channel, target)
end

local function PackRecords(records)
    local chunks, current = {}, ""
    for _, record in ipairs(records) do
        if #record > CHUNK_PAYLOAD then return nil end
        if current == "" then current = record
        elseif #current + #record + 1 <= CHUNK_PAYLOAD then current = current .. ";" .. record
        else chunks[#chunks + 1], current = current, record end
    end
    if current ~= "" then chunks[#chunks + 1] = current end
    if #chunks > MAX_CHUNKS then return nil end
    return chunks
end

local function SendTransfer(kind, requestId, target, channel, records, onDone)
    local chunks = PackRecords(records)
    if not chunks then return false end
    if not TOAST.CanQueueAddonMessages or not TOAST.CanQueueAddonMessages(#chunks + 2) then
        return false
    end
    local priority = kind == "RELAY" and TOAST.AddonMessagePriority.BACKGROUND
        or TOAST.AddonMessagePriority.SNAPSHOT
    local left, failed = #chunks + 2, false
    local function Complete(status)
        if status ~= "sent" then failed = true end
        left = left - 1
        if left == 0 and onDone then onDone(failed and "failed" or "sent") end
    end
    Queue(priority, { "CS1", "M", kind, requestId, #chunks }, channel, target, Complete)
    for index, chunk in ipairs(chunks) do
        Queue(priority, { "CS1", "D", kind, requestId, index, chunk }, channel, target, Complete)
    end
    Queue(priority, { "CS1", "E", kind, requestId, #records }, channel, target, Complete)
    outgoing[requestId] = { kind = kind, target = target, sentAt = Now() }
    return true
end

local function SendCachePackage(requestId, target)
    CurrentState()
    return SendTransfer("CACHE", requestId, target, "WHISPER", TOAST.BuildCrewCacheRecords())
end

local function SendSnapshot(requestId, target)
    return SendTransfer("SNAP", requestId, target, "WHISPER", TOAST.BuildCrewCacheRecords())
end

local function RequestGuildState(reason, immediate)
    if pendingGuildRequest or InCombatLockdown() or not PlayerCanUseTideGuildTransport() then return false end
    local now = Now()
    if immediate and now - lastGuildStateRequestAt < 120 then return false end
    pendingGuildRequest = true
    C_Timer.NewTimer(immediate and 0 or math.random(8, 20), function()
        pendingGuildRequest = false
        if not InCombatLockdown() and PlayerCanUseTideGuildTransport() then
            lastGuildStateRequestAt = Now()
            Queue(TOAST.AddonMessagePriority.BACKGROUND,
                { "CS1", "Q", NewRequestId(), reason or "refresh" }, "GUILD")
        end
    end)
    return true
end

local function RequestCacheSync()
    if InCombatLockdown() or not PlayerCanUseTideGuildTransport() then return end
    local sync, now = TOAST.GetCrewStateSync(), Now()
    local currentWeek = TOAST.GetCrewStateWeekId and TOAST.GetCrewStateWeekId()
    local cacheEmpty = #TOAST.BuildCrewCacheRecords() == 0
    if not cacheEmpty and sync.lastCacheSyncSuccessAt and sync.lastCacheSyncWeekId == currentWeek
        and now - sync.lastCacheSyncSuccessAt < 21600 then return end
    if sync.lastCacheSyncAttemptAt and now - sync.lastCacheSyncAttemptAt < 300 then return end
    local requestId = NewRequestId()
    syncRequests[requestId] = Now()
    TOAST.SetCrewStateSyncField("lastCacheSyncAttemptAt", now)
    Queue(TOAST.AddonMessagePriority.BACKGROUND,
        { "CS1", "Y", requestId, sync.keyCacheWeekId or "-" }, "GUILD")
end

local function StartIncoming(kind, requestId, count, sender, channel)
    if not ({ CACHE = true, SNAP = true, RELAY = true })[kind] then return end
    count = tonumber(count)
    if not count or count < 0 or count > MAX_CHUNKS then return end
    if kind == "RELAY" and channel ~= "GUILD" then return end
    if kind ~= "RELAY" and channel ~= "WHISPER" then return end
    if kind == "RELAY" and (not PlayerCanUseTideGuildTransport() or not IsTIDE(ResolveSender(sender))) then return end
    if kind == "CACHE" then
        local expected = expectedCache[requestId]
        if not PlayerCanUseTideGuildTransport() or not expected or expected.sender:lower() ~= sender:lower() then return end
    elseif kind == "SNAP" then
        local validBridge = bridge and bridge.requestId == requestId
            and bridge.target and bridge.target:lower() == sender:lower()
        local validSync = syncRequests[requestId] and IsTIDE(ResolveSender(sender))
        if not validBridge and not validSync then return end
    end
    incoming[sender .. "\031" .. kind .. "\031" .. requestId] = {
        kind = kind, requestId = requestId, sender = sender, chunks = {},
        chunkCount = count, channel = channel, startedAt = Now(),
    }
end

local function AddChunk(kind, requestId, index, records, sender, channel)
    local transfer = incoming[sender .. "\031" .. kind .. "\031" .. requestId]
    index = tonumber(index)
    if not transfer or transfer.channel ~= channel or not index or index < 1 or index > transfer.chunkCount
        or type(records) ~= "string" or #records > CHUNK_PAYLOAD then return end
    transfer.chunks[index] = transfer.chunks[index] or records
end

local function MergeCompleted(kind, requestId, recordCount, sender, channel)
    local key = sender .. "\031" .. kind .. "\031" .. requestId
    local transfer = incoming[key]
    incoming[key] = nil
    recordCount = tonumber(recordCount)
    if not transfer or transfer.channel ~= channel or not recordCount or recordCount < 0 then return nil end
    local records = {}
    for index = 1, transfer.chunkCount do
        local chunk = transfer.chunks[index]
        if not chunk then return nil end
        for record in chunk:gmatch("[^;]+") do records[#records + 1] = record end
    end
    if #records ~= recordCount then return nil end
    local accepted = {}
    for _, record in ipairs(records) do
        local _, status = TOAST.MergeCrewCacheRecord(record, sender, CurrentSeason())
        if status == "updated" then accepted[#accepted + 1] = record end
    end
    return accepted
end

local function SendAck(requestId, role, target)
    Queue(TOAST.AddonMessagePriority.CONTROL,
        { "CS1", "A", requestId, role, TOAST.VERSION }, "WHISPER", target)
end

-- One bridge session at a time, and bridge.phase carries the entire retry
-- policy:
--   candidate     -- BridgeTryNext may pick and whisper the next broker
--   awaiting_ack  -- BRIDGE_REQUEST sent, broker ACK outstanding
--   transferring  -- broker ACKed; our CACHE and its SNAP are in flight
--   backoff       -- a full candidate round failed; only its timer resumes
-- Any other caller must leave an existing bridge alone. Otherwise a login,
-- club load, regen or manual refresh opens a second request mid-transfer.
local BridgeTryNext

-- bridge.requestId is stable across a round, so timers also need the attempt
-- counter to tell their own candidate from a later one.
local function BridgeStale(requestId, attempt)
    return not bridge or bridge.requestId ~= requestId or bridge.attempt ~= attempt
end

local function BridgeRetry()
    if not bridge or bridge.phase == "backoff" then return end
    bridge.phase = "candidate"
    BridgeTryNext()
end

BridgeTryNext = function()
    if not bridge or bridge.phase ~= "candidate" then return end
    -- Prefixes are only broker hints here. The recipient's own local Tide
    -- guild gate decides whether it can actually acknowledge and broker.
    local candidates = CandidateList("broker")
    -- Nothing was sent, so this is not a failed round. Drop the session
    -- instead of parking it: a parked bridge would block every later event
    -- once a broker finally comes online.
    if candidates == nil or #candidates == 0 then bridge = nil return end
    for _, target in ipairs(candidates) do
        if not bridge.tried[target:lower()] then
            bridge.tried[target:lower()] = true
            bridge.attempt = bridge.attempt + 1
            bridge.target, bridge.phase = target, "awaiting_ack"
            local requestId, attempt = bridge.requestId, bridge.attempt
            Queue(TOAST.AddonMessagePriority.CONTROL,
                { "CS1", "B", requestId, TOAST.VERSION }, "WHISPER", target,
                function(status)
                    if status == "target_offline"
                        and not BridgeStale(requestId, attempt)
                        and bridge.phase == "awaiting_ack" then
                        BridgeRetry()
                    end
                end)
            C_Timer.NewTimer(ACK_TIMEOUT, function()
                if not BridgeStale(requestId, attempt) and bridge.phase == "awaiting_ack" then
                    BridgeRetry()
                end
            end)
            return
        end
    end
    bridge.round = bridge.round + 1
    local delay = BROKER_BACKOFF[bridge.round]
    -- Every candidate failed in every round. Persist a terminal rest period so
    -- a later event or a manual refresh cannot skip straight past the backoff.
    if not delay then
        TOAST.SetCrewStateSyncField("nextBridgeAttemptAt", Now() + BRIDGE_EXHAUSTED_COOLDOWN)
        bridge = nil
        return
    end
    bridge.tried = {}
    bridge.phase, bridge.backoffUntil = "backoff", Now() + delay
    C_Timer.NewTimer(delay, function()
        if bridge and bridge.phase == "backoff" then
            bridge.phase = "candidate"
            BridgeTryNext()
        end
    end)
end

local function StartBridge()
    if PlayerCanUseTideGuildTransport() or InCombatLockdown() then
        if PlayerCanUseTideGuildTransport() then
            bridgeBlockedReason, bridgeBlockLogged = nil, false
        end
        return
    end
    if not PlayerMember() then
        bridge = nil
        bridgeBlockedReason = "blocked: not in TOAST roster"
        if not bridgeBlockLogged and TOAST.LogCommsTrace then
            TOAST.LogCommsTrace("LOCAL", "LOCAL", { "CS1", "B" }, "self", bridgeBlockedReason)
            bridgeBlockLogged = true
        end
        return
    end
    bridgeBlockedReason, bridgeBlockLogged = nil, false
    local sync, now = TOAST.GetCrewStateSync and TOAST.GetCrewStateSync() or {}, Now()
    local lastSuccess = tonumber(sync.lastBridgeSuccessAt)
    if lastSuccess and now - lastSuccess < BRIDGE_SUCCESS_COOLDOWN then return end
    local nextAttemptAt = tonumber(sync.nextBridgeAttemptAt)
    if nextAttemptAt and now < nextAttemptAt then return end
    if bridge then return end
    bridge = { requestId = NewRequestId(), round = 0, attempt = 0, tried = {}, phase = "candidate" }
    BridgeTryNext()
end

local function StartHarvest()
    if not PlayerCanUseTideGuildTransport() or InCombatLockdown() then return end
    local candidates = CandidateList("external")
    if not candidates then return end
    for _, target in ipairs(candidates) do
        if not harvest[target:lower()] then
            local name, realm = SenderIdentity(target)
            local identity = TOAST.EncodeCrewIdentity(name, realm)
            if identity then
                local requestId = StableFetchRequestId(identity)
                harvest[target:lower()] = { requestId = requestId, claimedAt = Now() }
                claims[requestId] = claims[requestId]
                    or { kind = "FETCH", target = identity, candidates = {}, createdAt = Now() }
                Queue(TOAST.AddonMessagePriority.CONTROL,
                    { "CS1", "C", requestId, "FETCH", identity }, "GUILD")
                C_Timer.NewTimer(CLAIM_WINDOW, function()
                    -- A guild change can occur while the deterministic claim
                    -- window is open. Do not turn a former Tide client into
                    -- a whisper collector after it has lost the guild route.
                    if not PlayerCanUseTideGuildTransport() then return end
                    local claim = claims[requestId]
                    if not claim then return end
                    local winners = {}
                    for candidate in pairs(claim.candidates) do winners[#winners + 1] = candidate end
                    winners[#winners + 1] = FullName(UnitName("player"), GetRealmName())
                    table.sort(winners, function(a, b) return a:lower() < b:lower() end)
                    if winners[1]:lower() == FullName(UnitName("player"), GetRealmName()):lower() then
                        expectedCache[requestId] = {
                            sender = target, createdAt = Now(), purpose = "FETCH",
                        }
                        Queue(TOAST.AddonMessagePriority.BACKGROUND,
                            { "CS1", "F", requestId, identity }, "WHISPER", target)
                    end
                end)
            end
        end
    end
end

local function HandleSelfState(fields, sender)
    if #fields ~= 8 or not ResolveSender(sender) then return end
    local name, realm = SenderIdentity(sender)
    local version = fields[4]
    local season, score, mapID, level =
        tonumber(fields[5]), tonumber(fields[6]), tonumber(fields[7]), tonumber(fields[8])
    if type(version) ~= "string" or version == "" or #version > 32 then return end
    local key, keyStatus = TOAST.MergeCrewKey(name, realm, mapID, level, "TOAST_DIRECT", Now())
    local peer, scoreStatus = TOAST.MergeCrewScore(name, realm, season, score,
        "TOAST_DIRECT", Now(), CurrentSeason())
    if key or peer then
        TOAST.MarkCrewCapability(name, realm, 1, version)
        if TOAST.ObservePeerVersion then TOAST.ObservePeerVersion(sender, version) end
        if TOAST.MarkPresenceSenderSeen then TOAST.MarkPresenceSenderSeen(sender) end
        if (keyStatus == "updated" or scoreStatus == "updated") and TOAST.RefreshKeyColumn then
            TOAST.RefreshKeyColumn()
        end
    end
end

local function HandleLibKS(message, sender)
    local level, mapID, rating = message:match("^(%d+),(%d+),(%d+)$")
    if not level or not ResolveSender(sender) then return end
    local name, realm = SenderIdentity(sender)
    local key = TOAST.MergeCrewKey(name, realm, tonumber(mapID), tonumber(level), "LIBKS_DIRECT", Now())
    local season = CurrentSeason()
    if season then TOAST.MergeCrewScore(name, realm, season, tonumber(rating), "LIBKS_DIRECT", Now(), season) end
    if key and TOAST.MarkPresenceSenderSeen then TOAST.MarkPresenceSenderSeen(sender) end
end

local function HandleAck(fields, sender)
    if #fields ~= 5 or not ResolveSender(sender) then return end
    local requestId, role = fields[3], fields[4]
    -- The phase check keeps a late or duplicate ACK from restarting cache
    -- traffic that is already running for this bridge.
    if bridge and bridge.requestId == requestId and bridge.phase == "awaiting_ack"
        and bridge.target:lower() == sender:lower() and role == "BROKER" then
        bridge.phase = "transferring"
        local attempt = bridge.attempt
        local name, realm = SenderIdentity(sender)
        TOAST.MarkCrewCapability(name, realm, 1, fields[5])
        if TOAST.ObservePeerVersion then TOAST.ObservePeerVersion(sender, fields[5]) end
        if not SendCachePackage(requestId, sender) then
            BridgeRetry()
            return
        end
        C_Timer.NewTimer(TRANSFER_TTL, function()
            if not BridgeStale(requestId, attempt) and bridge.phase == "transferring" then
                BridgeRetry()
            end
        end)
        return
    end
    if role == "RECEIVED" then outgoing[requestId] = nil end
end

local function ServeBridge(requestId, requester)
    -- A broker asks the guild for fresh SELF_STATE only when the last sweep
    -- is old.  The short delay lets the jittered peer replies merge before
    -- its directed snapshot is built; cache packages are never broadcast.
    RequestGuildState("bridge", true)
    C_Timer.NewTimer(4, function()
        if PlayerCanUseTideGuildTransport() then SendSnapshot(requestId, requester) end
    end)
end

local function HandleClaim(fields, sender)
    if #fields ~= 5 or not PlayerCanUseTideGuildTransport() or not IsTIDE(ResolveSender(sender)) then return end
    local requestId, kind, target = fields[3], fields[4], fields[5]
    if kind ~= "SYNC" and kind ~= "FETCH" then return end
    local claim = claims[requestId]
    if not claim then
        claim = { kind = kind, target = target, candidates = {}, createdAt = Now() }
        claims[requestId] = claim
    end
    local own = FullName(UnitName("player"), GetRealmName())
    if own and sender:lower() == own:lower() then return end -- local GUILD echo; self is added once below
    if claim.kind == kind and claim.target == target then claim.candidates[sender] = true end
end

local function HandleTransferEnd(fields, sender, channel)
    local kind, requestId = fields[3], fields[4]
    local accepted = MergeCompleted(kind, requestId, fields[5], sender, channel)
    if not accepted then return end
    if kind == "SNAP" then
        local wasGuildSync = syncRequests[requestId] ~= nil
        syncRequests[requestId] = nil
        if wasGuildSync then
            TOAST.SetCrewStateSyncField("lastCacheSyncSuccessAt", Now())
            TOAST.SetCrewStateSyncField("lastCacheSyncWeekId", TOAST.GetCrewStateWeekId and TOAST.GetCrewStateWeekId())
        end
        SendAck(requestId, "RECEIVED", sender)
        -- One good bridge ends the automatic cycle for a while, and it has to
        -- survive a /reload, so the stamp goes through crew_state.
        if bridge and bridge.requestId == requestId then
            TOAST.SetCrewStateSyncField("lastBridgeSuccessAt", Now())
            TOAST.SetCrewStateSyncField("nextBridgeAttemptAt", nil)
            bridge = nil
        end
    elseif kind == "CACHE" then
        local purpose = expectedCache[requestId] and expectedCache[requestId].purpose
        expectedCache[requestId] = nil
        SendAck(requestId, "RECEIVED", sender)
        if PlayerCanUseTideGuildTransport() then
            if #accepted > 0 then
                SendTransfer("RELAY", requestId, nil, "GUILD", accepted)
            end
            if purpose == "BRIDGE" then ServeBridge(requestId, sender) end
        end
    end
end

local function HandleProtocol(parsed, channel, sender)
    local fields, kind = parsed.fields, parsed.type
    local senderMember = ResolveSender(sender)
    local own = FullName(UnitName("player"), GetRealmName())
    local localEcho = own and sender:lower() == own:lower()
    if TOAST.LogCommsTrace then
        TOAST.LogCommsTrace("IN", channel, fields, sender,
            localEcho and "local echo" or senderMember and "received" or "ignored unknown sender")
    end
    if not senderMember then return end
    -- WoW delivers our own guild broadcasts back through CHAT_MSG_ADDON.
    -- Their local state has already been written before the send, so they are
    -- useful in the trace but must never participate in a claim, request, or
    -- transfer response.
    if localEcho and channel == "GUILD" then return end
    if kind == "S" and (channel == "GUILD" or channel == "WHISPER") then
        HandleSelfState(fields, sender)
    elseif kind == "H" and #fields == 3 then
        local name, realm = SenderIdentity(sender)
        TOAST.MarkCrewCapability(name, realm, 1, fields[3])
        if TOAST.ObservePeerVersion then TOAST.ObservePeerVersion(sender, fields[3]) end
    elseif kind == "U" and #fields == 5 and (channel == "GUILD" or channel == "WHISPER")
        and IsTIDE(senderMember) then
        if TOAST.ObserveUpdatePolicy then
            TOAST.ObserveUpdatePolicy(sender, fields[3], fields[4], fields[5] == "1")
        end
    elseif kind == "Q" and channel == "GUILD" and #fields == 4 and PlayerCanUseTideGuildTransport() then
        C_Timer.NewTimer(math.random() * 3, function()
            if PlayerCanUseTideGuildTransport() then SendSelfState(fields[3], "GUILD") end
        end)
    elseif kind == "Y" and channel == "GUILD" and #fields == 4 and PlayerCanUseTideGuildTransport() then
        if ResolveSender(sender) == PlayerMember() then return end
        local requestId = fields[3]
        claims[requestId] = claims[requestId]
            or { kind = "SYNC", target = "-", candidates = {}, createdAt = Now() }
        Queue(TOAST.AddonMessagePriority.CONTROL,
            { "CS1", "C", requestId, "SYNC", "-" }, "GUILD")
        C_Timer.NewTimer(CLAIM_WINDOW, function()
            if not PlayerCanUseTideGuildTransport() then return end
            local claim = claims[requestId]
            if not claim then return end
            local candidates = { FullName(UnitName("player"), GetRealmName()) }
            for candidate in pairs(claim.candidates) do candidates[#candidates + 1] = candidate end
            table.sort(candidates, function(a, b) return a:lower() < b:lower() end)
            if candidates[1]:lower() == FullName(UnitName("player"), GetRealmName()):lower() then
                SendSnapshot(requestId, sender)
            end
        end)
    elseif kind == "C" and channel == "GUILD" then
        HandleClaim(fields, sender)
    elseif kind == "B" and channel == "WHISPER" and #fields == 4 and PlayerCanUseTideGuildTransport() then
        expectedCache[fields[3]] = {
            sender = sender, createdAt = Now(), purpose = "BRIDGE",
        }
        SendAck(fields[3], "BROKER", sender)
        SendUpdatePolicy("WHISPER", sender)
    elseif kind == "A" and channel == "WHISPER" then
        HandleAck(fields, sender)
    elseif kind == "F" and channel == "WHISPER" and #fields == 4 and not PlayerCanUseTideGuildTransport() then
        local ownIdentity = TOAST.EncodeCrewIdentity(UnitName("player"), GetRealmName())
        if ownIdentity and ownIdentity == fields[4] then SendCachePackage(fields[3], sender) end
    elseif kind == "X" and #fields == 4 then
        claims[fields[3]], expectedCache[fields[3]], syncRequests[fields[3]] = nil, nil, nil
        if bridge and bridge.requestId == fields[3] then BridgeRetry() end
    elseif kind == "M" and #fields == 5 then
        StartIncoming(fields[3], fields[4], fields[5], sender, channel)
    elseif kind == "D" and #fields == 6 then
        AddChunk(fields[3], fields[4], fields[5], fields[6], sender, channel)
    elseif kind == "E" and #fields == 5 then
        HandleTransferEnd(fields, sender, channel)
    end
end

local function Cleanup()
    local now = Now()
    for key, transfer in pairs(incoming) do
        if now - transfer.startedAt > TRANSFER_TTL then incoming[key] = nil end
    end
    for requestId, claim in pairs(claims) do
        if claim.createdAt and now - claim.createdAt > TRANSFER_TTL then claims[requestId] = nil end
    end
    for requestId, startedAt in pairs(syncRequests) do
        if now - startedAt > TRANSFER_TTL then syncRequests[requestId] = nil end
    end
    for requestId, expected in pairs(expectedCache) do
        if now - expected.createdAt > TRANSFER_TTL then expectedCache[requestId] = nil end
    end
    for requestId, transfer in pairs(outgoing) do
        if now - transfer.sentAt > TRANSFER_TTL then outgoing[requestId] = nil end
    end
    for target, state in pairs(harvest) do
        if now - state.claimedAt > 300 then harvest[target] = nil end
    end
end

function TOAST.RefreshUnknowns()
    if PlayerCanUseTideGuildTransport() then
        RequestCacheSync()
        StartHarvest()
    else
        StartBridge()
    end
end

local function CountTable(tableValue)
    local count = 0
    for _ in pairs(tableValue) do count = count + 1 end
    return count
end

function TOAST.GetCrewCommsStatus()
    local member = PlayerMember()
    local queue = TOAST.GetAddonQueueStatus and TOAST.GetAddonQueueStatus() or {}
    local sync = TOAST.GetCrewStateSync and TOAST.GetCrewStateSync() or {}
    return {
        role = RankKind(member) or "EXTERNAL",
        rosterKnown = member ~= nil,
        tideIdentity = PlayerHasTideIdentity(),
        guildTransport = PlayerCanUseTideGuildTransport(),
        guildName = LocalGuildName(),
        inCommunity = TOAST.inCommunity == true,
        queuePending = queue.pending or 0,
        bridgeTarget = bridge and bridge.target or nil,
        bridgePhase = bridge and bridge.phase or nil,
        bridgeBackoffUntil = bridge and bridge.backoffUntil or nil,
        bridgeRound = bridge and bridge.round or nil,
        bridgeBlockedReason = bridgeBlockedReason,
        bridgeLastSuccessAt = sync.lastBridgeSuccessAt,
        bridgeNextAttemptAt = sync.nextBridgeAttemptAt,
        incomingTransfers = CountTable(incoming),
        outgoingTransfers = CountTable(outgoing),
        activeClaims = CountTable(claims),
        expectedPackages = CountTable(expectedCache),
        harvestTargets = CountTable(harvest),
        lastGuildStateRequestAt = lastGuildStateRequestAt,
        lastCacheSyncAttemptAt = sync.lastCacheSyncAttemptAt,
        lastCacheSyncSuccessAt = sync.lastCacheSyncSuccessAt,
        keyCacheWeekId = sync.keyCacheWeekId,
        lastCacheSyncWeekId = sync.lastCacheSyncWeekId,
        knownKeys = TOAST.GetKnownCrewKeyCount and TOAST.GetKnownCrewKeyCount() or 0,
    }
end

TOAST.CrewCommsDebug = {
    CandidateList = CandidateList,
    RankKind = RankKind,
    PlayerCanUseTideGuildTransport = PlayerCanUseTideGuildTransport,
    PackRecords = PackRecords,
}

local events = CreateFrame("Frame")
events:RegisterEvent("ADDON_LOADED")
events:RegisterEvent("PLAYER_LOGIN")
events:RegisterEvent("PLAYER_GUILD_UPDATE")
events:RegisterEvent("CLUB_STREAMS_LOADED")
events:RegisterEvent("CHALLENGE_MODE_MAPS_UPDATE")
events:RegisterEvent("CHALLENGE_MODE_COMPLETED")
-- Bag-side keystone signals: a new key handed out, a reroll after someone
-- else's dungeon, and a downgrade at the Pandaren NPC all land here before
-- (or without) any challenge-mode event.
events:RegisterEvent("ITEM_CHANGED")
events:RegisterEvent("ITEM_PUSH")
events:RegisterEvent("BAG_UPDATE_DELAYED")
events:RegisterEvent("PLAYER_REGEN_ENABLED")
events:RegisterEvent("CHAT_MSG_ADDON")
events:SetScript("OnEvent", function(_, event, prefix, message, channel, sender)
    if event == "ADDON_LOADED" then
        if prefix ~= addonName then return end
        TOAST_DB = TOAST_DB or {}
        C_Timer.NewTicker(30, Cleanup)
        return
    elseif event == "PLAYER_LOGIN" then
        RecognizePlayer()
        TOAST.inCommunity = TOAST.IsInCommunity()
        C_Timer.NewTimer(math.random(8, 20), function()
            -- Through the observer, so the login broadcast doubles as its
            -- baseline read: the next check only publishes on a real change.
            PublishOwnState()
            if PlayerCanUseTideGuildTransport() then
                SendUpdatePolicy("GUILD")
            else
                StartBridge()
            end
        end)
        RequestGuildState("login")
        C_Timer.NewTimer(math.random(12, 24), function()
            if PlayerCanUseTideGuildTransport() then RequestCacheSync(); StartHarvest() end
        end)
        return
    elseif event == "PLAYER_GUILD_UPDATE" then
        C_Timer.NewTimer(1, TOAST.RefreshUnknowns)
        return
    elseif event == "CLUB_STREAMS_LOADED" then
        TOAST.inCommunity = TOAST.IsInCommunity()
        if PlayerCanUseTideGuildTransport() then StartHarvest() else StartBridge() end
        return
    elseif event == "PLAYER_REGEN_ENABLED" then
        TOAST.RefreshUnknowns()
        if keyObserver.deferred then ScheduleKeyCheck(1) end
        return
    elseif event == "CHALLENGE_MODE_MAPS_UPDATE" then
        ScheduleKeyCheck(KEY_CHECK_DEBOUNCE, { force = true })
        return
    elseif event == "CHALLENGE_MODE_COMPLETED" then
        -- The new key is not necessarily readable yet, so this only arms the
        -- observer; the retry ladder inside it decides when to publish.
        ScheduleKeyCheck(2, { expectChange = true, force = true })
        return
    elseif event == "ITEM_CHANGED" or event == "ITEM_PUSH" or event == "BAG_UPDATE_DELAYED" then
        ScheduleKeyCheck(KEY_CHECK_DEBOUNCE)
        return
    end
    if prefix == "LibKS" then HandleLibKS(message, sender); return end
    if prefix ~= "TOAST" then return end
    local parsed = TOAST.ParseCrewProtocolMessage(message)
    if parsed then
        HandleProtocol(parsed, channel, sender)
    elseif TOAST.LogCommsTrace then
        TOAST.LogCommsTrace("IN", channel, message, sender, "ignored malformed")
    end
end)
