// 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) 2023-2026 Bryan Biedenkapp, N2PLL
 *
 */
#include "fne/Defines.h"
#include "common/dmr/lc/csbk/CSBKFactory.h"
#include "common/dmr/lc/LC.h"
#include "common/dmr/lc/FullLC.h"
#include "common/dmr/SlotType.h"
#include "common/dmr/Sync.h"
#include "common/Clock.h"
#include "common/Log.h"
#include "common/Utils.h"
#include "lookups/AffiliationLookup.h"
#include "network/TrafficNetwork.h"
#include "network/callhandler/TagDMRData.h"
#include "HostFNE.h"
#include "FNEMain.h"

using namespace system_clock;
using namespace network;
using namespace network::callhandler;
using namespace network::callhandler::packetdata;
using namespace dmr;
using namespace dmr::defines;

#include <cassert>
#include <chrono>

// ---------------------------------------------------------------------------
//  Public Class Members
// ---------------------------------------------------------------------------

/* Initializes a new instance of the TagDMRData class. */

TagDMRData::TagDMRData(TrafficNetwork* network, bool debug) :
    m_network(network),
    m_parrotFrames(),
    m_parrotFramesReady(false),
    m_parrotPlayback(false),
    m_lastParrotPeerId(0U),
    m_lastParrotSrcId(0U),
    m_lastParrotDstId(0U),
    m_status(),
    m_statusPVCall(),
    m_rejectedCallStreams(),
    m_debug(debug)
{
    assert(network != nullptr);

    m_packetData = new DMRPacketData(network, this, debug);
}

/* Finalizes a instance of the TagDMRData class. */

TagDMRData::~TagDMRData()
{
    delete m_packetData;
}

/* Process a data frame from the network. */

bool TagDMRData::processFrame(const uint8_t* data, uint32_t len, uint32_t peerId, uint32_t ssrc, uint16_t pktSeq, uint32_t streamId, bool fromUpstream)
{
    hrc::hrc_t pktTime = hrc::now();

    DECLARE_UINT8_ARRAY(buffer, len);
    ::memcpy(buffer, data, len);

    uint8_t seqNo = data[4U];

    uint32_t srcId = GET_UINT24(data, 5U);
    uint32_t dstId = GET_UINT24(data, 8U);

    FLCO::E flco = (data[15U] & 0x40U) == 0x40U ? FLCO::PRIVATE : FLCO::GROUP;

    uint32_t slotNo = (data[15U] & 0x80U) == 0x80U ? 2U : 1U;

    DataType::E dataType = (DataType::E)(data[15U] & 0x0FU);

    data::NetData dmrData;
    dmrData.setSeqNo(seqNo);
    dmrData.setSlotNo(slotNo);
    dmrData.setSrcId(srcId);
    dmrData.setDstId(dstId);
    dmrData.setFLCO(flco);

    bool dataSync = (data[15U] & 0x20U) == 0x20U;
    bool voiceSync = (data[15U] & 0x10U) == 0x10U;

    if (dataSync) {
        dmrData.setData(data + 20U);
        dmrData.setDataType(dataType);
        dmrData.setN(0U);
    }
    else if (voiceSync) {
        dmrData.setData(data + 20U);
        dmrData.setDataType(DataType::VOICE_SYNC);
        dmrData.setN(0U);
    }
    else {
        uint8_t n = data[15U] & 0x0FU;
        dmrData.setData(data + 20U);
        dmrData.setDataType(DataType::VOICE);
        dmrData.setN(n);
    }

    if (dataSync && ((dataType == DataType::DATA_HEADER)    ||
                     (dataType == DataType::RATE_12_DATA)   ||
                     (dataType == DataType::RATE_34_DATA)   ||
                     (dataType == DataType::RATE_1_DATA))) {
        if (m_network->m_disablePacketData)
            return false;
        return m_packetData->processFrame(data, len, peerId, pktSeq, streamId, fromUpstream);
    }

    uint8_t frame[DMR_FRAME_LENGTH_BYTES];
    dmrData.getData(frame);

    // process a CSBK out into a class literal if possible
    std::unique_ptr<lc::CSBK> csbk;
    if (dataSync && (dataType == DataType::CSBK)) {
        csbk = lc::csbk::CSBKFactory::createCSBK(frame, dataType);
    }

    // perform TGID route rewrites if configured
    routeRewrite(buffer, peerId, dmrData, dataType, dstId, slotNo, false);
    dstId = GET_UINT24(buffer, 8U);

    // is the stream valid?
    if (validate(peerId, dmrData, csbk.get(), streamId)) {
        // is this peer ignored?
        if (!isPeerPermitted(peerId, dmrData, streamId, fromUpstream)) {
            return false;
        }

        // is this the end of the call stream?
        if (dataSync && (dataType == DataType::TERMINATOR_WITH_LC)) {
            if (srcId == 0U && dstId == 0U) {
                LogWarning(LOG_NET, "DMR, invalid TERMINATOR, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slot = %u, streamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, slotNo, streamId, fromUpstream);
                return false;
            }

            RxStatus status;
            {
                auto it = std::find_if(m_status.begin(), m_status.end(), [&](StatusMapPair& x) {
                    if (x.second.dstId == dstId && x.second.slotNo == slotNo) {
                        return true;
                    }
                    return false;
                });
                if (it == m_status.end()) {
                    LogError(LOG_NET, "DMR, tried to end call for non-existent call in progress?, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slot = %u, streamId = %u, fromUpstream = %u",
                        peerId, ssrc, srcId, dstId, slotNo, streamId, fromUpstream);
                }
                else {
                    status = it->second;
                }
            }

            uint64_t duration = hrc::diff(pktTime, status.callStartTime);

            auto it = std::find_if(m_status.begin(), m_status.end(), [&](StatusMapPair& x) {
                if (x.second.dstId == dstId && x.second.slotNo == slotNo) {
                    if (x.second.activeCall)
                        return true;
                }
                return false;
            });
            if (it != m_status.end()) {
                m_status[dstId].reset();

                // is this a parrot talkgroup? if so, clear any remaining frames from the buffer
                lookups::TalkgroupRuleGroupVoice tg = m_network->m_tidLookup->find(dstId);
                if (tg.config().parrot() && !m_parrotPlayback) {
                    if (m_parrotFrames.size() > 0) {
                        m_parrotFramesReady = true;
                        LogInfoEx(LOG_NET, "DMR, Parrot Playback will Start, peer = %u, ssrc = %u, srcId = %u", peerId, ssrc, srcId);
                        m_network->m_parrotDelayTimer.start();
                    }
                }

                // is this a private call?
                auto it = std::find_if(m_statusPVCall.begin(), m_statusPVCall.end(), [&](StatusMapPair& x) {
                    if (x.second.dstId == dstId) {
                        if (x.second.activeCall)
                            return true;
                    }
                    return false;
                });
                if (it != m_statusPVCall.end()) {
                    m_statusPVCall[dstId].reset();

                    m_network->m_globalAff->releaseGrant(dstId);

                    #define PRV_CALL_END_LOG "DMR, Private Call End, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slot = %u, duration = %u, streamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, slotNo, duration / 1000, streamId, fromUpstream
                    if (m_network->m_logUpstreamCallStartEnd && fromUpstream)
                        LogInfoEx(LOG_PEER, PRV_CALL_END_LOG);
                    else if (!fromUpstream)
                        LogInfoEx(LOG_MASTER, PRV_CALL_END_LOG);
                }
                else {
                    m_network->m_globalAff->releaseGrant(dstId);

                    #define CALL_END_LOG "DMR, Call End, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slot = %u, duration = %u, streamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, slotNo, duration / 1000, streamId, fromUpstream
                    if (m_network->m_logUpstreamCallStartEnd && fromUpstream)
                        LogInfoEx(LOG_PEER, CALL_END_LOG);
                    else if (!fromUpstream)
                        LogInfoEx(LOG_MASTER, CALL_END_LOG);
                }

                // clear any rejected call streams for this TG
                m_rejectedCallStreams.lock(false);
                m_rejectedCallStreams[dstId].clear();
                m_rejectedCallStreams.unlock();

                if (!tg.config().parrot()) {
                    m_network->m_totalActiveCalls--;
                    if (m_network->m_totalActiveCalls < 0)
                        m_network->m_totalActiveCalls = 0;
                }

                // report call event to InfluxDB
                if (m_network->m_enableInfluxDB) {
                    influxdb::QueryBuilder()
                        .meas("call_event")
                            .tag("peerId", std::to_string(peerId))
                            .tag("mode", "DMR")
                            .tag("streamId", std::to_string(streamId))
                            .tag("srcId", std::to_string(srcId))
                            .tag("dstId", std::to_string(dstId))
                                .field("duration", duration)
                                .field("slot", slotNo)
                            .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                        .requestAsync(m_network->m_influxServer);
                }

                m_network->eraseStreamPktSeq(peerId, streamId);
            } else {
                #define NONCALL_END_LOG "DMR, Non-Call Terminator, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slot = %u, streamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, slotNo, streamId, fromUpstream
                if (m_network->m_logUpstreamCallStartEnd && fromUpstream)
                    LogInfoEx(LOG_PEER, NONCALL_END_LOG);
                else if (!fromUpstream)
                    LogInfoEx(LOG_MASTER, NONCALL_END_LOG);

                m_status.lock(false);
                m_status[dstId].callStartTime = pktTime; // because Non-Call Terminators can just happen lets reset the callStartTime to pktTime to prevent insane durations
                m_status.unlock();
            }
        }

        // is this a new call stream?
        if (dataSync && (dataType == DataType::VOICE_LC_HEADER)) {
            if (srcId == 0U && dstId == 0U) {
                LogWarning(LOG_NET, "DMR, invalid call, peer = %u, ssrc = %u, srcId = %u, dstId = %u, streamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, streamId, fromUpstream);
                return false;
            }

            bool switchOver = (data[14U] & network::NET_CTRL_SWITCH_OVER) == network::NET_CTRL_SWITCH_OVER;

            auto it = std::find_if(m_status.begin(), m_status.end(), [&](StatusMapPair& x) {
                if (x.second.dstId == dstId && x.second.slotNo == slotNo) {
                    if (x.second.activeCall)
                        return true;
                }
                return false;
            });
            if (it != m_status.end()) {
                RxStatus status = it->second;

                // is the call being taken over?
                if (status.callTakeover) {
                    LogInfoEx((fromUpstream) ? LOG_PEER : LOG_MASTER, "DMR, Call Source Switched (Takeover), peer = %u, ssrc = %u, srcId = %u, dstId = %u, slotNo = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxSlotNo = %u, rxStreamId = %u, fromUpstream = %u",
                        peerId, ssrc, srcId, dstId, slotNo, streamId, status.peerId, status.srcId, status.dstId, status.slotNo, status.streamId, fromUpstream);

                    m_status.lock(false);
                    m_status[dstId].streamId = streamId;
                    m_status[dstId].srcId = srcId;
                    m_status[dstId].ssrc = ssrc;
                    m_status[dstId].callTakeover = false; // reset takeover flag
                    m_status.unlock();

                    status = m_status[dstId];
                }

                if (streamId != status.streamId) {
                    // perform TG switch over -- this can happen in special conditions where a TG may rapidly switch
                    // from one source to another (primarily from bridge resources)
                    if (switchOver && status.slotNo == slotNo) {
                        m_status.lock(false);
                        m_status[dstId].streamId = streamId;
                        m_status[dstId].ssrc = ssrc;
                        if (status.srcId == 0U)
                            m_status[dstId].srcId = srcId;
                        if (status.srcId != srcId) {
                            LogInfoEx((fromUpstream) ? LOG_PEER : LOG_MASTER, "DMR, Call Source Switched, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slotNo = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxSlotNo = %u, rxStreamId = %u, fromUpstream = %u",
                                peerId, ssrc, srcId, dstId, slotNo, streamId, status.peerId, status.srcId, status.dstId, status.slotNo, status.streamId, fromUpstream);
                            m_status[dstId].srcId = srcId;
                        }
                        m_status.unlock();
                    }
                    else {
                        if (status.srcId != 0U && status.srcId != srcId) {
                            bool hasCallPriority = false;

                            // determine if the peer trying to transmit has call priority
                            if (m_network->m_callCollisionTimeout > 0U) {
                                m_network->m_peers.shared_lock();
                                for (auto peer : m_network->m_peers) {
                                    if (peerId == peer.first) {
                                        FNEPeerConnection* conn = peer.second;
                                        if (conn != nullptr) {
                                            hasCallPriority = conn->hasCallPriority();
                                            break;
                                        }
                                    }
                                }
                                m_network->m_peers.shared_unlock();
                            }

                            // perform standard call collision if the call collision timeout is set *and*
                            //  the peer doesn't have call priority
                            if (m_network->m_callCollisionTimeout > 0U && !hasCallPriority) {
                                uint64_t lastPktDuration = hrc::diff(hrc::now(), status.lastPacket);
                                if ((lastPktDuration / 1000) > m_network->m_callCollisionTimeout) {
                                    LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "DMR, Call Collision, lasted more then %us with no further updates, resetting call source", m_network->m_callCollisionTimeout);

                                    m_status.lock(false);
                                    m_status[dstId].streamId = streamId;
                                    m_status[dstId].srcId = srcId;
                                    m_status[dstId].ssrc = ssrc;
                                    m_status.unlock();

                                    // because the call stream source has reset, clear any rejected call streams for 
                                    // this TG to allow the new source to transmit
                                    m_rejectedCallStreams.lock(false);
                                    m_rejectedCallStreams[dstId].clear();
                                    m_rejectedCallStreams.unlock();
                                } else {
                                    LogWarning((fromUpstream) ? LOG_PEER : LOG_MASTER, "DMR, Call Collision, peer = %u, ssrc = %u, srcId = %u, dstId = %u, slotNo = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxSlotNo = %u, rxStreamId = %u, fromUpstream = %u",
                                        peerId, ssrc, srcId, dstId, slotNo, streamId, status.peerId, status.srcId, status.dstId, status.slotNo, status.streamId, fromUpstream);

                                    m_rejectedCallStreams.lock(false);
                                    m_rejectedCallStreams[dstId].push_back(streamId);
                                    m_rejectedCallStreams.unlock();

                                    m_network->m_totalCallCollisions++;

                                    return false;
                                }
                            } else {
                                if (hasCallPriority && !m_network->m_disallowInCallCtrl) {
                                    LogInfoEx((fromUpstream) ? LOG_PEER : LOG_MASTER, "DMR, Call Source Switched (Priority), peer = %u, ssrc = %u, srcId = %u, dstId = %u, slotNo = %u, streamId = %u, rxPeer = %u, rxSrcId = %u, rxDstId = %u, rxSlotNo = %u, rxStreamId = %u, fromUpstream = %u",
                                        peerId, ssrc, srcId, dstId, slotNo, streamId, status.peerId, status.srcId, status.dstId, status.slotNo, status.streamId, fromUpstream);

                                    // since we're gonna switch over the stream and interrupt the current call inprogress lets try to ICC the transmitting peer
                                    if (m_network->isPeerLocal(m_status[dstId].ssrc))
                                        m_network->writePeerICC(m_status[dstId].peerId, m_status[dstId].streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, dstId, 0U, true, false,
                                            m_status[dstId].ssrc);
                                    else
                                        m_network->writePeerICC(m_status[dstId].peerId, m_status[dstId].streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, dstId, 0U, true, true,
                                            m_status[dstId].ssrc);
                                }

                                m_status.lock(false);
                                m_status[dstId].streamId = streamId;
                                m_status[dstId].srcId = srcId;
                                m_status[dstId].ssrc = ssrc;
                                m_status.unlock();
                            }
                        }
                    }
                }
            }
            else {
                // is this a parrot talkgroup? if so, clear any remaining frames from the buffer
                lookups::TalkgroupRuleGroupVoice tg = m_network->m_tidLookup->find(dstId);
                if (tg.config().parrot() && !m_parrotPlayback) {
                    m_parrotFramesReady = false;
                    if (m_parrotFrames.size() > 0) {
                        m_parrotFrames.lock(false);
                        for (auto& pkt : m_parrotFrames) {
                            if (pkt.buffer != nullptr) {
                                delete[] pkt.buffer;
                            }
                        }
                        m_parrotFrames.unlock();
                        m_parrotFrames.clear();
                    }
                }

                // this is a new call stream
                // bryanb: this could be problematic and is naive, if a dstId appears on both slots (which shouldn't happen)
                m_status.lock(false);
                m_status[dstId].callStartTime = pktTime;
                m_status[dstId].srcId = srcId;
                m_status[dstId].dstId = dstId;
                m_status[dstId].slotNo = slotNo;
                m_status[dstId].streamId = streamId;
                m_status[dstId].peerId = peerId;
                m_status[dstId].ssrc = ssrc;
                m_status[dstId].activeCall = true;
                m_status.unlock();

                // clear any rejected call streams for this TG
                m_rejectedCallStreams.lock(false);
                m_rejectedCallStreams[dstId].clear();
                m_rejectedCallStreams.unlock();

                if (!tg.config().parrot()) {
                    m_network->m_totalCallsProcessed++;
                    m_network->m_totalActiveCalls++;
                }

                // is this a private call?
                if (flco == FLCO::PRIVATE) {
                    m_statusPVCall.lock(false);
                    m_statusPVCall[dstId].callStartTime = pktTime;
                    m_statusPVCall[dstId].srcId = srcId;
                    m_statusPVCall[dstId].dstId = dstId;
                    m_statusPVCall[dstId].slotNo = slotNo;
                    m_statusPVCall[dstId].streamId = streamId;
                    m_statusPVCall[dstId].peerId = peerId;
                    m_statusPVCall[dstId].ssrc = ssrc;
                    m_statusPVCall[dstId].activeCall = true;

                    // find the SSRC of the peer that registered this unit
                    uint32_t regSSRC = m_network->findPeerUnitReg(srcId);
                    m_statusPVCall[dstId].dstPeerId = regSSRC;
                    m_statusPVCall.unlock();

                    if (!m_network->m_globalAff->isGranted(dstId))
                        m_network->m_globalAff->grantCh(dstId, srcId, ssrc, fne_lookups::GRANT_TIMER_TIMEOUT, false);

                    #define PRV_CALL_START_LOG "DMR, Private Call Start, peer = %u, ssrc = %u, srcId = %u, dstId = %u, streamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, streamId, fromUpstream
                    if (m_network->m_logUpstreamCallStartEnd && fromUpstream)
                        LogInfoEx(LOG_PEER, PRV_CALL_START_LOG);
                    else if (!fromUpstream)
                        LogInfoEx(LOG_MASTER, PRV_CALL_START_LOG);
                }
                else {
                    if (!m_network->m_globalAff->isGranted(dstId))
                        m_network->m_globalAff->grantCh(dstId, srcId, ssrc, fne_lookups::GRANT_TIMER_TIMEOUT, true);

                    #define CALL_START_LOG "DMR, Call Start, peer = %u, ssrc = %u, srcId = %u, dstId = %u, streamId = %u, fromUpstream = %u", peerId, ssrc, srcId, dstId, streamId, fromUpstream
                    if (m_network->m_logUpstreamCallStartEnd && fromUpstream)
                        LogInfoEx(LOG_PEER, CALL_START_LOG);
                    else if (!fromUpstream)
                        LogInfoEx(LOG_MASTER, CALL_START_LOG);
                }
            }
        }

        // is this a parrot talkgroup?
        lookups::TalkgroupRuleGroupVoice tg = m_network->m_tidLookup->find(dstId);
        if (tg.config().parrot()) {
            uint8_t* copy = new uint8_t[len];
            ::memcpy(copy, buffer, len);

            ParrotFrame parrotFrame = ParrotFrame();
            parrotFrame.buffer = copy;
            parrotFrame.bufferLen = len;

            parrotFrame.slotNo = slotNo;

            parrotFrame.pktSeq = pktSeq;
            parrotFrame.streamId = streamId;
            parrotFrame.peerId = peerId;

            parrotFrame.srcId = srcId;
            parrotFrame.dstId = dstId;

            m_parrotFrames.push_back(parrotFrame);

            if (m_network->m_parrotOnlyOriginating) {
                return true; // end here because parrot calls should never repeat anywhere
            }
        }

        // process CSBK from peer
        if (!processCSBK(buffer, peerId, dmrData)) {
            return false;
        }

        m_status.lock(false);
        m_status[dstId].lastPacket = hrc::now();
        m_status.unlock();

        bool noConnectedPeerRepeat = false;
        bool privateCallInProgress = false;

        // is this a private call in-progress?
        if (m_network->m_restrictPVCallToRegOnly) {
            if (flco == FLCO::PRIVATE) {
                privateCallInProgress = true;
            }

            if (privateCallInProgress) {
                // if we've not determined the destination peer, we have to repeat it everywhere
                if (m_statusPVCall[dstId].dstPeerId == 0U) {
                    noConnectedPeerRepeat = false;
                    privateCallInProgress = false; // trick the system to repeat everywhere
                } else {
                    // if this is a private call, check if the destination peer is one directly connected to us, if not
                    // flag the call so it only repeats to upstream neighbor peers
                    if (m_network->m_peers.size() > 0U && !noConnectedPeerRepeat) {
                        noConnectedPeerRepeat = true;
                        for (auto peer : m_network->m_peers) {
                            if (peerId != peer.first) {
                                FNEPeerConnection* conn = peer.second;
                                if (conn != nullptr) {
                                    if (conn->peerClass() == PEER_CONN_CLASS_NEIGHBOR) {
                                        continue;
                                    }
                                }

                                if (m_statusPVCall[dstId].dstPeerId == peer.first) {
                                    noConnectedPeerRepeat = false;
                                    break;
                                }
                            }
                        }
                    }
                }
            }
        }

        /*
        ** MASTER TRAFFIC
        */

        // repeat traffic to nodes peered to us as master
        if (m_network->m_peers.size() > 0U && !noConnectedPeerRepeat) {
            uint32_t i = 0U;
            udp::BufferQueue queue = udp::BufferQueue();

            m_network->m_peers.shared_lock();
            for (auto peer : m_network->m_peers) {
                if (peer.second == nullptr)
                    continue;
                if (peerId != peer.first) {
                    FNEPeerConnection* conn = peer.second;
                    if (ssrc == peer.first) {
                        // skip the peer if it is the source peer
                        continue;
                    }

                    if (m_network->m_restrictPVCallToRegOnly) {
                        // is this peer an upstream neighbor peer?
                        bool neighbor = false;
                        if (conn != nullptr) {
                            neighbor = conn->peerClass() == PEER_CONN_CLASS_NEIGHBOR;
                        }

                        // is this a private call?
                        if ((flco == FLCO::PRIVATE) && !neighbor) {
                            // is this a private call? if so only repeat to the peer that registered the unit
                            auto it = std::find_if(m_statusPVCall.begin(), m_statusPVCall.end(), [&](StatusMapPair& x) {
                                if (x.second.dstId == dstId) {
                                    if (x.second.activeCall)
                                        return true;
                                }
                                return false;
                            });
                            if (it != m_statusPVCall.end()) {
                                if (peer.first != m_statusPVCall[dstId].dstPeerId) {
                                    continue;
                                }
                            }
                        }
                    }

                    // is this peer ignored?
                    if (!isPeerPermitted(peer.first, dmrData, streamId)) {
                        continue;
                    }

                    // every MAX_QUEUED_PEER_MSGS peers flush the queue
                    if (i % MAX_QUEUED_PEER_MSGS == 0U) {
                        m_network->m_frameQueue->flushQueue(&queue);
                    }

                    DECLARE_UINT8_ARRAY(outboundPeerBuffer, len);
                    ::memcpy(outboundPeerBuffer, buffer, len);

                    // perform TGID route rewrites if configured
                    routeRewrite(outboundPeerBuffer, peer.first, dmrData, dataType, dstId, slotNo);

                    m_network->writePeerQueue(&queue, peer.first, ssrc, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, outboundPeerBuffer, len, pktSeq, streamId);
                    if (m_network->m_debug) {
                        LogDebugEx(LOG_DMR, "TagDMRData::processFrame()", "Master, ssrc = %u, srcPeer = %u, dstPeer = %u, seqNo = %u, srcId = %u, dstId = %u, flco = $%02X, slotNo = %u, len = %u, pktSeq = %u, stream = %u, fromUpstream = %u", 
                            ssrc, peerId, peer.first, seqNo, srcId, dstId, flco, slotNo, len, pktSeq, streamId, fromUpstream);
                    }

                    i++;
                }
            }
            m_network->m_frameQueue->flushQueue(&queue);
            m_network->m_peers.shared_unlock();
        }

        // if this is a private call, and we have already repeated to the connected peer that registered
        // the unit, don't repeat to any neighbor FNE peers
        if (privateCallInProgress && !noConnectedPeerRepeat) {
            return true;
        }

        /*
        ** PEER TRAFFIC (e.g. upstream networks this FNE is peered to)
        */

        // repeat traffic to master nodes we have connected to as a peer
        if (m_network->m_host->m_peerNetworks.size() > 0U && !tg.config().parrot()) {
            for (auto peer : m_network->m_host->m_peerNetworks) {
                uint32_t dstPeerId = peer.second->getPeerId();

                // don't try to repeat traffic to the source peer...if this traffic
                // is coming from a neighbor FNE peer
                if (dstPeerId != peerId) {
                    if (ssrc == dstPeerId) {
                        // skip the peer if it is the source peer
                        continue;
                    }

                    // skip peer if it isn't enabled
                    if (!peer.second->isEnabled()) {
                        continue;
                    }

                    // is this peer ignored?
                    if (!isPeerPermitted(dstPeerId, dmrData, streamId, true)) {
                        continue;
                    }

                    DECLARE_UINT8_ARRAY(outboundPeerBuffer, len);
                    ::memcpy(outboundPeerBuffer, buffer, len);

                    // perform TGID route rewrites if configured
                    routeRewrite(outboundPeerBuffer, dstPeerId, dmrData, dataType, dstId, slotNo);

                    // are we a replica peer?
                    if (peer.second->isReplica())
                        peer.second->writeMaster({ NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, outboundPeerBuffer, len, pktSeq, streamId, false, 0U, ssrc);
                    else
                        peer.second->writeMaster({ NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, outboundPeerBuffer, len, pktSeq, streamId);
                    if (m_network->m_debug) {
                        LogDebugEx(LOG_DMR, "TagDMRData::processFrame()", "Peers, ssrc = %u, srcPeer = %u, dstPeer = %u, seqNo = %u, srcId = %u, dstId = %u, flco = $%02X, slotNo = %u, len = %u, pktSeq = %u, stream = %u, fromUpstream = %u", 
                            ssrc, peerId, dstPeerId, seqNo, srcId, dstId, flco, slotNo, len, pktSeq, streamId, fromUpstream);
                    }
                }
            }
        }

        return true;
    }

    return false;
}

/* Process a grant request frame from the network. */

bool TagDMRData::processGrantReq(uint32_t srcId, uint32_t dstId, uint8_t slot, bool unitToUnit, uint32_t peerId, uint16_t pktSeq, uint32_t streamId)
{
    // if we have an Rx status for the destination deny the grant
    auto it = std::find_if(m_status.begin(), m_status.end(), [&](StatusMapPair& x) {
        if (x.second.dstId == dstId/* && x.second.slotNo == slot*/) {
            if (x.second.activeCall)
                return true;
        }
        return false;
    });
    if (it != m_status.end()) {
        return false;
    }

    // is the source ID a blacklisted ID?
    lookups::RadioId rid = m_network->m_ridLookup->find(srcId);
    if (!rid.radioDefault()) {
        if (!rid.radioEnabled()) {
            return false;
        }
    }

    lookups::TalkgroupRuleGroupVoice tg = m_network->m_tidLookup->find(dstId);

    // check TGID validity
    if (tg.isInvalid()) {
        return false;
    }

    if (!tg.config().active()) {
        return false;
    }

    // repeat traffic to the connected peers
    if (m_network->m_peers.size() > 0U) {
        for (auto peer : m_network->m_peers) {
            if (peerId != peer.first) {
                write_CSBK_Grant(peer.first, srcId, dstId, 4U, !unitToUnit);
            }
        }
    }

    if (!m_network->m_globalAff->isGranted(dstId)) {
        if (m_network->m_globalAff->grantCh(dstId, srcId, peerId, fne_lookups::GRANT_TIMER_TIMEOUT, !unitToUnit)) {
            return true;
        }
    }

    return false;
}

/* Helper to trigger a call takeover from a In-Call control event. */

void TagDMRData::triggerCallTakeover(uint32_t dstId)
{
    auto it = std::find_if(m_status.begin(), m_status.end(), [&](StatusMapPair& x) {
        if (x.second.dstId == dstId) {
            if (x.second.activeCall)
                return true;
        }
        return false;
    });
    if (it != m_status.end()) {
        m_status.lock(false);
        m_status[dstId].callTakeover = true;
        m_status.unlock();
    }
}

/* Helper to playback a parrot frame to the network. */

void TagDMRData::playbackParrot()
{
    if (m_parrotFrames.size() == 0) {
        m_parrotFramesReady = false;
        return;
    }

    m_parrotPlayback = true;

    auto& pkt = m_parrotFrames[0];
    m_parrotFrames.lock();
    if (pkt.buffer != nullptr) {
        // has the override source ID been set?
        if (m_network->m_parrotOverrideSrcId > 0U) {
            pkt.srcId = m_network->m_parrotOverrideSrcId;

            // override source ID
            SET_UINT24(m_network->m_parrotOverrideSrcId, pkt.buffer, 5U);

            /*
            ** bryanb: DMR is problematic because the VOICE_LC_HEADER, TERMINATOR_WITH_LC,
            ** and VOICE_PI_HEADER all contain the source ID in the LC portion of the frame
            ** and because we are not updating that the parrot playback will appear to come from
            ** the original source ID in those frames
            */
        }

        m_lastParrotPeerId = pkt.peerId;
        m_lastParrotSrcId = pkt.srcId;
        m_lastParrotDstId = pkt.dstId;

        if (m_network->m_parrotOnlyOriginating) {
            m_network->writePeer(pkt.peerId, pkt.peerId, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, pkt.buffer, pkt.bufferLen, pkt.pktSeq, pkt.streamId);
            if (m_network->m_debug) {
                LogDebugEx(LOG_DMR, "TagDMRData::playbackParrot()", "Parrot, dstPeer = %u, len = %u, pktSeq = %u, streamId = %u", 
                    pkt.peerId, pkt.bufferLen, pkt.pktSeq, pkt.streamId);
            }
        }
        else {
            // repeat traffic to the connected peers
            uint32_t i = 0U;
            udp::BufferQueue queue = udp::BufferQueue();

            m_network->m_peers.shared_lock();
            for (auto peer : m_network->m_peers) {
                // every MAX_QUEUED_PEER_MSGS peers flush the queue
                if (i % MAX_QUEUED_PEER_MSGS == 0U) {
                    m_network->m_frameQueue->flushQueue(&queue);
                }

                m_network->writePeerQueue(&queue, peer.first, pkt.peerId, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, pkt.buffer, pkt.bufferLen, pkt.pktSeq, pkt.streamId);
                if (m_network->m_debug) {
                    LogDebugEx(LOG_DMR, "TagDMRData::playbackParrot()", "Parrot, dstPeer = %u, len = %u, pktSeq = %u, streamId = %u", 
                        peer.first, pkt.bufferLen, pkt.pktSeq, pkt.streamId);
                }

                i++;
            }
            m_network->m_frameQueue->flushQueue(&queue);
            m_network->m_peers.shared_unlock();
        }

        delete[] pkt.buffer;
    }
    Thread::sleep(60);
    m_parrotFrames.unlock();
    m_parrotFrames.pop_front();
}

/* Helper to write a extended function packet on the RF interface. */

void TagDMRData::write_Ext_Func(uint32_t peerId, uint8_t slot, uint32_t func, uint32_t arg, uint32_t dstId)
{
    std::unique_ptr<lc::csbk::CSBK_EXT_FNCT> csbk = std::make_unique<lc::csbk::CSBK_EXT_FNCT>();
    csbk->setGI(false);
    csbk->setExtendedFunction(func);
    csbk->setSrcId(arg);
    csbk->setDstId(dstId);

    LogInfoEx(LOG_DMR, "DMR Slot %u, DT_CSBK, %s, op = $%02X, arg = %u, tgt = %u",
        slot, csbk->toString().c_str(), func, arg, dstId);

    write_CSBK(peerId, slot, csbk.get());
}

/* Helper to write a call alert packet on the RF interface. */

void TagDMRData::write_Call_Alrt(uint32_t peerId, uint8_t slot, uint32_t srcId, uint32_t dstId)
{
    std::unique_ptr<lc::csbk::CSBK_CALL_ALRT> csbk = std::make_unique<lc::csbk::CSBK_CALL_ALRT>();
    csbk->setGI(false);
    csbk->setSrcId(srcId);
    csbk->setDstId(dstId);

    LogInfoEx(LOG_DMR, "DMR Slot %u, DT_CSBK, %s, srcId = %u, dstId = %u",
        slot, csbk->toString().c_str(), srcId, dstId);

    write_CSBK(peerId, slot, csbk.get());
}

// ---------------------------------------------------------------------------
//  Private Class Members
// ---------------------------------------------------------------------------

/* Helper to route rewrite the network data buffer. */

void TagDMRData::routeRewrite(uint8_t* buffer, uint32_t peerId, dmr::data::NetData& dmrData, DataType::E dataType, uint32_t dstId, uint32_t slotNo, bool outbound)
{
    uint32_t rewriteDstId = dstId;
    uint32_t rewriteSlotNo = slotNo;

    // does the data require route rewriting?
    if (peerRewrite(peerId, rewriteDstId, rewriteSlotNo, outbound)) {
        // rewrite destination TGID in the frame
        SET_UINT24(rewriteDstId, buffer, 8U);

        // set or clear the e.Slot flag (if 0x80 is set Slot 2 otherwise Slot 1)
        if (rewriteSlotNo == 2 && (buffer[15U] & 0x80U) == 0x00U)
            buffer[15U] |= 0x80;
        if (rewriteSlotNo == 1 && (buffer[15U] & 0x80U) == 0x80U)
            buffer[15U] = buffer[15U] & ~0x80U;

        uint8_t data[DMR_FRAME_LENGTH_BYTES + 2U];
        dmrData.getData(data + 2U);

        if (dataType == DataType::VOICE_LC_HEADER ||
            dataType == DataType::TERMINATOR_WITH_LC) {
            // decode and reconstruct embedded DMR data
            lc::FullLC fullLC;
            std::unique_ptr<lc::LC> lc = fullLC.decode(data + 2U, dataType);
            if (lc == nullptr) {
                LogWarning(LOG_DMR, "DMR Slot %u, bad LC received from the network, replacing", slotNo);
                lc = std::make_unique<lc::LC>(dmrData.getFLCO(), dmrData.getSrcId(), rewriteDstId);
            }

            lc->setDstId(rewriteDstId);

            // Regenerate the LC data
            fullLC.encode(*lc, data + 2U, dataType);
            dmrData.setData(data + 2U);
        }
        else if (dataType == DataType::VOICE_PI_HEADER) {
            // decode and reconstruct embedded DMR data
            lc::FullLC fullLC;
            std::unique_ptr<lc::PrivacyLC> lc = fullLC.decodePI(data + 2U);
            if (lc == nullptr) {
                LogWarning(LOG_DMR, "DMR Slot %u, DT_VOICE_PI_HEADER, bad LC received, replacing", slotNo);
                lc = std::make_unique<lc::PrivacyLC>();
            }

            lc->setDstId(rewriteDstId);

            // Regenerate the LC data
            fullLC.encodePI(*lc, data + 2U);
            dmrData.setData(data + 2U);
        }

        dmrData.getData(buffer + 20U);
    }
}

/* Helper to route rewrite destination ID and slot. */

bool TagDMRData::peerRewrite(uint32_t peerId, uint32_t& dstId, uint32_t& slotNo, bool outbound)
{
    lookups::TalkgroupRuleGroupVoice tg;
    if (outbound) {
        tg = m_network->m_tidLookup->find(dstId);
    }
    else {
        tg = m_network->m_tidLookup->findByRewrite(peerId, dstId);
    }

    bool rewrote = false;
    if (tg.config().rewriteSize() > 0) {
        std::vector<lookups::TalkgroupRuleRewrite> rewrites = tg.config().rewrite();
        for (auto entry : rewrites) {
            if (entry.peerId() == peerId) {
                if (outbound) {
                    dstId = entry.tgId();
                    slotNo = entry.tgSlot();
                }
                else {
                    dstId = tg.source().tgId();
                    slotNo = tg.source().tgSlot();
                }
                rewrote = true;
                break;
            }
        }
    }

    return rewrote;
}

/* Helper to process CSBKs being passed from a peer. */

bool TagDMRData::processCSBK(uint8_t* buffer, uint32_t peerId, dmr::data::NetData& dmrData)
{
    // are we receiving a CSBK?
    if (dmrData.getDataType() == DataType::CSBK) {
        uint8_t data[DMR_FRAME_LENGTH_BYTES + 2U];
        dmrData.getData(data + 2U);

        std::unique_ptr<lc::CSBK> csbk = lc::csbk::CSBKFactory::createCSBK(data + 2U, DataType::CSBK);
        if (csbk != nullptr) {
            // report csbk event to InfluxDB
            if (m_network->m_enableInfluxDB && m_network->m_influxLogRawData) {
                const uint8_t* raw = csbk->getDecodedRaw();
                if (raw != nullptr) {
                    std::stringstream ss;
                    ss << std::hex <<
                        (int)raw[0] << (int)raw[1]  << (int)raw[2] << (int)raw[4] <<
                        (int)raw[5] << (int)raw[6]  << (int)raw[7] << (int)raw[8] <<
                        (int)raw[9] << (int)raw[10] << (int)raw[11];

                    influxdb::QueryBuilder()
                        .meas("csbk_event")
                            .tag("peerId", std::to_string(peerId))
                            .tag("lco", __INT_HEX_STR(csbk->getCSBKO()))
                            .tag("csbk", csbk->toString())
                                .field("raw", ss.str())
                            .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                        .requestAsync(m_network->m_influxServer);
                }
            }

            switch (csbk->getCSBKO()) {
            case CSBKO::BROADCAST:
                {
                    lc::csbk::CSBK_BROADCAST* osp = static_cast<lc::csbk::CSBK_BROADCAST*>(csbk.get());
                    if (osp->getAnncType() == BroadcastAnncType::ANN_WD_TSCC) {
                        if (m_network->m_disallowAdjStsBcast) {
                            // LogWarning(LOG_DMR, "PEER %u, passing BroadcastAnncType::ANN_WD_TSCC to internal peers is prohibited, dropping", peerId);
                            return false;
                        } else {
                            if (m_network->m_verbose) {
                                LogInfoEx(LOG_DMR, "DMR Slot %u, DT_CSBK, %s, sysId = $%03X, chNo = %u, peerId = %u", dmrData.getSlotNo(), csbk->toString().c_str(),
                                    osp->getSystemId(), osp->getLogicalCh1(), peerId);
                            }
                        }
                    }
                }
                break;
            default:
                break;
            }
        } else {
            std::string peerIdentity = m_network->resolvePeerIdentity(peerId);
            LogWarning(LOG_DMR, "PEER %u (%s), passing CSBK that failed to decode? csbk == nullptr", peerId, peerIdentity.c_str());
        }
    }

    return true;
}

/* Helper to determine if the peer is permitted for traffic. */

bool TagDMRData::isPeerPermitted(uint32_t peerId, data::NetData& data, uint32_t streamId, bool fromUpstream)
{
    // promiscuous hub mode performs no ACL checking and will pass all traffic
    if (g_promiscuousHub)
        return true;

    if (data.getFLCO() == FLCO::PRIVATE) {
        if (m_network->m_disallowU2U)
            return false;
        if (!m_network->checkU2UDroppedPeer(peerId))
            return true;
        return false;
    }

    FNEPeerConnection* connection = nullptr; // bryanb: this is a possible null ref concurrency issue
                                             //     it is possible if the timing is just right to get a valid 
                                             //     connection back initially, and then for it to be deleted
    if (peerId > 0 && (m_network->m_peers.find(peerId) != m_network->m_peers.end())) {
        connection = m_network->m_peers[peerId];
    }

    if (connection != nullptr) {
        // is this peer a replica peer?
        if (connection->isReplica()) {
            return true; // replica peers are *always* allowed to receive traffic and no other rules may filter
                         // these peers
        }

        // is this peer a SysView peer?
        if (connection->peerClass() == PEER_CONN_CLASS_SYSVIEW) {
            return true; // SysView peers are *always* allowed to receive traffic and no other rules may filter
                         // these peers
        }
    }

    // is this a group call?
    if (data.getFLCO() == FLCO::GROUP) {
        lookups::TalkgroupRuleGroupVoice tg = m_network->m_tidLookup->find(data.getDstId(), data.getSlotNo());

        std::vector<uint32_t> inclusion = tg.config().inclusion();
        std::vector<uint32_t> exclusion = tg.config().exclusion();

        // peer inclusion lists take priority over exclusion lists
        if (inclusion.size() > 0) {
            auto it = std::find(inclusion.begin(), inclusion.end(), peerId);
            if (it == inclusion.end()) {
                return false;
            }
        }
        else {
            if (exclusion.size() > 0) {
                auto it = std::find(exclusion.begin(), exclusion.end(), peerId);
                if (it != exclusion.end()) {
                    return false;
                }
            }
        }

        // peer always send list takes priority over any following affiliation rules
        std::vector<uint32_t> alwaysSend = tg.config().alwaysSend();
        if (alwaysSend.size() > 0) {
            auto it = std::find(alwaysSend.begin(), alwaysSend.end(), peerId);
            if (it != alwaysSend.end()) {
                return true; // skip any following checks and always send traffic
            }
        }

        // is this peer a conventional peer?
        if (m_network->m_allowConvSiteAffOverride) {
            if (connection != nullptr) {
                if (connection->peerClass() == PEER_CONN_CLASS_STANDARD && connection->isConventional()) {
                    fromUpstream = true; // we'll just set the fromUpstream flag to disable the affiliation check
                                         // for conventional peers
                }
            }
        }

        // is this peer a console peer?
        if (connection != nullptr) {
            if (connection->peerClass() == PEER_CONN_CLASS_CONSOLE) {
                fromUpstream = true; // we'll just set the fromUpstream flag to disable the affiliation check
                                     // for console peers
            }
        }

        // is this a TG that requires affiliations to repeat?
        // NOTE: neighbor FNE peers *always* repeat traffic regardless of affiliation
        if (tg.config().affiliated() && !fromUpstream) {
            uint32_t lookupPeerId = peerId;
            if (connection != nullptr) {
                if (connection->ccPeerId() > 0U)
                    lookupPeerId = connection->ccPeerId();
            }

            // check the affiliations for this peer to see if we can repeat traffic
            std::shared_ptr<fne_lookups::AffiliationLookup> aff = m_network->getPeerAffiliations(lookupPeerId);
            if (aff == nullptr) {
                std::string peerIdentity = m_network->resolvePeerIdentity(lookupPeerId);
                //LogError(LOG_NET, "PEER %u (%s) has an invalid affiliations lookup? This shouldn't happen BUGBUG.", lookupPeerId, peerIdentity.c_str());
                return false; // this will cause no traffic to pass for this peer now...I'm not sure this is good behavior
            }
            else {
                if (!aff->hasGroupAff(data.getDstId())) {
                    return false;
                }
            }
        }
    }

    return true;
}

/* Helper to validate the DMR call stream. */

bool TagDMRData::validate(uint32_t peerId, data::NetData& data, lc::CSBK* csbk, uint32_t streamId)
{
    // promiscuous hub mode performs no ACL checking and will pass all traffic
    if (g_promiscuousHub)
        return true;

    // is the source ID a blacklisted ID?
    bool rejectUnknownBadCall = false;
    lookups::RadioId rid = m_network->m_ridLookup->find(data.getSrcId());
    if (!rid.radioDefault()) {
        if (!rid.radioEnabled()) {
            // report error event to InfluxDB
            if (m_network->m_enableInfluxDB) {
                influxdb::QueryBuilder()
                    .meas("call_error_event")
                        .tag("peerId", std::to_string(peerId))
                        .tag("streamId", std::to_string(streamId))
                        .tag("srcId", std::to_string(data.getSrcId()))
                        .tag("dstId", std::to_string(data.getDstId()))
                            .field("message", std::string(INFLUXDB_ERRSTR_DISABLED_SRC_RID))
                            .field("slot", std::to_string(data.getSlotNo()))
                        .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                    .requestAsync(m_network->m_influxServer);
            }

            if (m_network->m_logDenials)
                LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_DISABLED_SRC_RID ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId());

            // report In-Call Control to the peer sending traffic
            m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo());
            return false;
        }
    }
    else {
        // if this is a default radio -- and we are rejecting undefined radios
        // report call error
        if (m_network->m_rejectUnknownRID) {
            rejectUnknownBadCall = true;
        }
    }

    // always validate a terminator if the source is valid
    if (data.getDataType() == DataType::TERMINATOR_WITH_LC)
        return true;

    // always validate a CSBK if the source is valid
    if (data.getDataType() == DataType::CSBK) {
        if (rejectUnknownBadCall)
            return false;

        if (csbk != nullptr) {
            // handle standard DMR reference opcodes
            switch (csbk->getCSBKO()) {
                case CSBKO::PV_GRANT:
                {
                    // is the destination ID a blacklisted ID?
                    lookups::RadioId rid = m_network->m_ridLookup->find(data.getDstId());
                    if (!rid.radioDefault()) {
                        if (!rid.radioEnabled()) {
                            return false;
                        }
                    }
                }
                break;
                case CSBKO::TV_GRANT:
                {
                    lookups::TalkgroupRuleGroupVoice tg = m_network->m_tidLookup->find(csbk->getDstId());

                    // check TGID validity
                    if (tg.isInvalid()) {
                        return false;
                    }

                    if (!tg.config().active()) {
                        return false;
                    }
                }
                break;
                case CSBKO::EXT_FNCT:
                {
                    if (csbk->getFID() == FID_MOT) {
                        const lc::csbk::CSBK_EXT_FNCT* iosp = static_cast<const lc::csbk::CSBK_EXT_FNCT*>(csbk);
                        if (iosp != nullptr) {
                            lookups::PeerId pid = m_network->m_peerListLookup->find(peerId);
                            uint32_t func = iosp->getExtendedFunction();
                            switch (func) {
                                case ExtendedFunctions::INHIBIT:
                                case ExtendedFunctions::UNINHIBIT:
                                    {
                                        if (!pid.peerDefault() && !pid.canIssueInhibit()) {
                                            LogWarning(LOG_DMR, "DMR, PEER %u attempted inhibit/unhibit, not authorized", peerId);
                                            return false;
                                        }
                                    }
                                    break;
                            }
                        }
                    }
                }
                break;
            }
        }

        return true;
    }

    // is the call stream rejected?
    m_rejectedCallStreams.lock(false);
    std::vector<uint32_t> rejectedStreams = m_rejectedCallStreams[data.getDstId()];
    if (std::find(rejectedStreams.begin(), rejectedStreams.end(), streamId) != rejectedStreams.end()) {
        // report error event to InfluxDB
        if (m_network->m_enableInfluxDB) {
            influxdb::QueryBuilder()
                .meas("call_error_event")
                    .tag("peerId", std::to_string(peerId))
                    .tag("streamId", std::to_string(streamId))
                    .tag("srcId", std::to_string(data.getSrcId()))
                    .tag("dstId", std::to_string(data.getDstId()))
                        .field("message", std::string(INFLUXDB_ERRSTR_CALL_NOT_PERMITTED))
                        .field("slot", std::to_string(data.getSlotNo()))
                    .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                .requestAsync(m_network->m_influxServer);
        }

        if (m_network->m_logDenials)
            LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_CALL_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId());

        m_rejectedCallStreams.unlock();

        // report In-Call Control to the peer sending traffic
        m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo());
        return false;
    }
    m_rejectedCallStreams.unlock();

    // is this a private call?
    if (data.getFLCO() == FLCO::PRIVATE) {
        // is the destination ID a blacklisted ID?
        lookups::RadioId rid = m_network->m_ridLookup->find(data.getDstId());
        if (!rid.radioDefault()) {
            if (!rid.radioEnabled()) {
                // report error event to InfluxDB
                if (m_network->m_enableInfluxDB) {
                    influxdb::QueryBuilder()
                        .meas("call_error_event")
                            .tag("peerId", std::to_string(peerId))
                            .tag("streamId", std::to_string(streamId))
                            .tag("srcId", std::to_string(data.getSrcId()))
                            .tag("dstId", std::to_string(data.getDstId()))
                                .field("message", std::string(INFLUXDB_ERRSTR_DISABLED_DST_RID))
                                .field("slot", std::to_string(data.getSlotNo()))
                            .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                        .requestAsync(m_network->m_influxServer);
                }

                if (m_network->m_logDenials)
                    LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_DISABLED_DST_RID ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId());

                return false;
            }
        }
        else {
            // if this is a default radio -- and we are rejecting undefined radios
            // report call error
            if (m_network->m_rejectUnknownRID) {
                // report error event to InfluxDB
                if (m_network->m_enableInfluxDB) {
                    influxdb::QueryBuilder()
                        .meas("call_error_event")
                            .tag("peerId", std::to_string(peerId))
                            .tag("streamId", std::to_string(streamId))
                            .tag("srcId", std::to_string(data.getSrcId()))
                            .tag("dstId", std::to_string(data.getDstId()))
                                .field("message", std::string(INFLUXDB_ERRSTR_ILLEGAL_RID_ACCESS))
                                .field("slot", std::to_string(data.getSlotNo()))
                            .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                        .requestAsync(m_network->m_influxServer);
                }

                if (m_network->m_logDenials)
                    LogWarning(LOG_DMR, "DMR slot %s, " INFLUXDB_ERRSTR_ILLEGAL_RID_ACCESS ", srcId = %u, dstId = %u", data.getSlotNo(), data.getSrcId(), data.getDstId());

                // report In-Call Control to the peer sending traffic
                m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo());
                return false;
            }
        }
    }

    // is this a group call?
    if (data.getFLCO() == FLCO::GROUP) {
        lookups::TalkgroupRuleGroupVoice tg = m_network->m_tidLookup->find(data.getDstId());
        if (tg.isInvalid()) {
            // report error event to InfluxDB
            if (m_network->m_enableInfluxDB) {
                influxdb::QueryBuilder()
                    .meas("call_error_event")
                        .tag("peerId", std::to_string(peerId))
                        .tag("streamId", std::to_string(streamId))
                        .tag("srcId", std::to_string(data.getSrcId()))
                        .tag("dstId", std::to_string(data.getDstId()))
                            .field("message", std::string(INFLUXDB_ERRSTR_INV_TALKGROUP))
                            .field("slot", std::to_string(data.getSlotNo()))
                        .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                    .requestAsync(m_network->m_influxServer);
            }

            if (m_network->m_logDenials)
                LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_INV_TALKGROUP ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId());

            // report In-Call Control to the peer sending traffic
            m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(),  data.getSlotNo());
            return false;
        }

        /*
        ** bryanb: yes the encryption else condition is missing here -- DMR is a bit of a mess when it comes to 
        **  handling encryption because AFAIK the TS.102 spec doesn't clearly indicate how to handle encryption; 
        **  DVM's FNE cannot reliably implement selectable, strapped, clear checking because it won't know reliably 
        **  if the call is encrypted or not until it receives a VOICE_LC_HEADER or VOICE_PI_HEADER frame, and by that 
        **  time it'll really be too late to reject the call
        */

        // peer always send list takes priority over any following affiliation rules
        bool isAlwaysPeer = false;
        std::vector<uint32_t> alwaysSend = tg.config().alwaysSend();
        if (alwaysSend.size() > 0) {
            auto it = std::find(alwaysSend.begin(), alwaysSend.end(), peerId);
            if (it != alwaysSend.end()) {
                isAlwaysPeer = true; // skip any following checks and always send traffic
                rejectUnknownBadCall = false;
            }
        }

        // fail call if the reject flag is set
        if (rejectUnknownBadCall) {
            // report error event to InfluxDB
            if (m_network->m_enableInfluxDB) {
                influxdb::QueryBuilder()
                    .meas("call_error_event")
                        .tag("peerId", std::to_string(peerId))
                        .tag("streamId", std::to_string(streamId))
                        .tag("srcId", std::to_string(data.getSrcId()))
                        .tag("dstId", std::to_string(data.getDstId()))
                            .field("message", std::string(INFLUXDB_ERRSTR_ILLEGAL_RID_ACCESS))
                            .field("slot", std::to_string(data.getSlotNo()))
                        .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                    .requestAsync(m_network->m_influxServer);
            }

            if (m_network->m_logDenials)
                LogWarning(LOG_DMR, "DMR slot %s, " INFLUXDB_ERRSTR_ILLEGAL_RID_ACCESS ", srcId = %u, dstId = %u", data.getSlotNo(), data.getSrcId(), data.getDstId());

            // report In-Call Control to the peer sending traffic
            m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo());
            return false;
        }

        // check the DMR slot number
        if (tg.source().tgSlot() != data.getSlotNo()) {
            // report error event to InfluxDB
            if (m_network->m_enableInfluxDB) {
                influxdb::QueryBuilder()
                    .meas("call_error_event")
                        .tag("peerId", std::to_string(peerId))
                        .tag("streamId", std::to_string(streamId))
                        .tag("srcId", std::to_string(data.getSrcId()))
                        .tag("dstId", std::to_string(data.getDstId()))
                            .field("message", std::string(INFLUXDB_ERRSTR_INV_SLOT))
                            .field("slot", std::to_string(data.getSlotNo()))
                        .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                    .requestAsync(m_network->m_influxServer);
            }

            if (m_network->m_logDenials)
                LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_INV_SLOT ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId());

            // report In-Call Control to the peer sending traffic
            m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo());
            return false;
        }

        // is the TGID active?
        if (!tg.config().active()) {
            // report error event to InfluxDB
            if (m_network->m_enableInfluxDB) {
                influxdb::QueryBuilder()
                    .meas("call_error_event")
                        .tag("peerId", std::to_string(peerId))
                        .tag("streamId", std::to_string(streamId))
                        .tag("srcId", std::to_string(data.getSrcId()))
                        .tag("dstId", std::to_string(data.getDstId()))
                            .field("message", std::string(INFLUXDB_ERRSTR_DISABLED_TALKGROUP))
                            .field("slot", std::to_string(data.getSlotNo()))
                        .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                    .requestAsync(m_network->m_influxServer);
            }

            if (m_network->m_logDenials)
                LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_DISABLED_TALKGROUP ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId());

            // report In-Call Control to the peer sending traffic
            m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo());
            return false;
        }

        // always peers can violate the rules...hurray
        if (!isAlwaysPeer) {
            // does the TGID have a permitted RID list?
            if (tg.config().permittedRIDs().size() > 0) {
                // does the transmitting RID have permission?
                std::vector<uint32_t> permittedRIDs = tg.config().permittedRIDs();
                if (std::find(permittedRIDs.begin(), permittedRIDs.end(), data.getSrcId()) == permittedRIDs.end()) {
                    // report error event to InfluxDB
                    if (m_network->m_enableInfluxDB) {
                        influxdb::QueryBuilder()
                            .meas("call_error_event")
                                .tag("peerId", std::to_string(peerId))
                                .tag("streamId", std::to_string(streamId))
                                .tag("srcId", std::to_string(data.getSrcId()))
                                .tag("dstId", std::to_string(data.getDstId()))
                                    .field("message", std::string(INFLUXDB_ERRSTR_RID_NOT_PERMITTED))
                                .timestamp(std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count())
                            .requestAsync(m_network->m_influxServer);
                    }

                    if (m_network->m_logDenials)
                        LogError(LOG_DMR, "DMR Slot %u, " INFLUXDB_ERRSTR_RID_NOT_PERMITTED ", peer = %u, srcId = %u, dstId = %u", data.getSlotNo(), peerId, data.getSrcId(), data.getDstId());

                    // report In-Call Control to the peer sending traffic
                    m_network->writePeerICC(peerId, streamId, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR, NET_ICC::REJECT_TRAFFIC, data.getDstId(), data.getSlotNo());
                    return false;
                }
            }
        }
    }

    return true;
}

/* Helper to write a grant packet. */

bool TagDMRData::write_CSBK_Grant(uint32_t peerId, uint32_t srcId, uint32_t dstId, uint8_t serviceOptions, bool grp)
{
    uint8_t slot = 0U;

    bool emergency = ((serviceOptions & 0xFFU) & 0x80U) == 0x80U;           // Emergency Flag
    bool privacy = ((serviceOptions & 0xFFU) & 0x40U) == 0x40U;             // Privacy Flag
    bool broadcast = ((serviceOptions & 0xFFU) & 0x10U) == 0x10U;           // Broadcast Flag
    uint8_t priority = ((serviceOptions & 0xFFU) & 0x03U);                  // Priority

    if (dstId == WUID_ALL || dstId == WUID_ALLZ || dstId == WUID_ALLL) {
        return true; // do not generate grant packets for $FFFF (All Call) TGID
    }

    // check the affiliations for this peer to see if we can grant traffic
    std::shared_ptr<fne_lookups::AffiliationLookup> aff = m_network->getPeerAffiliations(peerId);
    if (aff == nullptr) {
        std::string peerIdentity = m_network->resolvePeerIdentity(peerId);
        LogError(LOG_MASTER, "PEER %u (%s) has an invalid affiliations lookup? This shouldn't happen BUGBUG.", peerId, peerIdentity.c_str());
        return false; // this will cause no traffic to pass for this peer now...I'm not sure this is good behavior
    }
    else {
        if (!aff->hasGroupAff(dstId)) {
            return false;
        }
    }

    if (grp) {
        std::unique_ptr<lc::csbk::CSBK_TV_GRANT> csbk = std::make_unique<lc::csbk::CSBK_TV_GRANT>();
        if (broadcast)
            csbk->setCSBKO(CSBKO::BTV_GRANT);
        csbk->setLogicalCh1(0U);
        csbk->setSlotNo(slot);

        if (m_network->m_verbose) {
            LogInfoEx(LOG_DMR, "DMR, CSBK, %s, emerg = %u, privacy = %u, broadcast = %u, prio = %u, chNo = %u, slot = %u, srcId = %u, dstId = %u, peerId = %u",
                csbk->toString().c_str(), emergency, privacy, broadcast, priority, csbk->getLogicalCh1(), csbk->getSlotNo(), srcId, dstId, peerId);
        }

        csbk->setEmergency(emergency);
        csbk->setSrcId(srcId);
        csbk->setDstId(dstId);

        write_CSBK(peerId, 1U, csbk.get());
    }
    else {
        std::unique_ptr<lc::csbk::CSBK_PV_GRANT> csbk = std::make_unique<lc::csbk::CSBK_PV_GRANT>();
        csbk->setLogicalCh1(0U);
        csbk->setSlotNo(slot);

        if (m_network->m_verbose) {
            LogInfoEx(LOG_DMR, "DMR, CSBK, %s, emerg = %u, privacy = %u, broadcast = %u, prio = %u, chNo = %u, slot = %u, srcId = %u, dstId = %u, peerId = %u",
                csbk->toString().c_str(), emergency, privacy, broadcast, priority, csbk->getLogicalCh1(), csbk->getSlotNo(), srcId, dstId, peerId);
        }

        csbk->setEmergency(emergency);
        csbk->setSrcId(srcId);
        csbk->setDstId(dstId);

        write_CSBK(peerId, 1U, csbk.get());
    }

    return true;
}

/* Helper to write a NACK RSP packet. */

void TagDMRData::write_CSBK_NACK_RSP(uint32_t peerId, uint32_t dstId, uint8_t slot, uint8_t reason, uint8_t service)
{
    std::unique_ptr<lc::csbk::CSBK_NACK_RSP> csbk = std::make_unique<lc::csbk::CSBK_NACK_RSP>();
    csbk->setServiceKind(service);
    csbk->setReason(reason);
    csbk->setSrcId(WUID_ALL); // hmmm...
    csbk->setDstId(dstId);

    if (m_network->m_verbose) {
        LogInfoEx(LOG_DMR, "DMR Slot %u, CSBK, %s, reason = $%02X (%s), srcId = %u, dstId = %u",
            slot, csbk->toString().c_str(), reason, DMRUtils::rsnToString(reason).c_str(),
            csbk->getSrcId(), csbk->getDstId());
    }

    write_CSBK(peerId, slot, csbk.get());
}

/* Helper to write a network CSBK. */

void TagDMRData::write_CSBK(uint32_t peerId, uint8_t slot, lc::CSBK* csbk)
{
    uint8_t data[DMR_FRAME_LENGTH_BYTES + 2U];
    ::memset(data + 2U, 0x00U, DMR_FRAME_LENGTH_BYTES);

    SlotType slotType;
    slotType.setColorCode(0U);
    slotType.setDataType(DataType::CSBK);

    // Regenerate the CSBK data
    csbk->encode(data + 2U);

    // Regenerate the Slot Type
    slotType.encode(data + 2U);

    // Convert the Data Sync to be from the BS or MS as needed
    Sync::addDMRDataSync(data + 2U, true);

    data::NetData dmrData;
    dmrData.setSlotNo(slot);
    dmrData.setDataType(DataType::CSBK);
    dmrData.setSrcId(csbk->getSrcId());
    dmrData.setDstId(csbk->getDstId());
    dmrData.setFLCO(csbk->getGI() ? FLCO::GROUP : FLCO::PRIVATE);
    dmrData.setN(0U);
    dmrData.setSeqNo(0U);
    dmrData.setBER(0U);
    dmrData.setRSSI(0U);

    dmrData.setData(data + 2U);

    uint32_t streamId = m_network->createStreamId();
    uint32_t messageLength = 0U;
    UInt8Array message = m_network->createDMR_Message(messageLength, streamId, dmrData);
    if (message == nullptr) {
        return;
    }

    if (peerId > 0U) {
        m_network->writePeer(peerId, m_network->m_peerId, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId);
    } else {
        // repeat traffic to the connected peers
        if (m_network->m_peers.size() > 0U) {
            uint32_t i = 0U;
            udp::BufferQueue queue = udp::BufferQueue();

            m_network->m_peers.shared_lock();
            for (auto peer : m_network->m_peers) {
                // every MAX_QUEUED_PEER_MSGS peers flush the queue
                if (i % MAX_QUEUED_PEER_MSGS == 0U) {
                    m_network->m_frameQueue->flushQueue(&queue);
                }

                m_network->writePeerQueue(&queue, peer.first, m_network->m_peerId, { NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId);
                if (m_network->m_debug) {
                    LogDebugEx(LOG_DMR, "TagDMRData::write_CSBK()", "peer = %u, slotNo = %u, len = %u, stream = %u", 
                        peer.first, slot, messageLength, streamId);
                }
                i++;
            }
            m_network->m_frameQueue->flushQueue(&queue);
            m_network->m_peers.shared_unlock();
        }

        // repeat traffic to neighbor FNE peers
        if (m_network->m_host->m_peerNetworks.size() > 0U) {
            for (auto peer : m_network->m_host->m_peerNetworks) {
                uint32_t dstPeerId = peer.second->getPeerId();

                peer.second->writeMaster({ NET_FUNC::PROTOCOL, NET_SUBFUNC::PROTOCOL_SUBFUNC_DMR }, message.get(), messageLength, RTP_END_OF_CALL_SEQ, streamId);
                if (m_network->m_debug) {
                    LogDebugEx(LOG_DMR, "TagDMRData::write_CSBK()", "peer = %u, slotNo = %u, len = %u, stream = %u", 
                        dstPeerId, slot, messageLength, streamId);
                }
            }
        }
    }
}
