#include <mln/particle/field/field_profile_validator.hpp>

#include <gtest/gtest.h>

#include <limits>
#include <memory>
#include <string>
#include <vector>

using namespace mln;

namespace {

particle::DecodedFieldPlane plane(uint16_t semantic, particle::FieldValueClass valueClass, uint16_t unit,
                                  uint16_t group, particle::FieldComponent component, particle::FieldDType dtype,
                                  bool required, std::shared_ptr<const std::vector<uint8_t>> storage,
                                  std::string identity = "plane") {
    return {.key = {semantic, group, component}, .dtype = dtype, .valueClass = valueClass, .unitID = unit,
            .requiredFlag = required, .count = 4, .scale = 0.1f, .offset = 0.0f,
            .contentIdentity = std::move(identity), .storage = std::move(storage)};
}

std::shared_ptr<const std::vector<uint8_t>> fourLittleEndian16BitSamples() {
    return std::make_shared<const std::vector<uint8_t>>(std::vector<uint8_t>{1, 0, 2, 0, 3, 0, 4, 0});
}

particle::DecodedFieldFrame frame() {
    const auto numbers = std::make_shared<const std::vector<uint8_t>>(std::vector<uint8_t>{1, 2, 3, 4});
    const auto bits = std::make_shared<const std::vector<uint8_t>>(std::vector<uint8_t>{0x0F});
    particle::DecodedFieldFrame value;
    value.frameContentIdentity = {"future-product", 1, 1, 1'000};
    value.referenceTimeMs = 1'000;
    value.grid = {.layout = 1, .projection = 1, .sampleOrder = 1, .gridRegistration = 1, .width = 2, .height = 2,
                  .west = 130.0, .south = 30.0, .east = 131.0, .north = 31.0};
    value.validityContentIdentity = "validity";
    value.validityBits = bits;
    value.validityPlane = plane(900, particle::FieldValueClass::Validity, particle::FieldUnitNone, 0,
                                particle::FieldComponent::None, particle::FieldDType::Bitset, true, bits, "validity");
    value.planes.push_back(plane(777, particle::FieldValueClass::Scalar, 65432, 0, particle::FieldComponent::None,
                                 particle::FieldDType::UInt8, true, numbers, "scalar"));
    return value;
}

particle::ResolvedFieldProfileContract contract() {
    particle::ResolvedFieldProfileContract value;
    value.profileID = "future.product-42";
    value.profileVersion = 1;
    value.allowAdditionalPlanes = true;
    value.requiredPlanes = {{777, particle::FieldValueClass::Scalar, 65432, 0, particle::FieldComponent::None, true},
                            {900, particle::FieldValueClass::Validity, particle::FieldUnitNone, 0,
                             particle::FieldComponent::None, true}};
    return value;
}

particle::FieldProfileValidationCode validate(const particle::ResolvedFieldProfileContract& c,
                                               const particle::DecodedFieldFrame& f) {
    return particle::FieldProfileValidator::validate(c, f).code;
}

std::shared_ptr<const particle::DecodedFieldFrame> runtimeFrame(uint64_t time, uint64_t generation,
                                                                 uint64_t version, std::string content,
                                                                 std::string id = "field") {
    auto value = frame();
    value.frameContentIdentity = {std::move(id), generation, version, time};
    value.planes.front().contentIdentity = std::move(content);
    return std::make_shared<const particle::DecodedFieldFrame>(std::move(value));
}

} // namespace

TEST(ParticleFieldProfileValidator, AcceptsGenericProfileAndOpaqueScalarUnit) {
    EXPECT_EQ(validate(contract(), frame()), particle::FieldProfileValidationCode::Accepted);
}

TEST(ParticleFieldProfileValidator, EnforcesRequiredPlanesAndAdditionalPlanePolicy) {
    auto c = contract(); auto f = frame();
    f.planes.clear();
    EXPECT_TRUE(f.isStructurallyValid());
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::MissingRequiredPlane);
    f = frame(); f.planes.front().unitID = 99;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::PlaneDescriptorMismatch);
    f = frame(); f.planes.front().requiredFlag = false;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::PlaneDescriptorMismatch);
    f = frame(); f.planes.push_back(plane(778, particle::FieldValueClass::Scalar, 123, 0, particle::FieldComponent::None,
                                           particle::FieldDType::UInt8, true, f.planes.front().storage, "required-extra"));
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::UndeclaredRequiredPlane);
    f.planes.back().requiredFlag = false; c.allowAdditionalPlanes = false;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::AdditionalPlaneNotAllowed);
    c.allowAdditionalPlanes = true;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::Accepted);
}

TEST(ParticleFieldProfileValidator, EnforcesSingleSharedValidityStructuralInvariant) {
    auto f = frame();
    const auto numbers = f.planes.front().storage;

    f.planes.push_back(plane(901, particle::FieldValueClass::Validity, particle::FieldUnitNone, 0,
                             particle::FieldComponent::None, particle::FieldDType::Bitset, false, numbers, "second-validity"));
    EXPECT_FALSE(f.isStructurallyValid());
    EXPECT_EQ(validate(contract(), f), particle::FieldProfileValidationCode::InvalidFrame);

    f = frame();
    f.validityPlane.requiredFlag = false;
    EXPECT_FALSE(f.isStructurallyValid());
    EXPECT_EQ(validate(contract(), f), particle::FieldProfileValidationCode::InvalidFrame);

    f = frame(); f.validityPlane.valueClass = particle::FieldValueClass::Scalar;
    EXPECT_FALSE(f.isStructurallyValid());
    EXPECT_EQ(validate(contract(), f), particle::FieldProfileValidationCode::InvalidFrame);
    f = frame(); f.validityPlane.dtype = particle::FieldDType::UInt8;
    EXPECT_FALSE(f.isStructurallyValid());
    EXPECT_EQ(validate(contract(), f), particle::FieldProfileValidationCode::InvalidFrame);
    f = frame(); f.validityPlane.unitID = particle::FieldUnitMetersPerSecond;
    EXPECT_FALSE(f.isStructurallyValid());
    EXPECT_EQ(validate(contract(), f), particle::FieldProfileValidationCode::InvalidFrame);
    f = frame(); f.validityPlane.key.groupID = 1;
    EXPECT_FALSE(f.isStructurallyValid());
    EXPECT_EQ(validate(contract(), f), particle::FieldProfileValidationCode::InvalidFrame);
    f = frame(); f.validityPlane.key.component = particle::FieldComponent::UEast;
    EXPECT_FALSE(f.isStructurallyValid());
    EXPECT_EQ(validate(contract(), f), particle::FieldProfileValidationCode::InvalidFrame);

    f = frame();
    f.validityPlane.storage = std::make_shared<const std::vector<uint8_t>>(*f.validityBits);
    EXPECT_FALSE(f.isStructurallyValid());
    EXPECT_EQ(validate(contract(), f), particle::FieldProfileValidationCode::InvalidFrame);
}

TEST(ParticleFieldProfileValidator, EnforcesVelocityDirectionAndValidityCoreRules) {
    EXPECT_EQ(particle::FieldUnitNone, 0);
    EXPECT_EQ(particle::FieldUnitMetersPerSecond, 1);
    EXPECT_EQ(particle::FieldUnitDegrees, 4);
    auto c = contract(); auto f = frame(); const auto bytes = fourLittleEndian16BitSamples();
    f.planes.push_back(plane(100, particle::FieldValueClass::VelocityComponent, particle::FieldUnitMetersPerSecond, 3,
                             particle::FieldComponent::UEast, particle::FieldDType::Int16, false, bytes));
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidVelocityGroup);
    f.planes.push_back(plane(101, particle::FieldValueClass::VelocityComponent, particle::FieldUnitMetersPerSecond, 3,
                             particle::FieldComponent::VNorth, particle::FieldDType::Int16, false, bytes));
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::Accepted);
    f.planes.back().unitID = 99;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidCoreUnit);
    f = frame();
    f.planes.push_back(plane(100, particle::FieldValueClass::VelocityComponent, particle::FieldUnitMetersPerSecond, 0,
                             particle::FieldComponent::UEast, particle::FieldDType::Int16, false, bytes));
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidVelocityGroup);
    f = frame();
    f.planes.push_back(plane(100, particle::FieldValueClass::VelocityComponent, particle::FieldUnitMetersPerSecond, 3,
                             particle::FieldComponent::UEast, particle::FieldDType::Int16, false, bytes));
    f.planes.push_back(plane(101, particle::FieldValueClass::VelocityComponent, particle::FieldUnitMetersPerSecond, 3,
                             particle::FieldComponent::UEast, particle::FieldDType::Int16, false, bytes));
    f.planes.push_back(plane(102, particle::FieldValueClass::VelocityComponent, particle::FieldUnitMetersPerSecond, 3,
                             particle::FieldComponent::VNorth, particle::FieldDType::Int16, false, bytes));
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidVelocityGroup);
    f = frame(); f.planes.front().key.semanticID = 10; f.planes.front().valueClass = particle::FieldValueClass::DirectionAngle;
    f.planes.front().unitID = particle::FieldUnitDegrees;
    f.planes.front().dtype = particle::FieldDType::UInt16;
    f.planes.front().scale = 360.0f / 65536.0f;
    c.requiredPlanes.front().semanticID = 10; c.requiredPlanes.front().valueClass = particle::FieldValueClass::DirectionAngle;
    c.requiredPlanes.front().unitID = particle::FieldUnitDegrees;
    f.planes.front().storage = bytes;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::Accepted);
    f.planes.front().unitID = 2;
    c.requiredPlanes.front().unitID = 2;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::SemanticContractMismatch);
    c = contract(); f = frame(); f.validityPlane.dtype = particle::FieldDType::UInt8;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidFrame);
    f = frame(); f.validityPlane.unitID = 1;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidFrame);
    f = frame(); f.validityPlane.valueClass = particle::FieldValueClass::Scalar;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidFrame);
}

TEST(ParticleFieldProfileValidator, EnforcesReservedCoreSemanticContractsFromProtocolTable) {
    auto c = contract(); auto f = frame();
    // Contract and frame agree on the spoof, but reserved semantic 1 remains
    // VELOCITY_COMPONENT/MPS/U_EAST and cannot become an opaque scalar.
    c.requiredPlanes.front().semanticID = 1;
    f.planes.front().key.semanticID = 1;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::SemanticContractMismatch);

    c = contract(); f = frame();
    c.requiredPlanes.front() = {2, particle::FieldValueClass::VelocityComponent, particle::FieldUnitMetersPerSecond, 4,
                                particle::FieldComponent::UEast, true};
    f.planes.front() = plane(2, particle::FieldValueClass::VelocityComponent, particle::FieldUnitMetersPerSecond, 4,
                              particle::FieldComponent::UEast, particle::FieldDType::Int16, true,
                              fourLittleEndian16BitSamples(), "u-spoof");
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::SemanticContractMismatch);

    c = contract(); f = frame();
    c.requiredPlanes.back().semanticID = 3;
    c.requiredPlanes.back().valueClass = particle::FieldValueClass::Scalar;
    f.validityPlane.key.semanticID = 3;
    f.validityPlane.valueClass = particle::FieldValueClass::Scalar;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidFrame);
}

TEST(ParticleFieldProfileValidator, EnforcesDirectionCircularQuantizationContract) {
    auto c = contract(); auto f = frame();
    c.requiredPlanes.front() = {10, particle::FieldValueClass::DirectionAngle, particle::FieldUnitDegrees, 0,
                                particle::FieldComponent::None, true};
    f.planes.front() = plane(10, particle::FieldValueClass::DirectionAngle, particle::FieldUnitDegrees, 0,
                              particle::FieldComponent::None, particle::FieldDType::UInt16, true,
                              fourLittleEndian16BitSamples(), "direction");
    f.planes.front().scale = 360.0f / 65536.0f;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::Accepted);

    f.planes.front().unitID = particle::FieldUnitMetersPerSecond;
    c.requiredPlanes.front().unitID = particle::FieldUnitMetersPerSecond;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::SemanticContractMismatch);
    f.planes.front().unitID = particle::FieldUnitDegrees;
    c.requiredPlanes.front().unitID = particle::FieldUnitDegrees;
    f.planes.front().key.component = particle::FieldComponent::UEast;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::SemanticContractMismatch);
    f.planes.front().key.component = particle::FieldComponent::None;
    f.planes.front().key.groupID = 1;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::SemanticContractMismatch);
    f.planes.front().key.groupID = 0;
    f.planes.front().dtype = particle::FieldDType::UInt8;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidCoreUnit);
    f.planes.front().dtype = particle::FieldDType::UInt16;
    f.planes.front().scale = 0.1f;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidCoreUnit);
    f.planes.front().scale = 360.0f / 65536.0f;
    f.planes.front().offset = 1.0f;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::InvalidCoreUnit);
}

TEST(ParticleFieldProfileValidator, UsesSemanticAndGroupAsTheOnlyPlaneIdentity) {
    auto f = frame();
    auto duplicate = f.planes.front();
    duplicate.key.component = particle::FieldComponent::UEast;
    f.planes.push_back(duplicate);
    EXPECT_FALSE(f.isStructurallyValid());

    auto c = contract();
    auto duplicateRequirement = c.requiredPlanes.front();
    duplicateRequirement.component = particle::FieldComponent::UEast;
    c.requiredPlanes.push_back(duplicateRequirement);
    EXPECT_FALSE(c.isStructurallyValid());

    c = contract();
    f = frame();
    f.planes.front().key.component = particle::FieldComponent::UEast;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::PlaneDescriptorMismatch);
}

TEST(ParticleFieldProfileValidator, AppliesSchemaSharedValidityAndTimeRules) {
    auto c = contract(); auto f = frame(); c.contractSchemaVersion = 2;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::UnsupportedContractSchema);
    c = contract(); c.sharedValidityRequired = false;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::UnsupportedSharedValidityMode);
    c = contract(); f.frameContentIdentity.validTimeMs = 999;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::Accepted);
    c.timeRelation = particle::FieldTimeRelation::ValidGteReference;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::TimeRelationMismatch);
    f.frameContentIdentity.validTimeMs = 1'000;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::Accepted);
    c.timeRelation = particle::FieldTimeRelation::ValidEqReference; f.frameContentIdentity.validTimeMs = 1'001;
    EXPECT_EQ(validate(c, f), particle::FieldProfileValidationCode::TimeRelationMismatch);
}

TEST(ParticleFieldRuntime, RevisionIdentityAndIndependentCadence) {
    particle::FieldSeries series; const uint64_t base = 1'700'000'000'000ULL, hour = 3'600'000ULL;
    EXPECT_EQ(series.submit(runtimeFrame(base, 1, 1, "a")), particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(series.submit(runtimeFrame(base, 1, 1, "a")), particle::FieldSeriesSubmitResult::Duplicate);
    EXPECT_EQ(series.submit(runtimeFrame(base, 1, 1, "changed")), particle::FieldSeriesSubmitResult::RejectedIdentityConflict);
    EXPECT_EQ(series.submit(runtimeFrame(base, 2, 2, "b")), particle::FieldSeriesSubmitResult::ReplacedRevision);
    particle::EncodedFieldStore store;
    EXPECT_EQ(store.submit("wind-key", runtimeFrame(base, 1, 1, "w0", "wind")), particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(store.submit("wind-key", runtimeFrame(base + hour, 2, 1, "w1", "wind")), particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(store.submit("wave-key", runtimeFrame(base, 1, 1, "a0", "wave")), particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(store.submit("wave-key", runtimeFrame(base + 3 * hour, 2, 1, "a3", "wave")), particle::FieldSeriesSubmitResult::Inserted);
    const auto wind = store.resolve("wind-key", base + hour / 2), wave = store.resolve("wave-key", base + hour / 2);
    ASSERT_TRUE(wind); ASSERT_TRUE(wave); EXPECT_FLOAT_EQ(wind->mix, 0.5f); EXPECT_FLOAT_EQ(wave->mix, 1.0f / 6.0f);
}

TEST(ParticleFieldRuntime, ResolvesExactSingleFrameAndRejectsUncoveredTimes) {
    constexpr uint64_t base = 1'700'000'000'000ULL;
    particle::FieldSeries series;
    ASSERT_EQ(series.submit(runtimeFrame(base, 1, 1, "only")), particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(series.decodedByteSize(), 5u);
    const auto exact = series.resolve(base);
    ASSERT_TRUE(exact);
    EXPECT_EQ(exact->t0, exact->t1);
    EXPECT_FLOAT_EQ(exact->mix, 0.0f);
    EXPECT_EQ(exact->requestedForecastTimeMs, base);
    EXPECT_FALSE(series.resolve(base + 1));

    ASSERT_EQ(series.submit(runtimeFrame(base + 10'000, 1, 2, "next")),
              particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(series.decodedByteSize(), 10u);
    const auto middle = series.resolve(base + 5'000);
    ASSERT_TRUE(middle);
    EXPECT_NE(middle->t0, middle->t1);
    EXPECT_FLOAT_EQ(middle->mix, 0.5f);
    EXPECT_FALSE(series.resolve(base - 1));
    EXPECT_FALSE(series.resolve(base + 10'001));
}

TEST(ParticleFieldRuntime, FullWindowRollsBothDirectionsAndBreaksTiesAtEarliest) {
    constexpr uint64_t base = 1'700'000'000'000ULL;
    constexpr uint64_t hour = 3'600'000ULL;
    const auto fill = [&](particle::FieldSeries& series) {
        for (uint64_t i = 0; i < 8; ++i)
            ASSERT_EQ(series.submit(runtimeFrame(base + i * hour, 1, i + 1, "frame-" + std::to_string(i))),
                      particle::FieldSeriesSubmitResult::Inserted);
    };
    const auto times = [](const particle::FieldSeries& series) {
        std::vector<uint64_t> values;
        for (const auto& entry : series.entries()) values.push_back(entry->frameContentIdentity.validTimeMs);
        return values;
    };

    particle::FieldSeries future;
    fill(future);
    ASSERT_EQ(future.submit(runtimeFrame(base + 8 * hour, 1, 9, "future")),
              particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(future.size(), 8u);
    EXPECT_EQ(times(future).front(), base + hour);
    EXPECT_EQ(times(future).back(), base + 8 * hour);

    particle::FieldSeries past;
    fill(past);
    ASSERT_EQ(past.submit(runtimeFrame(base - hour, 1, 9, "past")), particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(past.size(), 8u);
    EXPECT_EQ(times(past).front(), base - hour);
    EXPECT_EQ(times(past).back(), base + 6 * hour);

    particle::FieldSeries internal;
    fill(internal);
    ASSERT_EQ(internal.submit(runtimeFrame(base + 3 * hour + hour / 2, 1, 9, "internal")),
              particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(internal.size(), 8u);
    EXPECT_EQ(times(internal), (std::vector<uint64_t>{base + hour, base + 2 * hour, base + 3 * hour,
                                                       base + 3 * hour + hour / 2, base + 4 * hour,
                                                       base + 5 * hour, base + 6 * hour, base + 7 * hour}));
}

TEST(ParticleFieldRuntime, UnorderedSubmissionsAreSortedAndReferenceCyclesStaySeparate) {
    constexpr uint64_t base = 1'700'000'000'000ULL;
    constexpr uint64_t hour = 3'600'000ULL;
    particle::EncodedFieldStore store;
    for (const uint64_t index : {4ULL, 1ULL, 3ULL, 0ULL, 2ULL})
        ASSERT_EQ(store.submit("wind", runtimeFrame(base + index * hour, 1, index + 1,
                                                     "wind-" + std::to_string(index))),
                  particle::FieldSeriesSubmitResult::Inserted);
    const auto* series = store.find("wind");
    ASSERT_TRUE(series);
    ASSERT_EQ(series->size(), 5u);
    for (size_t i = 0; i < series->size(); ++i)
        EXPECT_EQ(series->entries()[i]->frameContentIdentity.validTimeMs, base + i * hour);

    auto differentReference = std::make_shared<particle::DecodedFieldFrame>(*runtimeFrame(
        base + 5 * hour, 1, 6, "new-reference"));
    differentReference->referenceTimeMs = 2'000;
    differentReference->resolvedProfileID = "wind-profile-v2";
    differentReference->resolvedProfileVersion = 2;
    EXPECT_EQ(store.submit("wind", differentReference), particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(store.find("wind")->size(), 6u);
    EXPECT_FALSE(store.resolve("wind", base + 4 * hour + hour / 2));
    const auto exactNewBatch = store.resolve("wind", base + 5 * hour);
    ASSERT_TRUE(exactNewBatch);
    EXPECT_EQ(exactNewBatch->t0, exactNewBatch->t1);

    auto incompatibleGrid = std::make_shared<particle::DecodedFieldFrame>(*runtimeFrame(
        base + 6 * hour, 1, 7, "wrong-grid"));
    incompatibleGrid->grid.east = 132.0; // valid grid, incompatible with 131.0 bounds
    EXPECT_EQ(store.submit("wind", incompatibleGrid), particle::FieldSeriesSubmitResult::RejectedIncompatibleFrame);
    EXPECT_EQ(store.find("wind")->size(), 6u);
    EXPECT_EQ(store.submit("wave", runtimeFrame(base, 1, 1, "wave", "wave")),
              particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(store.find("wind")->size(), 6u);
    EXPECT_EQ(store.find("wave")->size(), 1u);
}

TEST(ParticleFieldRuntime, RejectedFullWindowInsertDoesNotEvictAndKeysRemainIndependent) {
    constexpr uint64_t base = 1'700'000'000'000ULL;
    constexpr uint64_t hour = 3'600'000ULL;
    particle::EncodedFieldStore store;
    for (uint64_t i = 0; i < 8; ++i)
        ASSERT_EQ(store.submit("wind", runtimeFrame(base + i * hour, 1, i + 1, "wind-" + std::to_string(i))),
                  particle::FieldSeriesSubmitResult::Inserted);
    const auto bytesBeforeRejectedFrame = store.decodedByteSize();
    auto incompatible = std::make_shared<particle::DecodedFieldFrame>(*runtimeFrame(
        base + 8 * hour, 1, 9, "wrong-grid"));
    incompatible->grid.east = 132.0; // valid grid, incompatible with 131.0 bounds
    ASSERT_EQ(store.submit("wind", incompatible), particle::FieldSeriesSubmitResult::RejectedIncompatibleFrame);
    ASSERT_EQ(store.find("wind")->size(), 8u);
    EXPECT_EQ(store.decodedByteSize(), bytesBeforeRejectedFrame);
    EXPECT_EQ(store.find("wind")->entries().front()->frameContentIdentity.validTimeMs, base);
    for (uint64_t i = 0; i < 8; ++i)
        ASSERT_EQ(store.submit("wave", runtimeFrame(base + (i + 20) * hour, 1, i + 1,
                                                     "wave-" + std::to_string(i), "wave-dataset")),
                  particle::FieldSeriesSubmitResult::Inserted);
    EXPECT_EQ(store.find("wind")->size(), 8u);
    EXPECT_EQ(store.find("wave")->size(), 8u);
    EXPECT_EQ(store.find("wave")->entries().front()->frameContentIdentity.validTimeMs, base + 20 * hour);
    EXPECT_FALSE(store.resolve("wave", base + 1));
}

TEST(ParticleFieldRuntime, RepeatedBidirectionalRollingAndClearStayBounded) {
    constexpr uint64_t base = 1'700'000'000'000ULL;
    constexpr uint64_t hour = 3'600'000ULL;
    particle::EncodedFieldStore store;
    auto fill = [&] {
        for (uint64_t i = 0; i < 8; ++i)
            ASSERT_EQ(store.submit("wind", runtimeFrame(base + i * hour, 1, i + 1,
                                                         "seed-" + std::to_string(i))),
                      particle::FieldSeriesSubmitResult::Inserted);
    };
    fill();
    for (uint64_t iteration = 0; iteration < 120; ++iteration) {
        const uint64_t time = iteration % 2 == 0 ? base + (8 + iteration) * hour
                                                 : base - (iteration + 1) * hour;
        ASSERT_EQ(store.submit("wind", runtimeFrame(time, 1, iteration + 20,
                                                     "rolling-" + std::to_string(iteration))),
                  particle::FieldSeriesSubmitResult::Inserted);
        ASSERT_EQ(store.find("wind")->size(), 8u);
        EXPECT_EQ(store.decodedByteSize(), 40u);
        if (iteration % 10 == 9) {
            EXPECT_TRUE(store.clear("wind"));
            EXPECT_EQ(store.decodedByteSize(), 0u);
            fill();
        }
    }
    store.clearAll();
    EXPECT_EQ(store.fieldCount(), 0u);
    EXPECT_EQ(store.decodedByteSize(), 0u);
}

TEST(ParticleFieldRuntime, ForecastClockOwnsRequestedTimeAndPlayback) {
    particle::ForecastClock clock;
    EXPECT_TRUE(clock.setRequestedForecastTime(1'700'000'000'000ULL)); EXPECT_FALSE(clock.advance(1'000));
    EXPECT_TRUE(clock.setPlaybackSpeed(2.0f)); clock.play(); EXPECT_TRUE(clock.advance(500));
    EXPECT_EQ(clock.requestedForecastTime(), 1'700'000'001'000ULL); clock.pause();
    EXPECT_FALSE(clock.advance(500)); EXPECT_FALSE(clock.setPlaybackSpeed(3.0f));
}

TEST(ParticleFieldRuntime, ForecastClockPreservesHalfMillisecondRemainder) {
    particle::ForecastClock clock;
    ASSERT_TRUE(clock.setRequestedForecastTime(1'000));
    ASSERT_TRUE(clock.setPlaybackSpeed(0.5f));
    clock.play();
    EXPECT_FALSE(clock.advance(1));
    EXPECT_EQ(clock.requestedForecastTime(), 1'000);
    ASSERT_TRUE(clock.setRequestedForecastTime(2'000));
    EXPECT_FALSE(clock.advance(1));
    EXPECT_EQ(clock.requestedForecastTime(), 2'000);
    EXPECT_TRUE(clock.advance(1));
    EXPECT_EQ(clock.requestedForecastTime(), 2'001);

    ASSERT_TRUE(clock.setRequestedForecastTime(1'000));
    EXPECT_FALSE(clock.advance(1));
    EXPECT_EQ(clock.requestedForecastTime(), 1'000);
    for (size_t i = 0; i < 8; ++i) EXPECT_EQ(clock.advance(1), i % 2 == 0);
    EXPECT_EQ(clock.requestedForecastTime(), 1'004);
    clock.pause();
    EXPECT_FALSE(clock.advance(10));
    EXPECT_EQ(clock.requestedForecastTime(), 1'004);

    ASSERT_TRUE(clock.setRequestedForecastTime(3'000));
    ASSERT_TRUE(clock.setPlaybackSpeed(1.0f));
    clock.play();
    EXPECT_TRUE(clock.advance(3));
    EXPECT_EQ(clock.requestedForecastTime(), 3'003);
    ASSERT_TRUE(clock.setPlaybackSpeed(2.0f));
    EXPECT_TRUE(clock.advance(3));
    EXPECT_EQ(clock.requestedForecastTime(), 3'009);

    ASSERT_TRUE(clock.setRequestedForecastTime(std::numeric_limits<uint64_t>::max()));
    EXPECT_FALSE(clock.advance(1));
    EXPECT_EQ(clock.requestedForecastTime(), std::numeric_limits<uint64_t>::max());
}