TradeSkillMaster/TradeSkillMaster_AuctionDB/Modules/ChannelSync.lua

-- ------------------------------------------------------------------------------ --
--                           TradeSkillMaster_AuctionDB                           --
--           http://www.curse.com/addons/wow/tradeskillmaster_auctiondb           --
--                                                                                --
--             A TradeSkillMaster Addon (http://tradeskillmaster.com)             --
--    All Rights Reserved* - Detailed license information included with addon.    --
-- ------------------------------------------------------------------------------ --

-- Channel sync for sharing AuctionDB scan data between players.

local TSM = select(2, ...)
local ChannelSync = TSM:NewModule("ChannelSync", "AceEvent-3.0")
local L = LibStub("AceLocale-3.0"):GetLocale("TradeSkillMaster_AuctionDB")

local CHANNEL_NAME = "TSM_AuctionDB"
local COMM_PREFIX = "TSMADB1"
local CHUNK_SIZE = 220
local BUNDLE_TIMEOUT = 45
local SEND_INTERVAL = 0.08
local MAX_TOTAL_CHUNKS = 800
local MAX_ITEMS_PER_PAYLOAD = 250

local private = {
	channelId = nil,
	channelName = nil,
	lastBroadcastHash = nil,
	incoming = {},
	sendQueue = {},
	isSending = false,
	notifiedNewer = {},
	cachedFullScan = {time=nil, itemIDs=nil},
}

local strbyte = string.byte

local libS = LibStub:GetLibrary("AceSerializer-3.0")
local libD = LibStub("LibDeflate", true)
local warnedNoDeflate
local debugEnabled = true

local function GetAddonVersion()
	local version = GetAddOnMetadata("TradeSkillMaster_AuctionDB", "X-Curse-Packaged-Version")
		or GetAddOnMetadata("TradeSkillMaster_AuctionDB", "Version")
	if not version or version == "" then
		return "unknown"
	end
	return version
end

local function VersionKey(version)
	local key = 0
	for part in tostring(version or ""):gmatch("%d+") do
		key = key * 1000 + tonumber(part)
	end
	return key
end

private.addonVersion = GetAddonVersion()
private.addonVersionKey = VersionKey(private.addonVersion)

local function DebugPrint(msg)
	if debugEnabled then
		TSM:Print("ChannelSync: " .. msg)
	end
end

local function CanEncode()
	if libD then return true end
	if not warnedNoDeflate then
		warnedNoDeflate = true
		TSM:Print("AuctionDB channel sync requires LibDeflate. Install or enable an addon that provides it.")
	end
	return false
end

local function HashString(str)
	local hash = 0
	for i = 1, #str do
		hash = (hash * 31 + strbyte(str, i)) % 4294967296
	end
	return hash
end

local function EnsureChannel()
	local channelId = GetChannelName(CHANNEL_NAME)
	if channelId == 0 then
		JoinChannelByName(CHANNEL_NAME)
		channelId = GetChannelName(CHANNEL_NAME)
	end
	private.channelId = channelId > 0 and channelId or nil
	private.channelName = private.channelId and CHANNEL_NAME or nil
end

local function GetChannelDisplayName(channelName, channelString, channelNumber, channelBaseName)
	if channelBaseName and channelBaseName ~= "" and not tonumber(channelBaseName) then
		return channelBaseName
	end
	if channelName and channelName ~= "" and not tonumber(channelName) then
		return channelName
	end
	if channelString and channelString ~= "" then
		local parsed = channelString:match("^%d+%.%s*(.+)$")
		if parsed then return parsed end
		local numeric = tonumber(channelString)
		if numeric then return select(2, GetChannelName(numeric)) end
		return channelString
	end
	if channelNumber then
		return select(2, GetChannelName(channelNumber))
	end
end

local function ChatFilter(_, _, msg, _, channelString, _, channelNumber, channelName, channelBaseName)
	local displayName = GetChannelDisplayName(channelName, channelString, channelNumber, channelBaseName)
	if displayName ~= CHANNEL_NAME then return end
	if strsub(msg, 1, #COMM_PREFIX) == COMM_PREFIX then
		return true
	end
end

function ChannelSync:OnEnable()
	self:RegisterEvent("PLAYER_ENTERING_WORLD", "OnPlayerEnteringWorld")
	self:RegisterEvent("CHAT_MSG_CHANNEL_NOTICE", "OnChannelNotice")
	self:RegisterEvent("CHAT_MSG_CHANNEL", "OnChannelMessage")
	if ChatFrame_AddMessageEventFilter then
		ChatFrame_AddMessageEventFilter("CHAT_MSG_CHANNEL", ChatFilter)
	end
	EnsureChannel()
end

function ChannelSync:OnPlayerEnteringWorld()
	EnsureChannel()
end

function ChannelSync:OnChannelNotice()
	EnsureChannel()
end

local function EncodePayload(payload)
	if not CanEncode() then return end
	local serialized = libS:Serialize(payload)
	if not serialized then
		DebugPrint("Serialize failed.")
		return
	end
	local compressed = libD:CompressDeflate(serialized, { level = 9 })
	if not compressed then
		DebugPrint("CompressDeflate failed.")
		return
	end
	return libD:EncodeForPrint(compressed)
end

local function DecodePayload(encoded)
	if not CanEncode() then return end
	local decoded = libD:DecodeForPrint(encoded)
	if not decoded then
		DebugPrint("DecodeForPrint failed.")
		return
	end
	local decompressed = libD:DecompressDeflate(decoded)
	if not decompressed then
		DebugPrint("DecompressDeflate failed.")
		return
	end
	local ok, payload = libS:Deserialize(decompressed)
	if not ok then
		DebugPrint("Deserialize failed.")
		return
	end
	return payload
end

local function MaybeNotifyNewerVersion(remoteVersion, sender)
	local remoteKey = VersionKey(remoteVersion)
	local localKey = private.addonVersionKey
	if remoteKey == 0 or localKey == 0 or remoteKey <= localKey then return end
	local noticeKey = tostring(sender or "unknown") .. ":" .. tostring(remoteVersion)
	if private.notifiedNewer[noticeKey] then return end
	private.notifiedNewer[noticeKey] = true
	TSM:Printf(L["A newer version of AuctionDB (%s) was received from %s. You are running %s."], remoteVersion, sender or "unknown", private.addonVersion)
end

function ChannelSync:BroadcastScanData(scanType, items)
	if scanType ~= "Full" and scanType ~= "Group" and scanType ~= "Search" then
		return
	end
	if TSM.processingData then
		TSMAPI:CreateTimeDelay("auctionDBChannelSyncBroadcast", 0.5, function()
			ChannelSync:BroadcastScanData(scanType, items)
		end)
		return
	end
	EnsureChannel()
	if not private.channelName or not private.channelId then return end

	local itemIDs = ChannelSync:CollectItemIDs(items)
	if not itemIDs or #itemIDs == 0 then return end

	local function QueueEncoded(encoded)
		local hash = HashString(encoded)
	if hash == private.lastBroadcastHash then return end
	private.lastBroadcastHash = hash
	local total = ceil(#encoded / CHUNK_SIZE)
	if total > MAX_TOTAL_CHUNKS then return false end
		tinsert(private.sendQueue, {hash = hash, encoded = encoded, total = total})
		if not private.isSending then
			private.isSending = true
			TSMAPI.Threading:Start(ChannelSync.SendQueueThread, 0.3)
		end
		return true
	end

	local function QueueBatch(startIndex, endIndex)
		local batch = {}
		for i = startIndex, endIndex do
			tinsert(batch, itemIDs[i])
		end
		local payload = ChannelSync:BuildPayloadFromItemIDs(scanType, batch)
		if not payload then return end
		local encoded = EncodePayload(payload)
		if not encoded then return end
		local total = ceil(#encoded / CHUNK_SIZE)
		if total > MAX_TOTAL_CHUNKS and #batch > 1 then
			local mid = floor((startIndex + endIndex) / 2)
			QueueBatch(startIndex, mid)
			QueueBatch(mid + 1, endIndex)
			return
		end
		if total > MAX_TOTAL_CHUNKS then
			TSM:Print("AuctionDB channel sync payload too large; skipping batch.")
			return
		end
		QueueEncoded(encoded)
	end

	for i = 1, #itemIDs, MAX_ITEMS_PER_PAYLOAD do
		QueueBatch(i, min(i + MAX_ITEMS_PER_PAYLOAD - 1, #itemIDs))
	end
end

function ChannelSync:CollectItemIDs(items)
	local itemIDs, list = {}, {}
	if items then
		if #items > 0 then
			for _, itemString in ipairs(items) do
				local itemID = TSMAPI:GetItemID(itemString)
				if itemID and not itemIDs[itemID] then
					itemIDs[itemID] = true
					tinsert(list, itemID)
				end
			end
		else
			for itemString in pairs(items) do
				local itemID = TSMAPI:GetItemID(itemString)
				if itemID and not itemIDs[itemID] then
					itemIDs[itemID] = true
					tinsert(list, itemID)
				end
			end
		end
	else
		local scanTime = TSM.db.realm.lastCompleteScan
		if private.cachedFullScan.time == scanTime and private.cachedFullScan.itemIDs then
			return private.cachedFullScan.itemIDs
		end
		for itemID in pairs(TSM.data) do
			TSM:DecodeItemData(itemID)
			if TSM.data[itemID].lastScan == scanTime then
				tinsert(list, itemID)
			end
		end
		private.cachedFullScan.time = scanTime
		private.cachedFullScan.itemIDs = list
	end
	return list
end

function ChannelSync:BuildPayloadFromItemIDs(scanType, itemIDList)
	local payloadItems = {}
	for _, itemID in ipairs(itemIDList) do
		local data = TSM.data[itemID]
		if data then
			TSM:EncodeItemData(itemID)
			if data.encoded then
				payloadItems[itemID] = data.encoded
			end
		end
	end
	if not next(payloadItems) then return end

	return {
		v = 1,
		scanType = scanType,
		scanTime = TSM.db.realm.lastCompleteScan,
		version = private.addonVersion,
		items = payloadItems,
	}
end

local function DecodeIncomingItem(itemID, encoded)
	local tmp = {}
	tmp[itemID] = {encoded = encoded}
	TSM:DecodeItemData(itemID, tmp)
	return tmp[itemID]
end

local function MergeIncomingData(payload, sender)
	if type(payload) ~= "table" then return end

	local incoming = {}
	if payload.items then
		incoming = payload.items
	elseif type(payload.scanData) == "string" then
		TSM:Deserialize(payload.scanData, incoming, true)
	else
		return
	end
	local updated = 0
	local lastItemID
	for itemID, data in pairs(incoming) do
		local incomingData = data
		if payload.items then
			incomingData = DecodeIncomingItem(itemID, data)
		end
		if incomingData then
			local existing = TSM.data[itemID]
			if existing then
				TSM:DecodeItemData(itemID)
				if incomingData.lastScan and (not existing.lastScan or incomingData.lastScan > existing.lastScan) then
					TSM.data[itemID] = incomingData
					updated = updated + 1
					lastItemID = itemID
				end
			else
				TSM.data[itemID] = incomingData
				updated = updated + 1
				lastItemID = itemID
			end
		end
	end

	if payload.scanTime and (not TSM.db.realm.lastCompleteScan or payload.scanTime > TSM.db.realm.lastCompleteScan) then
		TSM.db.realm.lastCompleteScan = payload.scanTime
	end

	TSM:Serialize()

	if updated == 0 and sender then
		DebugPrint("No items updated from " .. sender .. ".")
	elseif updated > 0 and sender then
		if updated == 1 and lastItemID then
			local link = select(2, GetItemInfo(lastItemID)) or ("item:" .. tostring(lastItemID))
			TSM:Printf("AuctionDB updated %s from %s.", link, sender)
		else
			TSM:Printf("AuctionDB updated %d items from %s.", updated, sender)
		end
	end
end

function ChannelSync:OnChannelMessage(_, msg, source, _, channelString, _, channelNumber, channelName, channelBaseName)
	local displayName = GetChannelDisplayName(channelName, channelString, channelNumber, channelBaseName)
	if displayName ~= CHANNEL_NAME then
		if strsub(msg or "", 1, #COMM_PREFIX) == COMM_PREFIX then
			DebugPrint("Prefix received; displayName=" .. tostring(displayName) .. " channelNumber=" .. tostring(channelNumber) .. " channelString=" .. tostring(channelString) .. " channelName=" .. tostring(channelName) .. " channelBaseName=" .. tostring(channelBaseName))
		end
		return
	end
	if strsub(msg or "", 1, #COMM_PREFIX) ~= COMM_PREFIX then
		DebugPrint("Message on channel without prefix: '" .. strsub(msg or "", 1, #COMM_PREFIX) .. "'.")
		return
	end

	source = ("-"):split(source or "")
	if strlower(source or "") == strlower(UnitName("player") or "") then return end

	local headerLen = #COMM_PREFIX + 16
	if #msg <= headerLen then return end
	local meta = strsub(msg, #COMM_PREFIX + 1, headerLen)
	local chunk = strsub(msg, headerLen + 1)
	local hashHex = strsub(meta, 1, 8)
	local seqHex = strsub(meta, 9, 12)
	local totalHex = strsub(meta, 13, 16)
	local hash = tonumber(hashHex, 16)
	local seq = tonumber(seqHex, 16)
	local total = tonumber(totalHex, 16)
	if not hash or not seq or not total then return end

	local bundle = private.incoming[hash]
	if not bundle or (time() - bundle.time) > BUNDLE_TIMEOUT then
		if bundle then
			DebugPrint("Bundle timeout " .. format("%08x", hash) .. " (" .. bundle.received .. "/" .. bundle.total .. ").")
		end
		bundle = {time = time(), total = total, chunks = {}, received = 0}
		private.incoming[hash] = bundle
	end

	if bundle.total ~= total then
		DebugPrint("Bundle total mismatch for " .. format("%08x", hash) .. ". Resetting.")
		private.incoming[hash] = {time = time(), total = total, chunks = {}, received = 0}
		bundle = private.incoming[hash]
	end

	if not bundle.chunks[seq] then
		bundle.chunks[seq] = chunk
		bundle.received = bundle.received + 1
	end

	if bundle.received < bundle.total then return end

	local parts = {}
	for i = 1, bundle.total do
		if not bundle.chunks[i] then return end
		tinsert(parts, bundle.chunks[i])
	end
	private.incoming[hash] = nil

	local payload = DecodePayload(table.concat(parts))
	if not payload then
		DebugPrint("Payload decode failed for bundle " .. format("%08x", hash))
		return
	end
	if payload.version then
		MaybeNotifyNewerVersion(payload.version, source)
	end
	MergeIncomingData(payload, source)
end

function ChannelSync:SendQueueThread()
	while #private.sendQueue > 0 do
		local job = private.sendQueue[1]
		for i = 1, job.total do
			local chunk = strsub(job.encoded, (i - 1) * CHUNK_SIZE + 1, i * CHUNK_SIZE)
			local msg = COMM_PREFIX .. string.format("%08x%04x%04x", job.hash, i, job.total) .. chunk
			SendChatMessage(msg, "CHANNEL", nil, private.channelId)
			self:Sleep(SEND_INTERVAL)
		end
		DebugPrint("Sent bundle " .. format("%08x", job.hash) .. " (" .. job.total .. " chunks).")
		tremove(private.sendQueue, 1)
	end
	private.isSending = false
end