// SPDX-License-Identifier: GPL-2.0-only /* * Digital Voice Modem - Converged FNE Software * GPLv2 Open Source. Use is subject to license terms. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * Copyright (C) 2026 DVMProject Authors * */ #include "fne/PatchStatusRegistry.h" #include "common/Log.h" #include #include #include #include #include constexpr uint32_t PatchStatusRegistry::DEFAULT_TTL_SECONDS; constexpr uint32_t PatchStatusRegistry::MIN_TTL_SECONDS; constexpr uint32_t PatchStatusRegistry::MAX_TTL_SECONDS; constexpr uint32_t PatchStatusRegistry::MAX_WAIT_MS; PatchStatusRegistry::PatchStatusRegistry() : m_mutex(), m_revisionChanged(), m_peerPatches(), m_revision(0U), m_defaultTtlSeconds(DEFAULT_TTL_SECONDS), m_minTtlSeconds(MIN_TTL_SECONDS), m_maxTtlSeconds(MAX_TTL_SECONDS) { /* stub */ } void PatchStatusRegistry::configure(uint32_t defaultTtlSeconds, uint32_t minTtlSeconds, uint32_t maxTtlSeconds) { if (minTtlSeconds == 0U) minTtlSeconds = MIN_TTL_SECONDS; if (maxTtlSeconds < minTtlSeconds) maxTtlSeconds = minTtlSeconds; std::lock_guard guard(m_mutex); m_minTtlSeconds = minTtlSeconds; m_maxTtlSeconds = maxTtlSeconds; m_defaultTtlSeconds = std::max(m_minTtlSeconds, std::min(defaultTtlSeconds, m_maxTtlSeconds)); } bool PatchStatusRegistry::publish(json::object& request, json::object& response, std::string& errorMessage) { if (!request["peerId"].is()) { errorMessage = "peerId was not a valid integer"; return false; } if (!request["patches"].is()) { errorMessage = "patches was not a valid array"; return false; } PeerPatchSnapshot incoming; incoming.peerId = request["peerId"].get(); if (incoming.peerId == 0U) { errorMessage = "peerId cannot be zero"; return false; } if (request["peerName"].is()) incoming.peerName = request["peerName"].get(); if (request["sequence"].is()) incoming.sequence = request["sequence"].get(); uint32_t ttlSeconds = defaultTtlSeconds(); if (request["ttlSeconds"].is()) ttlSeconds = request["ttlSeconds"].get(); ttlSeconds = clampTtl(ttlSeconds); incoming.updatedAt = nowMs(); incoming.expiresAt = incoming.updatedAt + (static_cast(ttlSeconds) * 1000U); json::array patches = request["patches"].get(); for (json::value& value : patches) { if (!value.is()) { errorMessage = "patches contained a non-object entry"; return false; } json::object patchObj = value.get(); PatchRecord patch; if (!parsePatch(patchObj, patch, errorMessage)) return false; incoming.patches.push_back(patch); } cleanupExpired(); { std::lock_guard guard(m_mutex); if (incoming.patches.empty()) m_peerPatches.erase(incoming.peerId); else m_peerPatches[incoming.peerId] = incoming; bumpRevisionLocked(); response = snapshotLocked(); response["acceptedPeerId"].set(incoming.peerId); response["ttlSeconds"].set(ttlSeconds); } m_revisionChanged.notify_all(); return true; } bool PatchStatusRegistry::removePeer(uint32_t peerId) { if (peerId == 0U) return false; bool removed = false; { std::lock_guard guard(m_mutex); removed = m_peerPatches.erase(peerId) > 0U; if (removed) bumpRevisionLocked(); } if (removed) m_revisionChanged.notify_all(); return removed; } uint32_t PatchStatusRegistry::cleanupExpired() { uint32_t removed = 0U; bool changed = false; uint64_t now = nowMs(); { std::lock_guard guard(m_mutex); for (auto it = m_peerPatches.begin(); it != m_peerPatches.end();) { if (it->second.expiresAt <= now) { it = m_peerPatches.erase(it); removed++; changed = true; } else { ++it; } } if (changed) bumpRevisionLocked(); } if (changed) m_revisionChanged.notify_all(); return removed; } json::object PatchStatusRegistry::snapshot() { cleanupExpired(); std::lock_guard guard(m_mutex); return snapshotLocked(); } json::object PatchStatusRegistry::waitForChanges(uint64_t sinceRevision, uint32_t waitMs) { waitMs = std::min(waitMs, MAX_WAIT_MS); cleanupExpired(); std::unique_lock lock(m_mutex); if (waitMs > 0U && sinceRevision >= m_revision) { m_revisionChanged.wait_for(lock, std::chrono::milliseconds(waitMs), [&]() { return m_revision > sinceRevision; }); } return snapshotLocked(); } uint64_t PatchStatusRegistry::revision() const { std::lock_guard guard(m_mutex); return m_revision; } uint32_t PatchStatusRegistry::defaultTtlSeconds() const { std::lock_guard guard(m_mutex); return m_defaultTtlSeconds; } uint32_t PatchStatusRegistry::minTtlSeconds() const { std::lock_guard guard(m_mutex); return m_minTtlSeconds; } uint32_t PatchStatusRegistry::maxTtlSeconds() const { std::lock_guard guard(m_mutex); return m_maxTtlSeconds; } uint64_t PatchStatusRegistry::nowMs() { return std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch()).count(); } std::string PatchStatusRegistry::normalizeMode(const std::string& mode) { std::string normalized = mode; std::transform(normalized.begin(), normalized.end(), normalized.begin(), [](unsigned char c) { return static_cast(std::tolower(c)); }); return normalized; } std::string PatchStatusRegistry::buildTalkgroupKey(const PatchMember& member) { std::ostringstream ss; ss << normalizeMode(member.mode) << ':' << member.tgid << ':' << static_cast(member.slot); return ss.str(); } json::object PatchStatusRegistry::memberToJson(const PatchMember& member) { json::object obj = json::object(); obj["system"].set(member.system); obj["mode"].set(member.mode); obj["tgid"].set(member.tgid); obj["slot"].set(member.slot); obj["key"].set(buildTalkgroupKey(member)); return obj; } json::object PatchStatusRegistry::patchToJson(const PatchRecord& patch) { json::object obj = json::object(); obj["patchId"].set(patch.patchId); obj["active"].set(patch.active); obj["oneWay"].set(patch.oneWay); json::array members = json::array(); for (const PatchMember& member : patch.members) members.push_back(json::value(memberToJson(member))); obj["members"].set(members); return obj; } json::object PatchStatusRegistry::peerSnapshotToJson(const PeerPatchSnapshot& peer) { json::object obj = json::object(); obj["peerId"].set(peer.peerId); obj["peerName"].set(peer.peerName); obj["sequence"].set(peer.sequence); obj["updatedAt"].set(peer.updatedAt); obj["expiresAt"].set(peer.expiresAt); json::array patches = json::array(); for (const PatchRecord& patch : peer.patches) patches.push_back(json::value(patchToJson(patch))); obj["patches"].set(patches); return obj; } bool PatchStatusRegistry::parsePatch(json::object& obj, PatchRecord& patch, std::string& errorMessage) const { if (obj["patchId"].is()) patch.patchId = obj["patchId"].get(); if (obj["active"].is()) patch.active = obj["active"].get(); if (obj["oneWay"].is()) patch.oneWay = obj["oneWay"].get(); if (!obj["members"].is()) { errorMessage = "patch members was not a valid array"; return false; } json::array members = obj["members"].get(); for (json::value& value : members) { if (!value.is()) { errorMessage = "patch members contained a non-object entry"; return false; } json::object memberObj = value.get(); PatchMember member; if (!parseMember(memberObj, member, errorMessage)) return false; patch.members.push_back(member); } return true; } bool PatchStatusRegistry::parseMember(json::object& obj, PatchMember& member, std::string& errorMessage) const { if (obj["system"].is()) member.system = obj["system"].get(); if (obj["mode"].is()) member.mode = normalizeMode(obj["mode"].get()); else member.mode = "unknown"; if (!obj["tgid"].is()) { errorMessage = "patch member tgid was not a valid integer"; return false; } member.tgid = obj["tgid"].get(); if (member.tgid == 0U) { errorMessage = "patch member tgid cannot be zero"; return false; } if (obj["slot"].is()) member.slot = obj["slot"].get(); else if (obj["slot"].is()) { uint32_t slot = obj["slot"].get(); if (slot > std::numeric_limits::max()) { errorMessage = "patch member slot was out of range"; return false; } member.slot = static_cast(slot); } return true; } uint32_t PatchStatusRegistry::clampTtl(uint32_t ttlSeconds) const { std::lock_guard guard(m_mutex); return std::max(m_minTtlSeconds, std::min(ttlSeconds, m_maxTtlSeconds)); } json::object PatchStatusRegistry::snapshotLocked() const { json::object response = json::object(); response["revision"].set(m_revision); json::array peers = json::array(); json::array patches = json::array(); json::object byTalkgroup = json::object(); for (const auto& entry : m_peerPatches) { const PeerPatchSnapshot& peer = entry.second; peers.push_back(json::value(peerSnapshotToJson(peer))); for (const PatchRecord& patch : peer.patches) { json::object patchObj = patchToJson(patch); patchObj["peerId"].set(peer.peerId); patchObj["peerName"].set(peer.peerName); patchObj["updatedAt"].set(peer.updatedAt); patchObj["expiresAt"].set(peer.expiresAt); patches.push_back(json::value(patchObj)); for (const PatchMember& member : patch.members) { std::string key = buildTalkgroupKey(member); json::array entries = json::array(); if (byTalkgroup[key].is()) entries = byTalkgroup[key].get(); json::object tgPatch = json::object(); tgPatch["peerId"].set(peer.peerId); tgPatch["peerName"].set(peer.peerName); tgPatch["patchId"].set(patch.patchId); tgPatch["active"].set(patch.active); tgPatch["oneWay"].set(patch.oneWay); tgPatch["updatedAt"].set(peer.updatedAt); tgPatch["expiresAt"].set(peer.expiresAt); tgPatch["member"].set(memberToJson(member)); entries.push_back(json::value(tgPatch)); byTalkgroup[key].set(entries); } } } response["peers"].set(peers); response["patches"].set(patches); response["byTalkgroup"].set(byTalkgroup); return response; } void PatchStatusRegistry::bumpRevisionLocked() { m_revision++; if (m_revision == 0U) m_revision = 1U; }