destiny-2-Sunrise/Sunrise/src/server/bap/encrypted/queuez/queuez_deferred_push.cpp

#include <Windows.h>

#include <algorithm>
#include <array>
#include <cstdio>

#include "../../../../core/logging/log.h"
#include "../../../../middleware/secure_channel/runtime.h"
#include "../../../../state/account/account_state.h"
#include "../../../../state/runtime/runtime.h"
#include "../internal.h"
#include "../push/activity/activity_keepalive_push.h"
#include "queuez_state_validation.h"

namespace sunrise::server::bap::encrypted {
namespace {

/** Widest re-push report, sized for the fields below. */
constexpr std::size_t kRepushReportLimit = 96;

/**
 * Logs one delayed re-push with its framed size, so it can be compared to the first copy.
 * @param stage Point in the deferred push the line reports.
 * @param bytes Framed size of the published notification.
 */
void report_repush(const char* stage, std::size_t bytes) noexcept {
    std::array<char, kRepushReportLimit> line{};
    const int count = std::snprintf(
        line.data(), line.size(), "ev=queuez stage=%s result=ok bytes=%zu", stage, bytes);
    if (count > 0) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::info,
                         {line.data(), static_cast<std::size_t>(count)});
    }
}

/** Publishes and commits one world reward through the ordinary acquisition notification path. */
[[nodiscard]] bool consume_world_item_acquisition(Session& session,
                                                  Scratch& scratch,
                                                  std::span<std::byte> response,
                                                  std::size_t& written,
                                                  bool& touchesScratch) noexcept {
    if (!session.worldItemAcquisitionArmed) {
        return false;
    }
    touchesScratch = true;
    queuez::ItemAcquisition acquisition{};
    const state::PendingItemAcquisition pending = session.pendingWorldItemAcquisition;
    if (!queuez::stage_item_acquisition(session.queuez,
                                        pending.accountSoid,
                                        pending.characterSoid,
                                        pending.acquiredInstanceSoid,
                                        pending.profileChanged,
                                        acquisition)) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=world_acquisition result=fail reason=stage");
        return false;
    }
    auto nextSendNonce = session.sendNonce;
    std::size_t framedSize = 0;
    if (!push::append_item_acquisition_notification(scratch,
                                                    acquisition,
                                                    pending,
                                                    state::bap().sessionKey,
                                                    nextSendNonce,
                                                    scratch.framed,
                                                    framedSize)
        || framedSize == 0 || framedSize > response.size()) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=world_acquisition result=fail reason=encode");
        return false;
    }
    if (!state::commit_item_acquisition(session.pendingWorldItemAcquisition)) {
        session.worldItemAcquisitionArmed = false;
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=world_acquisition result=fail reason=commit");
        return false;
    }
    std::copy_n(scratch.framed.begin(), framedSize, response.begin());
    written = framedSize;
    middleware::secure_channel::advance_nonce(nextSendNonce);
    session.sendNonce = nextSendNonce;
    session.queuez = acquisition.after;
    session.worldItemAcquisitionArmed = false;
    report_repush("world_acquisition", framedSize);
    return true;
}

/** Publishes and commits one profile material reward through its acquisition notification. */
[[nodiscard]] bool consume_world_profile_item_acquisition(Session& session,
                                                          Scratch& scratch,
                                                          std::span<std::byte> response,
                                                          std::size_t& written,
                                                          bool& touchesScratch) noexcept {
    if (!session.worldProfileItemAcquisitionArmed) {
        return false;
    }
    touchesScratch = true;
    queuez::ProfileItemAcquisition acquisition{};
    const state::PendingProfileItemAcquisition pending =
        session.pendingWorldProfileItemAcquisition;
    if (!queuez::stage_profile_item_acquisition(session.queuez,
                                                pending.accountSoid,
                                                pending.acquiredInstanceSoid,
                                                pending.actionSource,
                                                pending.appended,
                                                acquisition)) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=world_profile_acquisition result=fail reason=stage");
        return false;
    }
    auto nextSendNonce = session.sendNonce;
    std::size_t framedSize = 0;
    if (!push::append_profile_item_acquisition_notification(scratch,
                                                            acquisition,
                                                            pending,
                                                            state::bap().sessionKey,
                                                            nextSendNonce,
                                                            scratch.framed,
                                                            framedSize)
        || framedSize == 0 || framedSize > response.size()) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=world_profile_acquisition result=fail reason=encode");
        return false;
    }
    if (!state::commit_profile_item_acquisition(session.pendingWorldProfileItemAcquisition)) {
        session.worldProfileItemAcquisitionArmed = false;
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=world_profile_acquisition result=fail reason=commit");
        return false;
    }
    std::copy_n(scratch.framed.begin(), framedSize, response.begin());
    written = framedSize;
    middleware::secure_channel::advance_nonce(nextSendNonce);
    session.sendNonce = nextSendNonce;
    session.queuez = acquisition.after;
    session.worldProfileItemAcquisitionArmed = false;
    report_repush("world_profile_acquisition", framedSize);
    return true;
}

/** Publishes the current account graph to a peer invalidated by another connection. */
[[nodiscard]] bool consume_account_resync(Session& session,
                                          Scratch& scratch,
                                          std::span<std::byte> response,
                                          std::size_t& written,
                                          bool& touchesScratch) noexcept {
    if (!session.accountResyncArmed || session.accountResyncGeneration == 0) {
        return false;
    }
    touchesScratch = true;
    auto nextSendNonce = session.sendNonce;
    std::size_t framedSize = 0;
    queuez::SessionState currentQueuez{};
    if (!push::append_account_resync_notification(scratch,
                                                  session.queuez,
                                                  state::bap().sessionKey,
                                                  nextSendNonce,
                                                  scratch.framed,
                                                  framedSize,
                                                  currentQueuez)) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=peer_resync result=fail reason=family4");
        return false;
    }
    if (currentQueuez.family0Active) {
        queuez::SessionState appearanceAfter{};
        if (!push::append_account_resync_appearance_notification(scratch,
                                                                 currentQueuez,
                                                                 state::bap().sessionKey,
                                                                 nextSendNonce,
                                                                 scratch.framed,
                                                                 framedSize,
                                                                 appearanceAfter)) {
            core::log::write(core::log::Channel::server,
                             core::log::Level::warn,
                             "ev=queuez stage=peer_resync result=fail reason=family0");
            return false;
        }
        currentQueuez = appearanceAfter;
    }
    if (currentQueuez.family3Active) {
        queuez::SessionState rosterAfter{};
        if (!push::append_account_resync_roster_notification(scratch,
                                                             currentQueuez,
                                                             state::bap().sessionKey,
                                                             nextSendNonce,
                                                             scratch.framed,
                                                             framedSize,
                                                             rosterAfter)) {
            core::log::write(core::log::Channel::server,
                             core::log::Level::warn,
                             "ev=queuez stage=peer_resync result=fail reason=family3");
            return false;
        }
        currentQueuez = rosterAfter;
    }
    if (framedSize == 0 || framedSize > response.size() || !queuez::valid(currentQueuez)) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=peer_resync result=fail reason=output");
        return false;
    }
    std::copy_n(scratch.framed.begin(), framedSize, response.begin());
    written = framedSize;
    session.sendNonce = nextSendNonce;
    session.queuez = currentQueuez;
    session.accountGeneration = session.accountResyncGeneration;
    session.accountResyncArmed = false;
    report_repush("peer_resync", framedSize);
    return true;
}

/**
 * Sends the owed banner re-push once its delay has passed.
 * The banner has no subscribe of its own, so the timer is its only second chance.
 * @param session Auth, nonce and queuez state owned by the connection.
 * @param scratch Transform buffers owned by the lock.
 * @param response Whole-frame storage owned by the caller.
 * @param written Gets the encoded notification size in bytes.
 * @param touchesScratch Set before any scratch buffer is used.
 * @return True when a whole banner notification is published.
 */
[[nodiscard]] bool consume_banner_repush(Session& session,
                                         Scratch& scratch,
                                         std::span<std::byte> response,
                                         std::size_t& written,
                                         bool& touchesScratch) noexcept {
    if (!session.bannerRepushArmed || session.bannerRepushRoot == 0
        || GetTickCount64() < session.bannerRepushDueTick) {
        return false;
    }
    // Nothing is owed while the account owns no character to name. The arm stays set, because it
    // is the banner's only second chance.
    if (state::account::banner_character_soid(state::account_snapshot()) == 0) {
        return false;
    }
    touchesScratch = true;

    // The same body the subscribe answer builds, so the version and this host's mirror stay in
    // step. `append_banner_notification` fixes the version at zero and a pick has moved past it.
    middleware::queuez::Subscription subscription{};
    subscription.familyType = queuez::kBannerFamilyType;
    subscription.familyRootSoid = session.bannerRepushRoot;

    auto nextSendNonce = session.sendNonce;
    std::size_t framedSize = 0;
    queuez::SessionState bannerAfter{};
    bool armsRepush = false;
    bool armsBannerRepush = false;
    push::append_queuez_notification(scratch,
                                     session.queuez,
                                     subscription,
                                     state::bap().sessionKey,
                                     nextSendNonce,
                                     scratch.framed,
                                     framedSize,
                                     bannerAfter,
                                     armsRepush,
                                     armsBannerRepush);
    if (framedSize == 0 || framedSize > response.size()) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=banner_repush result=fail");
        return false;
    }
    std::copy_n(scratch.framed.begin(), framedSize, response.begin());
    written = framedSize;
    session.sendNonce = nextSendNonce;
    // The frame is committed here, so the recorded delivery and the arm are committed with it.
    if (valid(bannerAfter)) {
        session.queuez = bannerAfter;
    }
    session.bannerRepushArmed = false;
    report_repush("banner_repush", framedSize);
    return true;
}

/**
 * Re-derives the selected character's appearance and roster once the ability-bucket rebuild owed
 * by a subclass selection has landed. The refresh sent inline with the opcode-801 response can
 * still carry empty buckets, because that rebuild runs off the Client content-extraction pump.
 * @param session Auth, nonce and queuez state owned by the connection.
 * @param scratch Transform buffers owned by the lock.
 * @param response Whole-frame storage owned by the caller.
 * @param written Gets the encoded notification size in bytes.
 * @param touchesScratch Set before any scratch buffer is used.
 * @return True when at least one owed record refreshes.
 */
[[nodiscard]] bool consume_ability_refresh(Session& session,
                                           Scratch& scratch,
                                           std::span<std::byte> response,
                                           std::size_t& written,
                                           bool& touchesScratch) noexcept {
    if (!session.abilityRefreshArmed || GetTickCount64() < session.abilityRefreshDueTick) {
        return false;
    }
    // Nothing is owed until a family that reads abilities is subscribed. The arm stays set, the
    // same way the banner re-push below keeps its own.
    if (!session.queuez.family0Active && !session.queuez.family3Active) {
        return false;
    }
    touchesScratch = true;

    auto nextSendNonce = session.sendNonce;
    std::size_t framedSize = 0;
    queuez::SessionState current = session.queuez;
    bool wrote = false;
    if (current.family0Active) {
        queuez::SessionState appearanceAfter{};
        if (push::append_account_resync_appearance_notification(scratch,
                                                                current,
                                                                state::bap().sessionKey,
                                                                nextSendNonce,
                                                                scratch.framed,
                                                                framedSize,
                                                                appearanceAfter)) {
            current = appearanceAfter;
            wrote = true;
        }
    }
    if (current.family3Active) {
        queuez::SessionState rosterAfter{};
        if (push::append_account_resync_roster_notification(scratch,
                                                            current,
                                                            state::bap().sessionKey,
                                                            nextSendNonce,
                                                            scratch.framed,
                                                            framedSize,
                                                            rosterAfter)) {
            current = rosterAfter;
            wrote = true;
        }
    }
    if (!wrote || framedSize == 0 || framedSize > response.size() || !queuez::valid(current)) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         "ev=queuez stage=ability_refresh result=fail");
        return false;
    }
    std::copy_n(scratch.framed.begin(), framedSize, response.begin());
    written = framedSize;
    session.sendNonce = nextSendNonce;
    session.queuez = current;
    // The frame is committed here, so the arm is committed with it. Disarming any earlier drops
    // the owed refresh on a transient encode failure.
    session.abilityRefreshArmed = false;
    report_repush("ability_refresh", framedSize);
    return true;
}

} // namespace

/**
 * Sends the owed Family-4 re-push once its delay has passed.
 * @param session Auth, nonce and queuez state owned by the connection.
 * @param scratch Transform buffers owned by the lock.
 * @param response Whole-frame storage owned by the caller.
 * @param written Gets the encoded notification size in bytes.
 * @param touchesScratch Set before any scratch buffer is used.
 * @return True when a whole Family-4 notification is published.
 */
bool consume_deferred(Session& session,
                      Scratch& scratch,
                      std::span<std::byte> response,
                      std::size_t& written,
                      bool& touchesScratch) noexcept {
    written = 0;
    if (!session.authenticated) {
        return false;
    }
    if (consume_world_item_acquisition(session, scratch, response, written, touchesScratch)) {
        return true;
    }
    if (session.worldItemAcquisitionArmed) {
        return false;
    }
    if (consume_world_profile_item_acquisition(session, scratch, response, written, touchesScratch)) {
        return true;
    }
    if (session.worldProfileItemAcquisitionArmed) {
        return false;
    }
    if (consume_account_resync(session, scratch, response, written, touchesScratch)) {
        return true;
    }
    // A failed resync remains armed and blocks unrelated deferred output until it can be retried.
    if (session.accountResyncArmed) {
        return false;
    }
    if (consume_ability_refresh(session, scratch, response, written, touchesScratch)) {
        return true;
    }
    if (!session.family4RepushArmed || session.family4RepushRoot == 0
        || GetTickCount64() < session.family4RepushDueTick) {
        return consume_banner_repush(session, scratch, response, written, touchesScratch)
               || push::activity::consume_activity_keepalive(
                   session, scratch, response, written, touchesScratch);
    }
    // One attempt is owed, and it is spent whether or not it lands.
    touchesScratch = true;

    middleware::queuez::Subscription subscription{};
    subscription.familyType = queuez::kAccountFamilyType;
    subscription.familyRootSoid = session.family4RepushRoot;

    auto nextSendNonce = session.sendNonce;
    std::size_t framedSize = 0;
    queuez::SessionState after{};
    bool armsRepush = false;
    bool armsBannerRepush = false;
    push::append_queuez_notification(scratch,
                                     session.queuez,
                                     subscription,
                                     state::bap().sessionKey,
                                     nextSendNonce,
                                     scratch.framed,
                                     framedSize,
                                     after,
                                     armsRepush,
                                     armsBannerRepush);
    if (framedSize == 0 || framedSize > response.size()) {
        // Neither failure clears on a retry. Holding the arm starves the keepalive, and the client
        // drops the activity session once the keepalive stops.
        session.family4RepushArmed = false;
        core::log::write(core::log::Channel::server,
                         core::log::Level::warn,
                         framedSize == 0 ? "ev=queuez stage=repush result=fail reason=encode"
                                         : "ev=queuez stage=repush result=fail reason=capacity");
        return false;
    }
    std::copy_n(scratch.framed.begin(), framedSize, response.begin());
    written = framedSize;
    session.sendNonce = nextSendNonce;
    if (queuez::valid(after)) {
        session.queuez = after;
    }
    session.family4RepushArmed = false;
    report_repush("repush", framedSize);
    return true;
}

} // namespace sunrise::server::bap::encrypted