#include <algorithm>
#include <array>
#include <cstddef>
#include <cstdio>
#include <limits>
#include <optional>
#include <span>

#include "../../../../../core/logging/log.h"
#include "../../../../../middleware/datagen/family4/account/account_encoder.h"
#include "../../../../../middleware/datagen/family4/account/layout.h"
#include "../../../../../middleware/datagen/family4/account/selection_patch/\
account_selection_patch_encoder.h"
#include "../../../../../middleware/datagen/family4/character/character_encoder.h"
#include "../../../../../middleware/datagen/family4/character/layout.h"
#include "../../../../../middleware/datagen/family4/instance/instance_encoder.h"
#include "../../../../../middleware/datagen/family4/instance/layout.h"
#include "../../../../../state/runtime/runtime.h"
#include "internal.h"
#include "snapshot_storage.h"

namespace sunrise::server::bap::encrypted::push::snapshot {
namespace {

namespace family4_datagen = middleware::datagen::family4;
namespace selection_patch = middleware::datagen::family4::account::selection_patch;

/** Logs the exact resolved and encoded fields used by the opcode-403 character upsert. */
void report_equipment_object(const queuez::EquipmentSwap& swap,
                             const state::PendingEquipmentSwap& mutation,
                             const Resolved& selected,
                             const family4_datagen::character::layout::Object& object) noexcept {
    constexpr std::size_t kMissing = (std::numeric_limits<std::size_t>::max)();
    std::size_t requestedRow = kMissing;
    std::size_t previousRow = kMissing;
    std::size_t nativeSlot = kMissing;
    bool requestedEquipped = false;
    bool previousEquipped = false;
    for (std::size_t index = 0; index < selected.loadout.itemCount; ++index) {
        const auto& item = selected.loadout.items[index];
        if (item.instance.instanceSoid == mutation.requestedInstanceSoid) {
            requestedRow = item.inventoryRow;
            nativeSlot = item.equipmentSlot;
            requestedEquipped = item.equipped;
        } else if (item.instance.instanceSoid == mutation.previousInstanceSoid) {
            previousRow = item.inventoryRow;
            previousEquipped = item.equipped;
        }
    }
    const std::uint64_t equippedSoid = nativeSlot < object.equippedInstanceSoids.size()
                                           ? object.equippedInstanceSoids[nativeSlot]
                                           : 0;
    const std::uint64_t requestedRowSoid = requestedRow < object.inventoryItems.size()
                                               ? object.inventoryItems[requestedRow].instanceSoid
                                               : 0;
    const std::uint64_t previousRowSoid = previousRow < object.inventoryItems.size()
                                              ? object.inventoryItems[previousRow].instanceSoid
                                              : 0;
    const std::int32_t requestedSerial = requestedRow < object.inventoryItems.size()
                                             ? object.inventoryItems[requestedRow].mutationSerial
                                             : -1;
    const std::int32_t previousSerial = previousRow < object.inventoryItems.size()
                                            ? object.inventoryItems[previousRow].mutationSerial
                                            : -1;
    std::array<char, core::log::kLineCapacity> line{};
    const int count = std::snprintf(
        line.data(),
        line.size(),
        "ev=equip stage=character_object result=ok family_version=%d root=0x%llX definition=%u "
        "character=0x%llX items=%zu next_serial=%u requested=0x%llX requested_row=%zu "
        "requested_row_soid=0x%llX requested_equipped=%u native_slot=%zu equipped_soid=0x%llX "
        "requested_serial=%d previous=0x%llX previous_row=%zu previous_row_soid=0x%llX "
        "previous_equipped=%u previous_serial=%d",
        swap.after.family4Version,
        static_cast<unsigned long long>(swap.after.family4RootSoid),
        swap.characterDefinitionId,
        static_cast<unsigned long long>(mutation.characterSoid),
        selected.loadout.itemCount,
        object.nextInventorySerial,
        static_cast<unsigned long long>(mutation.requestedInstanceSoid),
        requestedRow,
        static_cast<unsigned long long>(requestedRowSoid),
        static_cast<unsigned>(requestedEquipped),
        nativeSlot,
        static_cast<unsigned long long>(equippedSoid),
        requestedSerial,
        static_cast<unsigned long long>(mutation.previousInstanceSoid),
        previousRow,
        static_cast<unsigned long long>(previousRowSoid),
        static_cast<unsigned>(previousEquipped),
        previousSerial);
    if (count > 0) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::debug,
                         {line.data(), static_cast<std::size_t>(count)});
    }
}

} // namespace

/** Builds the Family-4 increment that moves the character object to the picked character. */
bool prepare_selection_move(Scratch& scratch,
                            const queuez::SelectCharacter& select,
                            Prepared& prepared) noexcept {
    const Reservation reservation = reserve_prior(scratch, prepared);
    if (reservation.rawWriteOffset > scratch.plaintext.size()
        || reservation.compressedWriteOffset > scratch.sealed.size()) {
        return report_failure("move_reservation");
    }
    const state::AccountState account = state::account_snapshot();
    const std::optional<std::size_t> selectedIndex = find_character_index(account);
    Resolved selected{};
    if (!state::account::valid(account) || !selectedIndex.has_value()
        || !resolve(account, *selectedIndex, selected)
        || account.characters[selected.characterIndex].soid != select.selectedCharacterSoid) {
        return report_failure("move_selection");
    }

    Prepared staged{};
    staged.rawClearSize = reservation.rawClearSize;
    staged.compressedClearSize = reservation.compressedClearSize;
    const auto rawStorage = std::span(scratch.plaintext).subspan(reservation.rawWriteOffset);
    std::size_t compressedExtent = reservation.compressedWriteOffset;
    std::size_t objectCount = 0;
    // The first select resends the character object even when the pick names the current one. With
    // no pending change all three operations go out. Only a pending change may skip it, and only
    // when the character did not move.
    if (!select.patchAccount || select.previousCharacterSoid != select.selectedCharacterSoid) {
        // A zero previous key means the snapshot published no character object. There is no slot
        // to release, so the pick adds one. Releasing a key the Client never held raises queuez
        // error 4.
        if (select.previousCharacterSoid != 0) {
            // The release carries no payload. The encoding field is never read for an empty
            // object, so it just matches its neighbours.
            staged.objects[objectCount++] = middleware::queuez::Object{
                select.characterDefinitionId,
                select.previousCharacterSoid,
                middleware::queuez::Encoding::oodle,
                {},
            };
        }
        if (family4_datagen::character::layout::kObjectSize > rawStorage.size()) {
            return report_failure("move_character_storage");
        }
        const auto characterBytes =
            rawStorage.first(family4_datagen::character::layout::kObjectSize);
        if (!family4_datagen::character::encode(account.characters[selected.characterIndex],
                                                selected.loadout,
                                                selected.lightEvaluation,
                                                characterBytes)) {
            return report_failure("move_character_object");
        }
        if (!append_object(scratch,
                           characterBytes,
                           select.characterDefinitionId,
                           select.selectedCharacterSoid,
                           staged.objects[objectCount++],
                           compressedExtent)) {
            return report_failure("move_character_object");
        }
        staged.rawClearSize = (std::max)(staged.rawClearSize,
                                         reservation.rawWriteOffset
                                             + family4_datagen::character::layout::kObjectSize);
    }

    // The character object is already compressed into sealed storage, so the raw span is free
    // for whichever account body this move carries.
    if (select.patchAccount) {
        std::size_t patchSize = 0;
        if (!selection_patch::encode(select.selectedCharacterSoid, rawStorage, patchSize)
            || patchSize != selection_patch::kPayloadSize) {
            return report_failure("move_account_patch");
        }
        staged.objects[objectCount++] = middleware::queuez::Object{
            select.accountDefinitionId,
            select.after.family4RootSoid,
            middleware::queuez::Encoding::tagReflection,
            rawStorage.first(patchSize),
        };
        staged.rawClearSize =
            (std::max)(staged.rawClearSize, reservation.rawWriteOffset + patchSize);
    } else {
        if (family4_datagen::account::layout::kObjectSize > rawStorage.size()) {
            return report_failure("move_account_storage");
        }
        const auto accountBytes = rawStorage.first(family4_datagen::account::layout::kObjectSize);
        if (!family4_datagen::account::encode(account, accountBytes)) {
            return report_failure("move_account_object");
        }
        if (!append_object(scratch,
                           accountBytes,
                           select.accountDefinitionId,
                           select.after.family4RootSoid,
                           staged.objects[objectCount++],
                           compressedExtent)) {
            return report_failure("move_account_object");
        }
        staged.rawClearSize =
            (std::max)(staged.rawClearSize,
                       reservation.rawWriteOffset + family4_datagen::account::layout::kObjectSize);
    }
    staged.compressedClearSize = (std::max)(reservation.compressedClearSize, compressedExtent);
    staged.family = middleware::queuez::Family{
        kAccountFamilyType,
        select.after.family4RootSoid,
        select.after.family4Version,
        0,
        std::span(staged.objects).first(objectCount),
    };
    if (!commit(staged, prepared)) {
        clear_after(scratch, reservation);
        return report_failure("move_commit");
    }
    return true;
}

/** Builds a single-character Family-4 upsert from an uncommitted equipment after-image. */
bool prepare_equipment_swap(Scratch& scratch,
                            const queuez::EquipmentSwap& swap,
                            const state::PendingEquipmentSwap& mutation,
                            Prepared& prepared) noexcept {
    const Reservation reservation = reserve_prior(scratch, prepared);
    if (reservation.rawWriteOffset > scratch.plaintext.size()
        || reservation.compressedWriteOffset > scratch.sealed.size()) {
        return report_failure("equip_reservation");
    }
    state::AccountState account = state::account_snapshot();
    if (!mutation.prepared || mutation.characterSoid == 0
        || mutation.characterSoid != swap.characterSoid
        || mutation.characterIndex >= account.characterCount
        || account.characters[mutation.characterIndex].soid != mutation.characterSoid) {
        return report_failure("equip_mutation");
    }
    account.characters[mutation.characterIndex] = mutation.afterCharacter;
    Resolved selected{};
    const std::optional<std::size_t> selectedIndex = find_character_index(account);
    if (!state::account::valid(account) || !selectedIndex.has_value()
        || *selectedIndex != mutation.characterIndex
        || !resolve(account, mutation.characterIndex, selected)
        || selected.characterObjectId != swap.characterDefinitionId) {
        return report_failure("equip_selection");
    }

    const auto rawStorage = std::span(scratch.plaintext).subspan(reservation.rawWriteOffset);
    if (family4_datagen::character::layout::kObjectSize > rawStorage.size()) {
        return report_failure("equip_character_storage");
    }
    const auto characterBytes = rawStorage.first(family4_datagen::character::layout::kObjectSize);
    if (!family4_datagen::character::encode(account.characters[mutation.characterIndex],
                                            selected.loadout,
                                            selected.lightEvaluation,
                                            characterBytes)) {
        return report_failure("equip_character_object");
    }
    report_equipment_object(swap,
                            mutation,
                            selected,
                            *reinterpret_cast<const family4_datagen::character::layout::Object*>(
                                characterBytes.data()));

    Prepared staged{};
    staged.rawClearSize =
        (std::max)(reservation.rawClearSize,
                   reservation.rawWriteOffset + family4_datagen::character::layout::kObjectSize);
    std::size_t compressedExtent = reservation.compressedWriteOffset;
    if (!append_object(scratch,
                       characterBytes,
                       swap.characterDefinitionId,
                       swap.characterSoid,
                       staged.objects.front(),
                       compressedExtent)) {
        return report_failure("equip_character_object");
    }
    staged.compressedClearSize = (std::max)(reservation.compressedClearSize, compressedExtent);
    staged.family = middleware::queuez::Family{
        kAccountFamilyType,
        swap.after.family4RootSoid,
        swap.after.family4Version,
        0,
        std::span(staged.objects).first(1),
    };
    if (!commit(staged, prepared)) {
        clear_after(scratch, reservation);
        return report_failure("equip_commit");
    }
    return true;
}

/** Builds a single-character Family-4 upsert from an uncommitted item-state after-image. */
bool prepare_item_state(Scratch& scratch,
                        const queuez::EquipmentSwap& update,
                        const state::PendingItemState& mutation,
                        Prepared& prepared) noexcept {
    const Reservation reservation = reserve_prior(scratch, prepared);
    if (reservation.rawWriteOffset > scratch.plaintext.size()
        || reservation.compressedWriteOffset > scratch.sealed.size()) {
        return report_failure("item_state_reservation");
    }
    state::AccountState account = state::account_snapshot();
    if (!mutation.prepared || mutation.characterSoid == 0
        || mutation.characterSoid != update.characterSoid
        || mutation.characterIndex >= account.characterCount
        || account.characters[mutation.characterIndex].soid != mutation.characterSoid) {
        return report_failure("item_state_mutation");
    }
    account.characters[mutation.characterIndex] = mutation.afterCharacter;
    Resolved selected{};
    const std::optional<std::size_t> selectedIndex = find_character_index(account);
    if (!state::account::valid(account) || !selectedIndex.has_value()
        || *selectedIndex != mutation.characterIndex
        || !resolve(account, mutation.characterIndex, selected)
        || selected.characterObjectId != update.characterDefinitionId) {
        return report_failure("item_state_selection");
    }

    const auto rawStorage = std::span(scratch.plaintext).subspan(reservation.rawWriteOffset);
    if (family4_datagen::character::layout::kObjectSize > rawStorage.size()) {
        return report_failure("item_state_character_storage");
    }
    const auto characterBytes = rawStorage.first(family4_datagen::character::layout::kObjectSize);
    if (!family4_datagen::character::encode(account.characters[mutation.characterIndex],
                                            selected.loadout,
                                            selected.lightEvaluation,
                                            characterBytes)) {
        return report_failure("item_state_character_object");
    }
    const auto& encoded =
        *reinterpret_cast<const family4_datagen::character::layout::Object*>(characterBytes.data());
    if (mutation.itemIndex >= mutation.afterCharacter.inventory.count && !mutation.targetEquipped) {
        return report_failure("item_state_inventory_index");
    }
    std::size_t matchingRows = 0;
    for (const auto& row : encoded.inventoryItems) {
        matchingRows +=
            static_cast<std::size_t>(row.instanceSoid == mutation.targetInstanceSoid
                                     && row.definitionIndex == mutation.targetDefinitionIndex
                                     && row.flags == mutation.afterFlags);
    }
    if (matchingRows != 1) {
        return report_failure("item_state_character_shape");
    }

    Prepared staged{};
    staged.rawClearSize =
        (std::max)(reservation.rawClearSize,
                   reservation.rawWriteOffset + family4_datagen::character::layout::kObjectSize);
    std::size_t compressedExtent = reservation.compressedWriteOffset;
    if (!append_object(scratch,
                       characterBytes,
                       update.characterDefinitionId,
                       update.characterSoid,
                       staged.objects.front(),
                       compressedExtent)) {
        return report_failure("item_state_character_object");
    }
    staged.compressedClearSize = (std::max)(reservation.compressedClearSize, compressedExtent);
    staged.family = middleware::queuez::Family{
        kAccountFamilyType,
        update.after.family4RootSoid,
        update.after.family4Version,
        0,
        std::span(staged.objects).first(1),
    };
    if (!commit(staged, prepared)) {
        clear_after(scratch, reservation);
        return report_failure("item_state_commit");
    }
    return true;
}

/** Builds a resident item upsert followed by charged account balances when the cost consumes. */
bool prepare_socket_plug(Scratch& scratch,
                         const queuez::SocketPlug& socketPlug,
                         const state::PendingSocketPlug& mutation,
                         Prepared& prepared) noexcept {
    const Reservation reservation = reserve_prior(scratch, prepared);
    if (reservation.rawWriteOffset > scratch.plaintext.size()
        || reservation.compressedWriteOffset > scratch.sealed.size()) {
        return report_failure("socket_plug_reservation");
    }

    state::AccountState account{};
    if (!mutation.prepared || mutation.accountSoid == 0 || mutation.characterSoid == 0
        || mutation.targetInstanceSoid == 0 || mutation.accountSoid != socketPlug.accountSoid
        || mutation.characterSoid != socketPlug.characterSoid
        || mutation.targetInstanceSoid != socketPlug.targetInstanceSoid
        || mutation.profileChanged != socketPlug.updatesAccount
        || mutation.accountSoid != socketPlug.after.family4RootSoid
        || socketPlug.accountDefinitionId == 0 || !state::preview_socket_plug(mutation, account)
        || mutation.characterIndex >= account.characterCount
        || account.primarySoid != mutation.accountSoid
        || account.characters[mutation.characterIndex].soid != mutation.characterSoid) {
        return report_failure("socket_plug_mutation");
    }

    Resolved selected{};
    const std::optional<std::size_t> selectedIndex = find_character_index(account);
    if (!state::account::valid(account) || !selectedIndex.has_value()
        || *selectedIndex != mutation.characterIndex
        || !resolve(account, mutation.characterIndex, selected)
        || selected.itemInstanceObjectId != socketPlug.itemInstanceDefinitionId) {
        return report_failure("socket_plug_selection");
    }

    family4_datagen::loadout::ResolvedInstances changed{};
    for (std::size_t index = 0; index < selected.loadout.itemCount; ++index) {
        const family4_datagen::loadout::ResolvedItem& item = selected.loadout.items[index];
        if (item.instance.instanceSoid != mutation.targetInstanceSoid) {
            continue;
        }
        if (changed.itemCount != 0
            || item.instance.baseDefinitionIndex != mutation.targetDefinitionIndex
            || mutation.socketLane >= item.instance.ordinarySockets.plugs.size()
            || item.instance.ordinarySockets.state
                   != family4_datagen::instance::OrdinarySocketBlockState::present
            || !item.instance.ordinarySockets.plugs[mutation.socketLane].has_value()
            || *item.instance.ordinarySockets.plugs[mutation.socketLane]
                   != mutation.plugDefinitionIndex) {
            return report_failure("socket_plug_item_shape");
        }
        changed.items[0] = {item.equipmentSlot, item.instance};
        changed.itemCount = 1;
    }
    if (changed.itemCount != 1) {
        return report_failure("socket_plug_item_missing");
    }

    const auto rawStorage = std::span(scratch.plaintext).subspan(reservation.rawWriteOffset);
    const std::size_t requiredRawSize = socketPlug.updatesAccount
                                            ? family4_datagen::account::layout::kObjectSize
                                            : family4_datagen::instance::layout::kObjectSize;
    if (requiredRawSize > rawStorage.size()) {
        return report_failure("socket_plug_item_storage");
    }
    Prepared staged{};
    staged.rawClearSize =
        (std::max)(reservation.rawClearSize,
                   reservation.rawWriteOffset + family4_datagen::instance::layout::kObjectSize);
    std::size_t compressedExtent = reservation.compressedWriteOffset;
    std::size_t itemCursor = 0;
    if (!append_items(scratch,
                      rawStorage,
                      socketPlug.itemInstanceDefinitionId,
                      changed,
                      0,
                      staged,
                      itemCursor,
                      compressedExtent)
        || itemCursor != 1) {
        clear_after(scratch, reservation);
        return report_failure("socket_plug_item_object");
    }

    std::size_t objectCount = 1;
    if (socketPlug.updatesAccount) {
        const auto accountBytes = rawStorage.first(family4_datagen::account::layout::kObjectSize);
        if (!family4_datagen::account::encode(account, accountBytes)
            || !append_object(scratch,
                              accountBytes,
                              socketPlug.accountDefinitionId,
                              socketPlug.accountSoid,
                              staged.objects[1],
                              compressedExtent)) {
            clear_after(scratch, reservation);
            return report_failure("socket_plug_account_object");
        }
        staged.rawClearSize =
            (std::max)(staged.rawClearSize,
                       reservation.rawWriteOffset + family4_datagen::account::layout::kObjectSize);
        objectCount = 2;
    }

    staged.compressedClearSize = (std::max)(reservation.compressedClearSize, compressedExtent);
    staged.family = middleware::queuez::Family{
        kAccountFamilyType,
        socketPlug.after.family4RootSoid,
        socketPlug.after.family4Version,
        0,
        std::span(staged.objects).first(objectCount),
    };
    if (!commit(staged, prepared)) {
        clear_after(scratch, reservation);
        return report_failure("socket_plug_commit");
    }

    std::array<char, core::log::kLineCapacity> line{};
    const int count = std::snprintf(
        line.data(),
        line.size(),
        "ev=socket_plug stage=family4_objects result=ok family_version=%d root=0x%llX "
        "character=0x%llX instance=0x%llX item_definition=%u target_definition=%u "
        "target_bucket=%u lane=%u plug_definition=%u plug_bucket=%u equipped=%u "
        "material_set=%u material_set_hash=%u material_rows=%u account_update=%u objects=%zu "
        "order=%s",
        socketPlug.after.family4Version,
        static_cast<unsigned long long>(socketPlug.after.family4RootSoid),
        static_cast<unsigned long long>(socketPlug.characterSoid),
        static_cast<unsigned long long>(socketPlug.targetInstanceSoid),
        socketPlug.itemInstanceDefinitionId,
        static_cast<unsigned>(mutation.targetDefinitionIndex),
        static_cast<unsigned>(mutation.targetBucketId),
        static_cast<unsigned>(mutation.socketLane),
        static_cast<unsigned>(mutation.plugDefinitionIndex),
        static_cast<unsigned>(mutation.plugBucketId),
        static_cast<unsigned>(mutation.targetEquipped),
        static_cast<unsigned>(mutation.materialRequirementSetIndex),
        mutation.materialRequirementSetHash,
        static_cast<unsigned>(mutation.materialRequirementCount),
        static_cast<unsigned>(socketPlug.updatesAccount),
        objectCount,
        socketPlug.updatesAccount ? "item_account" : "item");
    if (count > 0) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::debug,
                         {line.data(), static_cast<std::size_t>(count)});
    }
    return true;
}

/** Builds a single full account-object upsert from an uncommitted profile-stack after-image. */
bool prepare_profile_item_acquisition(Scratch& scratch,
                                      const queuez::ProfileItemAcquisition& acquisition,
                                      const state::PendingProfileItemAcquisition& mutation,
                                      Prepared& prepared) noexcept {
    const Reservation reservation = reserve_prior(scratch, prepared);
    if (reservation.rawWriteOffset > scratch.plaintext.size()
        || reservation.compressedWriteOffset > scratch.sealed.size()) {
        return report_failure("profile_acquire_reservation");
    }
    if (!mutation.prepared || mutation.accountSoid == 0
        || mutation.accountSoid != acquisition.accountSoid
        || mutation.acquiredInstanceSoid != acquisition.acquiredInstanceSoid
        || mutation.actionSource != acquisition.actionSource
        || acquisition.appendedResident != (mutation.appended && mutation.actionSource)
        || acquisition.accountSoid != acquisition.after.family4RootSoid
        || acquisition.accountDefinitionId == 0
        || (acquisition.appendedResident && acquisition.itemInstanceDefinitionId == 0)) {
        return report_failure("profile_acquire_mutation");
    }
    state::AccountState account{};
    if (!state::preview_profile_item_acquisition(mutation, account)
        || account.primarySoid != acquisition.accountSoid || !state::account::valid(account)) {
        return report_failure("profile_acquire_account");
    }

    const auto rawStorage = std::span(scratch.plaintext).subspan(reservation.rawWriteOffset);
    if (family4_datagen::account::layout::kObjectSize > rawStorage.size()) {
        return report_failure("profile_acquire_account_storage");
    }
    const auto accountBytes = rawStorage.first(family4_datagen::account::layout::kObjectSize);
    if (!family4_datagen::account::encode(account, accountBytes)) {
        return report_failure("profile_acquire_account_encode");
    }

    // The native account observer compares profile quantities but only emits pickup feedback when
    // the changed row's mutation serial also appears in this transient 16-record bank at 0x6978.
    // Keep the descriptor local to this one incremental upsert; ordinary snapshots encode an empty
    // bank, while the row's rising mutation serial remains persistent State.
    constexpr std::uint16_t kAcquisitionChangeSequence = 0;
    constexpr std::uint16_t kAcquisitionChangeNextWriteSlot = 1;
    constexpr std::uint16_t kAcquisitionChangeNextSequence = 1;
    constexpr std::uint8_t kAcquisitionChangeKind = 1;
    constexpr std::uint16_t kAcquisitionChangeFlags = 0;
    auto& accountObject =
        *reinterpret_cast<family4_datagen::account::layout::Object*>(accountBytes.data());
    std::size_t acquiredRow = accountObject.profileItems.size();
    for (std::size_t row = 0; row < accountObject.profileItems.size(); ++row) {
        const auto& inventoryRow = accountObject.profileItems[row];
        if (inventoryRow.mutationSerial != mutation.acquiredMutationSerial) {
            continue;
        }
        if (acquiredRow != accountObject.profileItems.size()) {
            clear_after(scratch, reservation);
            return report_failure("profile_acquire_row_duplicate");
        }
        acquiredRow = row;
    }
    const auto recordIsZero =
        [](const family4_datagen::account::layout::ProfileInventoryChangeRecord& record) noexcept {
            return record.sequence == 0 && record.reserved == 0 && record.mutationSerial == 0
                   && record.kind == 0 && record.reservedKind == 0 && record.flags == 0;
        };
    const bool recordsAreZero = std::all_of(accountObject.profileInventoryChanges.records.cbegin(),
                                            accountObject.profileInventoryChanges.records.cend(),
                                            recordIsZero);
    if (acquiredRow >= accountObject.profileItems.size()
        || accountObject.profileItems[acquiredRow].quantity != mutation.acquiredQuantity
        || accountObject.profileInventoryChanges.writeSlot != 0
        || accountObject.profileInventoryChanges.nextSequence != 0 || !recordsAreZero) {
        clear_after(scratch, reservation);
        return report_failure("profile_acquire_inventory_change_state");
    }
    accountObject.profileInventoryChanges.writeSlot = kAcquisitionChangeNextWriteSlot;
    accountObject.profileInventoryChanges.nextSequence = kAcquisitionChangeNextSequence;
    auto& acquisitionChange = accountObject.profileInventoryChanges.records.front();
    acquisitionChange.sequence = kAcquisitionChangeSequence;
    acquisitionChange.mutationSerial = mutation.acquiredMutationSerial;
    acquisitionChange.kind = kAcquisitionChangeKind;
    acquisitionChange.flags = kAcquisitionChangeFlags;

    Prepared staged{};
    staged.rawClearSize =
        (std::max)(reservation.rawClearSize,
                   reservation.rawWriteOffset + family4_datagen::account::layout::kObjectSize);
    std::size_t compressedExtent = reservation.compressedWriteOffset;
    const std::size_t accountObjectIndex = acquisition.appendedResident ? 1U : 0U;
    if (!append_object(scratch,
                       accountBytes,
                       acquisition.accountDefinitionId,
                       acquisition.accountSoid,
                       staged.objects[accountObjectIndex],
                       compressedExtent)) {
        return report_failure("profile_acquire_account_object");
    }
    std::size_t objectCount = 1;
    if (acquisition.appendedResident) {
        if (mutation.profileIndex >= account.profileItemCount) {
            return report_failure("profile_acquire_instance_index");
        }
        family4_datagen::instance::ResolvedInstance instance{};
        const state::account::inventory::ProfileItem& profileItem =
            account.profileItems[mutation.profileIndex];
        const auto instanceBytes = rawStorage.first(family4_datagen::instance::layout::kObjectSize);
        if (!resolve_profile_item_instance(profileItem, instance)
            || instance.instanceSoid != acquisition.acquiredInstanceSoid
            || !family4_datagen::instance::encode(instance, instanceBytes)
            || !append_object(scratch,
                              instanceBytes,
                              acquisition.itemInstanceDefinitionId,
                              acquisition.acquiredInstanceSoid,
                              staged.objects.front(),
                              compressedExtent)) {
            return report_failure("profile_acquire_instance_object");
        }
        objectCount = 2;
    }
    staged.compressedClearSize = (std::max)(reservation.compressedClearSize, compressedExtent);
    staged.family = middleware::queuez::Family{
        kAccountFamilyType,
        acquisition.after.family4RootSoid,
        acquisition.after.family4Version,
        0,
        std::span(staged.objects).first(objectCount),
    };
    if (!commit(staged, prepared)) {
        clear_after(scratch, reservation);
        return report_failure("profile_acquire_commit");
    }

    std::array<char, core::log::kLineCapacity> line{};
    const int count =
        std::snprintf(line.data(),
                      line.size(),
                      "ev=profile_acquire stage=account_object result=ok family_version=%d "
                      "account=0x%llX definition=%u item_count=%zu definition_hash=%u quantity=%d "
                      "native_row=%zu mutation_serial=%d change_slot=%u change_next_sequence=%u "
                      "change_kind=%u account_payload_bytes=%zu objects=%zu object_order=%s",
                      acquisition.after.family4Version,
                      static_cast<unsigned long long>(acquisition.accountSoid),
                      acquisition.accountDefinitionId,
                      mutation.afterItemCount,
                      mutation.acquiredDefinitionHash,
                      mutation.acquiredQuantity,
                      acquiredRow,
                      mutation.acquiredMutationSerial,
                      static_cast<unsigned>(kAcquisitionChangeNextWriteSlot),
                      static_cast<unsigned>(kAcquisitionChangeNextSequence),
                      static_cast<unsigned>(kAcquisitionChangeKind),
                      prepared.family.objects[accountObjectIndex].payload.size(),
                      objectCount,
                      acquisition.appendedResident ? "item-account" : "account");
    if (count > 0) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::debug,
                         {line.data(), static_cast<std::size_t>(count)});
    }
    return true;
}

/** Builds a new item-instance upsert before its character after-image. */
bool prepare_item_acquisition(Scratch& scratch,
                              const queuez::ItemAcquisition& acquisition,
                              const state::PendingItemAcquisition& mutation,
                              Prepared& prepared) noexcept {
    const Reservation reservation = reserve_prior(scratch, prepared);
    if (reservation.rawWriteOffset > scratch.plaintext.size()
        || reservation.compressedWriteOffset > scratch.sealed.size()) {
        return report_failure("acquire_reservation");
    }

    state::AccountState account{};
    if (!mutation.prepared || mutation.characterSoid == 0 || mutation.acquiredInstanceSoid == 0
        || mutation.accountSoid == 0 || mutation.accountSoid != acquisition.accountSoid
        || mutation.characterSoid != acquisition.characterSoid
        || mutation.acquiredInstanceSoid != acquisition.acquiredInstanceSoid
        || mutation.profileChanged != acquisition.updatesAccount
        || acquisition.accountSoid != acquisition.after.family4RootSoid
        || acquisition.accountDefinitionId == 0 || acquisition.after.family4ResidentCount == 0
        || acquisition.after.family4Residents[acquisition.after.family4ResidentCount - 1U]
                   .objectSoid
               != mutation.acquiredInstanceSoid
        || acquisition.after.family4Residents[acquisition.after.family4ResidentCount - 1U]
                   .definitionId
               != acquisition.itemInstanceDefinitionId) {
        return report_failure("acquire_mutation");
    }
    if (!state::preview_item_acquisition(mutation, account)
        || mutation.characterIndex >= account.characterCount
        || account.primarySoid != acquisition.accountSoid
        || account.characters[mutation.characterIndex].soid != mutation.characterSoid) {
        return report_failure("acquire_account");
    }

    Resolved selected{};
    const std::optional<std::size_t> selectedIndex = find_character_index(account);
    if (!state::account::valid(account) || !selectedIndex.has_value()
        || *selectedIndex != mutation.characterIndex
        || !resolve(account, mutation.characterIndex, selected)
        || selected.characterObjectId != acquisition.characterDefinitionId
        || selected.itemInstanceObjectId != acquisition.itemInstanceDefinitionId) {
        return report_failure("acquire_selection");
    }

    family4_datagen::loadout::ResolvedInstances acquired{};
    std::int32_t acquiredMutationSerial = -1;
    for (std::size_t index = 0; index < selected.loadout.itemCount; ++index) {
        const family4_datagen::loadout::ResolvedItem& item = selected.loadout.items[index];
        if (item.instance.instanceSoid != mutation.acquiredInstanceSoid) {
            continue;
        }
        if (acquired.itemCount != 0 || item.equipped || item.inventoryRow != mutation.inventoryRow
            || item.equipmentSlot != mutation.equipmentSlot) {
            return report_failure("acquire_item_row");
        }
        acquired.items[0] = {item.equipmentSlot, item.instance};
        acquired.itemCount = 1;
        acquiredMutationSerial = item.mutationSerial;
    }
    if (acquired.itemCount != 1 || acquiredMutationSerial < 0) {
        return report_failure("acquire_item_missing");
    }

    const auto rawStorage = std::span(scratch.plaintext).subspan(reservation.rawWriteOffset);
    if (family4_datagen::character::layout::kObjectSize > rawStorage.size()) {
        return report_failure("acquire_character_storage");
    }
    const auto characterBytes = rawStorage.first(family4_datagen::character::layout::kObjectSize);
    if (!family4_datagen::character::encode(account.characters[mutation.characterIndex],
                                            selected.loadout,
                                            selected.lightEvaluation,
                                            characterBytes)) {
        return report_failure("acquire_character_object");
    }

    // The native character-object observer requires a transient inventory-change record whose
    // serial matches the newly filled row. Without it the row is accepted, but the observer never
    // queues the acquisition feedback. Keep this record local to this one acquisition push; the
    // canonical encoder leaves the bank empty on every later snapshot.
    constexpr std::size_t kBitsPerFlagByte = 8;
    constexpr std::int32_t kEncodedOccupiedRowWatermark = 1;
    constexpr std::uint16_t kAcquisitionChangeSequence = 0;
    constexpr std::uint16_t kAcquisitionChangeNextWriteSlot = 1;
    constexpr std::uint16_t kAcquisitionChangeNextSequence = 1;
    constexpr std::uint8_t kAcquisitionChangeKind = 1;
    constexpr std::uint16_t kAcquisitionChangeFlags = 0;
    auto& characterObject =
        *reinterpret_cast<family4_datagen::character::layout::Object*>(characterBytes.data());
    const std::size_t acquiredRow = mutation.inventoryRow;
    if (acquiredRow >= characterObject.inventoryItems.size()
        || acquiredRow >= characterObject.instanceProgressWatermarks.size()) {
        clear_after(scratch, reservation);
        return report_failure("acquire_new_item_row");
    }
    const std::size_t newItemFlagIndex = acquiredRow / kBitsPerFlagByte;
    const std::byte newItemFlagMask = std::byte{1U} << (acquiredRow % kBitsPerFlagByte);
    const auto recordIsZero =
        [](const family4_datagen::character::layout::InventoryChangeRecord& record) noexcept {
            return record.sequence == 0 && record.reserved == 0 && record.mutationSerial == 0
                   && record.kind == 0 && record.reservedKind == 0 && record.flags == 0;
        };
    const bool unknownIsZero =
        std::all_of(characterObject.inventoryChangeUnknown.cbegin(),
                    characterObject.inventoryChangeUnknown.cend(),
                    [](std::byte value) noexcept { return value == std::byte{}; });
    const bool recordsAreZero = std::all_of(characterObject.inventoryChanges.records.cbegin(),
                                            characterObject.inventoryChanges.records.cend(),
                                            recordIsZero);
    const auto& acquiredInventoryRow = characterObject.inventoryItems[acquiredRow];
    if (newItemFlagIndex >= characterObject.newItemFlags.size()
        || acquiredInventoryRow.instanceSoid != mutation.acquiredInstanceSoid
        || acquiredInventoryRow.mutationSerial != acquiredMutationSerial
        || (characterObject.newItemFlags[newItemFlagIndex] & newItemFlagMask) != newItemFlagMask
        || characterObject.instanceProgressWatermarks[acquiredRow] != kEncodedOccupiedRowWatermark
        || !unknownIsZero || characterObject.inventoryChanges.writeSlot != 0
        || characterObject.inventoryChanges.nextSequence != 0 || !recordsAreZero) {
        clear_after(scratch, reservation);
        return report_failure("acquire_inventory_change_state");
    }
    characterObject.inventoryChanges.writeSlot = kAcquisitionChangeNextWriteSlot;
    characterObject.inventoryChanges.nextSequence = kAcquisitionChangeNextSequence;
    auto& acquisitionChange = characterObject.inventoryChanges.records.front();
    acquisitionChange.sequence = kAcquisitionChangeSequence;
    acquisitionChange.mutationSerial = acquiredMutationSerial;
    acquisitionChange.kind = kAcquisitionChangeKind;
    acquisitionChange.flags = kAcquisitionChangeFlags;
    if (!std::all_of(characterObject.inventoryChanges.records.cbegin() + 1,
                     characterObject.inventoryChanges.records.cend(),
                     recordIsZero)) {
        clear_after(scratch, reservation);
        return report_failure("acquire_inventory_change_records");
    }

    Prepared staged{};
    staged.rawClearSize =
        (std::max)(reservation.rawClearSize,
                   reservation.rawWriteOffset + family4_datagen::character::layout::kObjectSize);
    std::size_t compressedExtent = reservation.compressedWriteOffset;
    if (!append_object(scratch,
                       characterBytes,
                       acquisition.characterDefinitionId,
                       acquisition.characterSoid,
                       staged.objects[0],
                       compressedExtent)) {
        return report_failure("acquire_character_object");
    }

    std::size_t itemCursor = 0;
    if (!append_items(scratch,
                      rawStorage,
                      acquisition.itemInstanceDefinitionId,
                      acquired,
                      1,
                      staged,
                      itemCursor,
                      compressedExtent)
        || itemCursor != 1) {
        clear_after(scratch, reservation);
        return report_failure("acquire_item_object");
    }

    const auto& acquiredObject =
        *reinterpret_cast<const family4_datagen::instance::layout::Object*>(rawStorage.data());
    if (acquiredObject.instanceSoid != mutation.acquiredInstanceSoid
        || acquiredObject.roll.progress
               != family4_datagen::instance::layout::kInitialInstanceProgress) {
        clear_after(scratch, reservation);
        return report_failure("acquire_item_progress");
    }
    const std::int32_t acquiredInstanceProgress = acquiredObject.roll.progress;

    std::size_t objectCount = 2;
    if (acquisition.updatesAccount) {
        if (family4_datagen::account::layout::kObjectSize > rawStorage.size()) {
            clear_after(scratch, reservation);
            return report_failure("acquire_account_storage");
        }
        const auto accountBytes = rawStorage.first(family4_datagen::account::layout::kObjectSize);
        if (!family4_datagen::account::encode(account, accountBytes)
            || !append_object(scratch,
                              accountBytes,
                              acquisition.accountDefinitionId,
                              acquisition.accountSoid,
                              staged.objects[2],
                              compressedExtent)) {
            clear_after(scratch, reservation);
            return report_failure("acquire_account_object");
        }
        staged.rawClearSize =
            (std::max)(staged.rawClearSize,
                       reservation.rawWriteOffset + family4_datagen::account::layout::kObjectSize);
        objectCount = 3;
    }

    // Creation increments publish the dependency before the reference to it. Compression order is
    // irrelevant because each descriptor already owns its sealed span, so exchange only the wire
    // descriptors: new item first, then the character after-image. Dismantle deliberately uses the
    // inverse dependency order (drop the character reference, then release the item).
    std::swap(staged.objects[0], staged.objects[1]);

    staged.compressedClearSize = (std::max)(reservation.compressedClearSize, compressedExtent);
    staged.family = middleware::queuez::Family{
        kAccountFamilyType,
        acquisition.after.family4RootSoid,
        acquisition.after.family4Version,
        0,
        std::span(staged.objects).first(objectCount),
    };
    if (!commit(staged, prepared)) {
        clear_after(scratch, reservation);
        return report_failure("acquire_commit");
    }

    std::array<char, core::log::kLineCapacity> line{};
    const int count = std::snprintf(
        line.data(),
        line.size(),
        "ev=acquire stage=family4_objects result=ok family_version=%d root=0x%llX "
        "character=0x%llX character_definition=%u instance=0x%llX item_definition=%u "
        "definition_hash=%u inventory_row=%u equipment_slot=%u next_serial=%u objects=%zu "
        "order=%s new_item_flag=1 watermark=1 acquired_row_serial=%d "
        "inventory_change_write_slot=%u inventory_change_next_sequence=%u "
        "inventory_change_record=0 inventory_change_sequence=%u "
        "inventory_change_serial=%d inventory_change_kind=%u inventory_change_flags=%u "
        "instance_progress=%d",
        acquisition.after.family4Version,
        static_cast<unsigned long long>(acquisition.after.family4RootSoid),
        static_cast<unsigned long long>(acquisition.characterSoid),
        acquisition.characterDefinitionId,
        static_cast<unsigned long long>(acquisition.acquiredInstanceSoid),
        acquisition.itemInstanceDefinitionId,
        mutation.acquiredDefinitionHash,
        static_cast<unsigned>(mutation.inventoryRow),
        static_cast<unsigned>(mutation.equipmentSlot),
        mutation.afterCharacter.nextInventorySerial,
        objectCount,
        acquisition.updatesAccount ? "item_character_account" : "item_character",
        acquiredMutationSerial,
        static_cast<unsigned>(kAcquisitionChangeNextWriteSlot),
        static_cast<unsigned>(kAcquisitionChangeNextSequence),
        static_cast<unsigned>(kAcquisitionChangeSequence),
        acquiredMutationSerial,
        static_cast<unsigned>(kAcquisitionChangeKind),
        static_cast<unsigned>(kAcquisitionChangeFlags),
        acquiredInstanceProgress);
    if (count > 0) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::debug,
                         {line.data(), static_cast<std::size_t>(count)});
    }
    return true;
}

/** Builds a character upsert followed by one empty item-instance release descriptor. */
bool prepare_item_dismantle(Scratch& scratch,
                            const queuez::ItemDismantle& dismantle,
                            const state::PendingItemDismantle& mutation,
                            Prepared& prepared) noexcept {
    const Reservation reservation = reserve_prior(scratch, prepared);
    if (reservation.rawWriteOffset > scratch.plaintext.size()
        || reservation.compressedWriteOffset > scratch.sealed.size()) {
        return report_failure("dismantle_reservation");
    }

    state::AccountState account = state::account_snapshot();
    if (!mutation.prepared || mutation.characterSoid == 0 || mutation.dismantledInstanceSoid == 0
        || mutation.dismantledItem.instanceSoid != mutation.dismantledInstanceSoid
        || mutation.characterSoid != dismantle.characterSoid
        || mutation.dismantledInstanceSoid != dismantle.dismantledInstanceSoid
        || mutation.characterIndex >= account.characterCount
        || account.characters[mutation.characterIndex].soid != mutation.characterSoid) {
        return report_failure("dismantle_mutation");
    }
    account.characters[mutation.characterIndex] = mutation.afterCharacter;

    Resolved selected{};
    const std::optional<std::size_t> selectedIndex = find_character_index(account);
    if (!state::account::valid(account) || !selectedIndex.has_value()
        || *selectedIndex != mutation.characterIndex
        || !resolve(account, mutation.characterIndex, selected)
        || selected.characterObjectId != dismantle.characterDefinitionId
        || selected.itemInstanceObjectId != dismantle.itemInstanceDefinitionId) {
        return report_failure("dismantle_selection");
    }
    for (std::size_t index = 0; index < selected.loadout.itemCount; ++index) {
        if (selected.loadout.items[index].instance.instanceSoid
            == mutation.dismantledInstanceSoid) {
            return report_failure("dismantle_item_present");
        }
    }

    const auto rawStorage = std::span(scratch.plaintext).subspan(reservation.rawWriteOffset);
    if (family4_datagen::character::layout::kObjectSize > rawStorage.size()) {
        return report_failure("dismantle_character_storage");
    }
    const auto characterBytes = rawStorage.first(family4_datagen::character::layout::kObjectSize);
    if (!family4_datagen::character::encode(account.characters[mutation.characterIndex],
                                            selected.loadout,
                                            selected.lightEvaluation,
                                            characterBytes)) {
        return report_failure("dismantle_character_object");
    }

    Prepared staged{};
    staged.rawClearSize =
        (std::max)(reservation.rawClearSize,
                   reservation.rawWriteOffset + family4_datagen::character::layout::kObjectSize);
    std::size_t compressedExtent = reservation.compressedWriteOffset;
    if (!append_object(scratch,
                       characterBytes,
                       dismantle.characterDefinitionId,
                       dismantle.characterSoid,
                       staged.objects[0],
                       compressedExtent)) {
        return report_failure("dismantle_character_object");
    }
    // Queuez represents a release with the ordinary object key and an empty payload. The
    // encoding selector is not read for empty descriptors; oodle matches the surrounding objects.
    staged.objects[1] = middleware::queuez::Object{
        dismantle.itemInstanceDefinitionId,
        dismantle.dismantledInstanceSoid,
        middleware::queuez::Encoding::oodle,
        {},
    };
    staged.compressedClearSize = (std::max)(reservation.compressedClearSize, compressedExtent);
    staged.family = middleware::queuez::Family{
        kAccountFamilyType,
        dismantle.after.family4RootSoid,
        dismantle.after.family4Version,
        0,
        std::span(staged.objects).first(2),
    };
    if (!commit(staged, prepared)) {
        clear_after(scratch, reservation);
        return report_failure("dismantle_commit");
    }

    std::array<char, core::log::kLineCapacity> line{};
    const int count =
        std::snprintf(line.data(),
                      line.size(),
                      "ev=dismantle stage=family4_objects result=ok family_version=%d root=0x%llX "
                      "character=0x%llX character_definition=%u instance=0x%llX item_definition=%u "
                      "definition_hash=%u inventory_index=%zu inventory_row=%u equipment_slot=%u "
                      "moved_items=%zu items_after=%zu next_serial=%u",
                      dismantle.after.family4Version,
                      static_cast<unsigned long long>(dismantle.after.family4RootSoid),
                      static_cast<unsigned long long>(dismantle.characterSoid),
                      dismantle.characterDefinitionId,
                      static_cast<unsigned long long>(dismantle.dismantledInstanceSoid),
                      dismantle.itemInstanceDefinitionId,
                      mutation.dismantledItem.definitionHash,
                      mutation.inventoryIndex,
                      static_cast<unsigned>(mutation.inventoryRow),
                      static_cast<unsigned>(mutation.equipmentSlot),
                      mutation.movedInventoryItemCount,
                      mutation.afterCharacter.inventory.count,
                      mutation.afterCharacter.nextInventorySerial);
    if (count > 0) {
        core::log::write(core::log::Channel::server,
                         core::log::Level::debug,
                         {line.data(), static_cast<std::size_t>(count)});
    }
    return true;
}

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