#include <mbgl/style/sources/raster_array_source.hpp>

#include <mbgl/particle/field/offline_fixture_particle_adapter.hpp>
#include <mbgl/particle/field/local_particle_field_feed.hpp>
#include <mbgl/storage/file_source.hpp>
#include <mbgl/storage/resource.hpp>
#include <mbgl/style/source_observer.hpp>
#include <mbgl/style/sources/raster_array_source_impl.hpp>
#include <mbgl/util/async_request.hpp>
#include <mbgl/util/logging.hpp>

#include <atomic>
#include <algorithm>
#include <cmath>

namespace mbgl::style {

namespace {

// Snapshot revision is process-wide rather than source-instance local: a
// Style recreation can replace an east source with a north source under the
// same source ID, and that must always invalidate the prior GPU field.
std::atomic_uint64_t nextRasterArrayFieldGeneration{0};

} // namespace

RasterArrayTimelineRuntime::RasterArrayTimelineRuntime(std::vector<Entry> entries_)
    : entries(std::move(entries_)) {
    if (entries.size() < 2) {
        throw std::invalid_argument("particle timeline requires at least two snapshots");
    }
    for (size_t index = 1; index < entries.size(); ++index) {
        if (entries[index - 1].timestamp >= entries[index].timestamp) {
            throw std::invalid_argument("particle timeline timestamps must be strictly increasing");
        }
    }
    if (entries.size() > cacheCapacity_) {
        throw std::invalid_argument("particle timeline exceeds bounded validation cache");
    }
}

bool RasterArrayTimelineRuntime::append(Entry entry) {
    std::lock_guard<std::mutex> lock(mutex);
    if (!entry.field || !entry.field->isValid() || entries.size() >= cacheCapacity_ ||
        entries.back().timestamp >= entry.timestamp) {
        return false;
    }
    const bool continuesEndedTail = state == RasterArrayPlaybackState::Ended && t0 + 1 == entries.size() - 1;
    entries.push_back(std::move(entry));
    if (continuesEndedTail) {
        state = RasterArrayPlaybackState::Paused;
        lastTick = {};
    }
    return true;
}

RasterArrayTimelineFrame RasterArrayTimelineRuntime::frameLocked() const {
    return {entries[t0].field, entries[t0 + 1].field, mix, t0, t0 + 1, windowRevision, cadenceRevision, state};
}

RasterArrayTimelineFrame RasterArrayTimelineRuntime::frame() const {
    std::lock_guard<std::mutex> lock(mutex);
    return frameLocked();
}

float RasterArrayTimelineRuntime::playbackSpeed() const {
    std::lock_guard<std::mutex> lock(mutex);
    return speed;
}

size_t RasterArrayTimelineRuntime::size() const {
    std::lock_guard<std::mutex> lock(mutex);
    return entries.size();
}

RasterArrayTimelineRuntime::Entry RasterArrayTimelineRuntime::lastEntry() const {
    std::lock_guard<std::mutex> lock(mutex);
    return entries.back();
}

bool RasterArrayTimelineRuntime::moveToLocked(const size_t nextT0) {
    if (nextT0 + 1 >= entries.size()) return false;
    t0 = nextT0;
    mix = 0.0f;
    ++windowRevision;
    return true;
}

RasterArrayCadenceAdvance RasterArrayTimelineRuntime::advanceIfDue(const std::chrono::steady_clock::time_point now) {
    std::lock_guard<std::mutex> lock(mutex);
    if (targetFps_ != RasterArrayTargetFps::Auto && lastCadenceTick.time_since_epoch().count() != 0) {
        const auto interval = std::chrono::duration<double>(1.0 / static_cast<uint8_t>(targetFps_));
        if (now - lastCadenceTick < interval) return {frameLocked(), false};
    }
    lastCadenceTick = now;
    ++cadenceRevision;
    ++cadenceTickCount;
    if (lastCadenceLog.time_since_epoch().count() == 0 || now - lastCadenceLog >= std::chrono::seconds(1)) {
        lastCadenceLog = now;
        Log::Info(Event::Render, "WeatherCadence: targetFps=" + std::to_string(static_cast<uint8_t>(targetFps_)) +
                                     " weatherTicks=" + std::to_string(cadenceTickCount));
    }
    if (lastTick == std::chrono::steady_clock::time_point{}) {
        lastTick = now;
        return {frameLocked(), true};
    }
    const auto elapsed = std::chrono::duration<float, std::milli>(now - lastTick).count();
    lastTick = now;
    if (state != RasterArrayPlaybackState::Playing) return {frameLocked(), true};
    const auto duration = static_cast<float>(entries[t0 + 1].timestamp - entries[t0].timestamp);
    mix += elapsed * speed / duration;
    while (mix >= 1.0f) {
        if (gap || invalid) {
            mix = 1.0f;
            stateBeforeWait = state;
            state = RasterArrayPlaybackState::Waiting;
            Log::Warning(Event::Render,
                         std::string("ParticleTimeline: event=TIMELINE_REJECTED reason=") +
                             (invalid ? "INVALID_NEXT" : "GAP_HOLD") + " t1=" + entries[t0 + 1].field->fieldID);
            return {frameLocked(), true};
        }
        if (!moveToLocked(t0 + 1)) {
            mix = 1.0f;
            state = RasterArrayPlaybackState::Ended;
            Log::Info(Event::Render, "ParticleTimeline: event=END t1=" + entries[t0 + 1].field->fieldID);
            return {frameLocked(), true};
        }
        Log::Info(Event::Render,
                  "ParticleTimeline: event=WINDOW_BOUNDARY oldT1=" + entries[t0].field->fieldID +
                      " newT1=" + entries[t0 + 1].field->fieldID + " result=APPLIED reuse=OLD_T1_TO_NEW_T0");
        Log::Info(Event::Render,
                  "ParticleTimeline: event=TIMELINE_CACHE_HIT index=" + std::to_string(t0 + 1) +
                      " cacheSize=" + std::to_string(entries.size()) + " capacity=" + std::to_string(cacheCapacity_));
    }
    return {frameLocked(), true};
}

RasterArrayTimelineFrame RasterArrayTimelineRuntime::tick() {
    return advanceIfDue(std::chrono::steady_clock::now()).frame;
}

bool RasterArrayTimelineRuntime::setTargetFps(const RasterArrayTargetFps target) {
    std::lock_guard<std::mutex> lock(mutex);
    targetFps_ = target;
    lastCadenceTick = {};
    return true;
}

RasterArrayTargetFps RasterArrayTimelineRuntime::targetFps() const {
    std::lock_guard<std::mutex> lock(mutex);
    return targetFps_;
}

bool RasterArrayTimelineRuntime::cadenceDue(const std::chrono::steady_clock::time_point now) const {
    std::lock_guard<std::mutex> lock(mutex);
    if (targetFps_ == RasterArrayTargetFps::Auto || lastCadenceTick.time_since_epoch().count() == 0) return true;
    const auto interval = std::chrono::duration<double>(1.0 / static_cast<uint8_t>(targetFps_));
    return now - lastCadenceTick >= interval;
}

void RasterArrayTimelineRuntime::markCadenceTick(const std::chrono::steady_clock::time_point now) {
    std::lock_guard<std::mutex> lock(mutex);
    lastCadenceTick = now;
}

bool RasterArrayTimelineRuntime::play() {
    std::lock_guard<std::mutex> lock(mutex);
    if (state == RasterArrayPlaybackState::Waiting) {
        Log::Warning(Event::Render, "ParticleTimeline: event=PLAYBACK_REJECTED reason=WAITING");
        return false;
    }
    if (state == RasterArrayPlaybackState::Ended) {
        t0 = 0;
        mix = 0.0f;
        ++windowRevision;
        Log::Info(Event::Render, "ParticleTimeline: event=PLAYBACK_PLAY reason=RESTART_FROM_END");
    }
    state = RasterArrayPlaybackState::Playing;
    lastTick = std::chrono::steady_clock::now();
    Log::Info(Event::Render,
              "ParticleTimeline: event=PLAYBACK_PLAY speed=" + std::to_string(speed) + " cursor=" + std::to_string(t0));
    return true;
}
bool RasterArrayTimelineRuntime::pause() {
    std::lock_guard<std::mutex> lock(mutex);
    if (state == RasterArrayPlaybackState::Waiting || state == RasterArrayPlaybackState::Ended) {
        Log::Warning(Event::Render, "ParticleTimeline: event=PLAYBACK_REJECTED reason=ENDED_OR_WAITING");
        return false;
    }
    state = RasterArrayPlaybackState::Paused;
    Log::Info(Event::Render, "ParticleTimeline: event=PLAYBACK_PAUSE cursor=" + std::to_string(t0));
    return true;
}
bool RasterArrayTimelineRuntime::setSpeed(const float value) {
    if (value != 0.5f && value != 1.0f && value != 2.0f) return false;
    std::lock_guard<std::mutex> lock(mutex);
    const auto oldSpeed = speed;
    speed = value;
    Log::Info(Event::Render,
              "ParticleTimeline: event=PLAYBACK_SPEED oldSpeed=" + std::to_string(oldSpeed) +
                  " newSpeed=" + std::to_string(speed) + " cursor=" + std::to_string(t0));
    return true;
}
bool RasterArrayTimelineRuntime::seek(const size_t index) {
    std::lock_guard<std::mutex> lock(mutex);
    if (index >= entries.size()) return false;
    const auto fromState = state;
    const auto fromCursor = t0;
    if (index + 1 == entries.size()) {
        t0 = index - 1;
        mix = 1.0f;
        state = RasterArrayPlaybackState::Ended;
    } else {
        moveToLocked(index);
        state = fromState == RasterArrayPlaybackState::Playing ? RasterArrayPlaybackState::Playing
                                                               : RasterArrayPlaybackState::Paused;
    }
    Log::Info(Event::Render,
              "ParticleTimeline: event=PLAYBACK_SEEK fromState=" + std::to_string(static_cast<int>(fromState)) +
                  " toState=" + std::to_string(static_cast<int>(state)) + " fromCursor=" + std::to_string(fromCursor) +
                  " toCursor=" + std::to_string(t0) + " targetIndex=" + std::to_string(index) + " t0Index=" +
                  std::to_string(t0) + " t1Index=" + std::to_string(t0 + 1) + " mix=" + std::to_string(mix));
    return true;
}

bool RasterArrayTimelineRuntime::seekTimestamp(const uint64_t timestamp) {
    std::lock_guard<std::mutex> lock(mutex);
    if (gap || invalid || entries.size() < 2 || timestamp < entries.front().timestamp || timestamp > entries.back().timestamp) return false;
    const auto previousT0 = t0;
    const auto previousState = state;
    const auto upper = std::upper_bound(entries.begin(), entries.end(), timestamp, [](uint64_t value, const Entry& entry) {
        return value < entry.timestamp;
    });
    if (upper == entries.begin()) return false;
    if (upper == entries.end()) {
        t0 = entries.size() - 2;
        mix = 1.0f;
        state = RasterArrayPlaybackState::Ended;
    } else {
        t0 = static_cast<size_t>(std::distance(entries.begin(), upper) - 1);
        const auto duration = static_cast<float>(entries[t0 + 1].timestamp - entries[t0].timestamp);
        mix = duration > 0.0f ? static_cast<float>(timestamp - entries[t0].timestamp) / duration : 0.0f;
        state = previousState == RasterArrayPlaybackState::Playing ? RasterArrayPlaybackState::Playing
                                                                    : RasterArrayPlaybackState::Paused;
    }
    if (t0 != previousT0) ++windowRevision;
    lastTick = std::chrono::steady_clock::now();
    Log::Info(Event::Render, "ParticleTimeline: event=PLAYBACK_SEEK_TIMESTAMP timestamp=" + std::to_string(timestamp) +
                                " t0Index=" + std::to_string(t0) + " t1Index=" + std::to_string(t0 + 1) +
                                " mix=" + std::to_string(mix));
    return true;
}
bool RasterArrayTimelineRuntime::step(const int direction) {
    std::lock_guard<std::mutex> lock(mutex);
    const auto target = direction < 0 ? (t0 == 0 ? 0 : t0 - 1) : std::min(t0 + 1, entries.size() - 2);
    moveToLocked(target);
    state = RasterArrayPlaybackState::Paused;
    Log::Info(Event::Render, "ParticleTimeline: event=STEP direction=" + std::to_string(direction));
    return true;
}
bool RasterArrayTimelineRuntime::injectGap() {
    std::lock_guard<std::mutex> lock(mutex);
    gap = true;
    invalid = false;
    Log::Info(Event::Render, "ParticleTimeline: event=GAP_INJECTED");
    return true;
}
bool RasterArrayTimelineRuntime::injectInvalid() {
    std::lock_guard<std::mutex> lock(mutex);
    invalid = true;
    gap = false;
    Log::Info(Event::Render, "ParticleTimeline: event=INVALID_INJECTED");
    return true;
}
bool RasterArrayTimelineRuntime::recover() {
    std::lock_guard<std::mutex> lock(mutex);
    gap = invalid = false;
    if (state == RasterArrayPlaybackState::Waiting) state = stateBeforeWait;
    Log::Info(Event::Render, "ParticleTimeline: event=RECOVERED state=" + std::to_string(static_cast<int>(state)));
    return true;
}

RasterArraySource::RasterArraySource(std::string id,
                                     std::string fixtureID,
                                     std::string metadataURL,
                                     std::string fieldURL,
                                     std::string nextFixtureID,
                                     std::string nextMetadataURL,
                                     std::string nextFieldURL,
                                     std::vector<RasterArrayRuntimeFieldDescriptor> runtimeFieldDescriptors)
    : Source(makeMutable<Impl>(std::move(id),
                               std::move(fixtureID),
                               std::move(metadataURL),
                               std::move(fieldURL),
                               std::move(nextFixtureID),
                               std::move(nextMetadataURL),
                               std::move(nextFieldURL),
                               std::move(runtimeFieldDescriptors))) {}

RasterArraySource::RasterArraySource(std::string id, std::vector<RasterArrayLocalFieldDescriptor> localFieldDescriptors)
    : Source(makeMutable<Impl>(std::move(id), std::move(localFieldDescriptors))) {}

RasterArraySource::RasterArraySource(std::string id)
    : Source(makeMutable<Impl>(std::move(id))) {}

RasterArraySource::~RasterArraySource() {
    weakFactory.invalidateWeakPtrs();
}

const RasterArraySource::Impl& RasterArraySource::impl() const {
    return static_cast<const Impl&>(*baseImpl);
}

void RasterArraySource::loadDescription(FileSource& fileSource) {
    if (impl().hostFieldSubmission) {
        if (loaded) return;
        loaded = true;
        Log::Info(Event::Render, "ParticleField: event=HOST_FIELD_INGRESS_READY sourceId=" + getID());
        observer->onSourceLoaded(*this);
        return;
    }
    if (metadataRequest || fieldRequest) return;
    if (!impl().localFields.empty()) {
        localFields.reserve(impl().localFields.size());
        for (const auto& descriptor : impl().localFields) {
            LocalFieldCandidate candidate;
            candidate.descriptor = descriptor;
            localFields.push_back(std::move(candidate));
            const auto index = localFields.size() - 1;
            auto& local = localFields.back();
            local.metadataRequest = fileSource.request(Resource::source(local.descriptor.metadataURL),
                [this, self = makeWeakPtr(), index](const Response& response) {
                    if (auto guard = self.lock(); self) {
                        if (response.error || response.noContent || !response.data) {
                            observer->onSourceError(*this, std::make_exception_ptr(std::runtime_error("particle local metadata request failed")));
                            return;
                        }
                        localFields[index].metadata = *response.data;
                        tryLoadLocalField(index);
                    }
                });
            local.fieldRequest = fileSource.request(Resource::source(local.descriptor.fieldURL),
                [this, self = makeWeakPtr(), index](const Response& response) {
                    if (auto guard = self.lock(); self) {
                        if (response.error || response.noContent || !response.data) {
                            observer->onSourceError(*this, std::make_exception_ptr(std::runtime_error("particle local field request failed")));
                            return;
                        }
                        localFields[index].fieldBytes = std::vector<char>(response.data->begin(), response.data->end());
                        tryLoadLocalField(index);
                    }
                });
        }
        return;
    }
    metadataRequest = fileSource.request(
        Resource::source(impl().metadataURL), [this, self = makeWeakPtr()](const Response& response) {
            if (auto guard = self.lock(); self) {
                if (response.error || response.noContent || !response.data) {
                    observer->onSourceError(
                        *this, std::make_exception_ptr(std::runtime_error("particle metadata request failed")));
                    return;
                }
                metadata = *response.data;
                tryLoad();
            }
        });
    fieldRequest = fileSource.request(
        Resource::source(impl().fieldURL), [this, self = makeWeakPtr()](const Response& response) {
            if (auto guard = self.lock(); self) {
                if (response.error || response.noContent || !response.data) {
                    observer->onSourceError(
                        *this, std::make_exception_ptr(std::runtime_error("particle field request failed")));
                    return;
                }
                fieldBytes = std::vector<char>(response.data->begin(), response.data->end());
                tryLoad();
            }
        });
    if (!impl().nextFixtureID.empty()) {
        if (impl().nextMetadataURL.empty() || impl().nextFieldURL.empty()) {
            observer->onSourceError(
                *this,
                std::make_exception_ptr(std::runtime_error("particle next fixture requires metadata and field URLs")));
            return;
        }
        nextMetadataRequest = fileSource.request(
            Resource::source(impl().nextMetadataURL), [this, self = makeWeakPtr()](const Response& response) {
                if (auto guard = self.lock(); self) {
                    if (response.error || response.noContent || !response.data) {
                        observer->onSourceError(
                            *this,
                            std::make_exception_ptr(std::runtime_error("particle next metadata request failed")));
                        return;
                    }
                    nextMetadata = *response.data;
                    tryLoad();
                }
            });
        nextFieldRequest = fileSource.request(
            Resource::source(impl().nextFieldURL), [this, self = makeWeakPtr()](const Response& response) {
                if (auto guard = self.lock(); self) {
                    if (response.error || response.noContent || !response.data) {
                        observer->onSourceError(
                            *this, std::make_exception_ptr(std::runtime_error("particle next field request failed")));
                        return;
                    }
                    nextFieldBytes = std::vector<char>(response.data->begin(), response.data->end());
                    tryLoad();
                }
            });
    }
    runtimeFields.reserve(impl().runtimeFields.size());
    for (const auto& descriptor : impl().runtimeFields) {
        RuntimeFieldCandidate runtimeCandidate;
        runtimeCandidate.descriptor = descriptor;
        runtimeFields.push_back(std::move(runtimeCandidate));
        const auto index = runtimeFields.size() - 1;
        auto& candidate = runtimeFields.back();
        candidate.metadataRequest = fileSource.request(
            Resource::source(candidate.descriptor.metadataURL),
            [this, self = makeWeakPtr(), index](const Response& response) {
                if (auto guard = self.lock(); self) {
                    if (response.error || response.noContent || !response.data) {
                        observer->onSourceError(
                            *this,
                            std::make_exception_ptr(std::runtime_error("particle runtime metadata request failed")));
                        return;
                    }
                    runtimeFields[index].metadata = *response.data;
                    tryLoadRuntimeField(index);
                }
            });
        candidate.fieldRequest = fileSource.request(
            Resource::source(candidate.descriptor.fieldURL),
            [this, self = makeWeakPtr(), index](const Response& response) {
                if (auto guard = self.lock(); self) {
                    if (response.error || response.noContent || !response.data) {
                        observer->onSourceError(
                            *this,
                            std::make_exception_ptr(std::runtime_error("particle runtime field request failed")));
                        return;
                    }
                    runtimeFields[index].fieldBytes = std::vector<char>(response.data->begin(), response.data->end());
                    tryLoadRuntimeField(index);
                }
            });
    }
}

void RasterArraySource::tryLoad() {
    if (!metadata || !fieldBytes || (!impl().nextFixtureID.empty() && (!nextMetadata || !nextFieldBytes))) return;
    try {
        const auto fieldGeneration = ++nextRasterArrayFieldGeneration;
        auto field = particle::OfflineFixtureParticleAdapter::load(
            impl().fixtureID, *metadata, *fieldBytes, fieldGeneration);
        if (impl().nextFixtureID.empty()) {
            baseImpl = makeMutable<Impl>(impl(), std::move(field));
        } else {
            auto nextField = particle::OfflineFixtureParticleAdapter::load(
                impl().nextFixtureID, *nextMetadata, *nextFieldBytes, ++nextRasterArrayFieldGeneration);
            if (!field.isTimeInterpolationCompatibleWith(nextField)) {
                throw std::invalid_argument("particle time-window fixtures are incompatible");
            }
            baseImpl = makeMutable<Impl>(impl(), std::move(field), std::move(nextField));
        }
        loaded = true;
        const auto& loadedField = *impl().field;
        Log::Info(Event::Render,
                  "ParticleField: source loaded fixtureId=" + impl().fixtureID + " metadata=" + impl().metadataURL +
                      " field=" + impl().fieldURL + " generation=" + std::to_string(loadedField.generation) +
                      " dataVersion=" + std::to_string(loadedField.dataVersion));
        if (impl().nextField) {
            Log::Info(Event::Render,
                  "ParticleField: source loaded next fieldId=" + impl().nextField->fieldID +
                          " generation=" + std::to_string(impl().nextField->generation) +
                          " dataVersion=" + std::to_string(impl().nextField->dataVersion));
        }
        observer->onSourceLoaded(*this);
        observer->onSourceChanged(*this);
        tryInitializeTimeline();
    } catch (const std::exception& error) {
        Log::Warning(Event::Render, "ParticleField: event=ADAPTER_REJECTED sourceId=" + getID() +
                                         " error=" + error.what());
        observer->onSourceError(*this, std::make_exception_ptr(std::runtime_error(error.what())));
    }
}

void RasterArraySource::tryLoadRuntimeField(const size_t index) {
    auto& candidate = runtimeFields.at(index);
    if (candidate.snapshot || !candidate.metadata || !candidate.fieldBytes) return;
    try {
        // A candidate is already a Core-owned runtime snapshot when it leaves
        // the adapter. Do not defer the required revision assignment until
        // timeline construction; that made valid preload data fail closed.
        candidate.snapshot = particle::OfflineFixtureParticleAdapter::load(
            candidate.descriptor.fixtureID, *candidate.metadata, *candidate.fieldBytes,
            ++nextRasterArrayFieldGeneration);
        Log::Info(Event::Render,
                  "ParticleField: runtime candidate ready fieldId=" + candidate.snapshot->fieldID +
                      " dataVersion=" + std::to_string(candidate.snapshot->dataVersion));
        tryInitializeTimeline();
    } catch (const std::exception& error) {
        Log::Warning(Event::Render, "ParticleField: event=ADAPTER_REJECTED sourceId=" + getID() +
                                         " candidateIndex=" + std::to_string(index) + " fieldId=" +
                                         candidate.descriptor.fixtureID + " generation=" +
                                         std::to_string(candidate.snapshot ? candidate.snapshot->generation : 0) +
                                         " validTime=" + std::to_string(candidate.snapshot ? candidate.snapshot->validTime : 0) +
                                         " error=" + error.what());
        observer->onSourceError(*this, std::make_exception_ptr(std::runtime_error(error.what())));
    }
}

void RasterArraySource::tryLoadLocalField(const size_t index) {
    auto& candidate = localFields.at(index);
    if (candidate.snapshot || !candidate.metadata || !candidate.fieldBytes) return;
    const auto result = particle::LocalParticleFieldFeed::load(*candidate.metadata, *candidate.fieldBytes);
    if (!result) {
        const auto fieldID = candidate.metadata->find("fieldId") == std::string::npos ? "" : "metadata";
        Log::Warning(Event::Render, "ParticleField: event=ADAPTER_REJECTED sourceId=" + getID() +
                                         " errorCode=" + particle::particleFieldAdapterErrorName(result.error) +
                                         " fieldId=" + fieldID + " payloadBytes=" + std::to_string(candidate.fieldBytes->size()) +
                                         " error=" + result.message);
        observer->onSourceError(*this, std::make_exception_ptr(std::runtime_error(result.message)));
        return;
    }
    candidate.snapshot = std::move(*result.snapshot);
    Log::Info(Event::Render, "ParticleField: event=PRODUCTION_FEED_LOAD fieldId=" + candidate.snapshot->fieldID +
                                  " generation=" + std::to_string(candidate.snapshot->generation) +
                                  " validTime=" + std::to_string(candidate.snapshot->validTime) +
                                  " payloadBytes=" + std::to_string(candidate.fieldBytes->size()));
    tryInitializeTimeline();
}

void RasterArraySource::tryInitializeTimeline() {
    if (timeline) return;
    if (!impl().localFields.empty()) {
        if (localFields.size() != impl().localFields.size() || localFields.size() < 2) return;
        for (const auto& candidate : localFields) if (!candidate.snapshot) return;
        std::vector<RasterArrayTimelineRuntime::Entry> entries;
        entries.reserve(localFields.size());
        for (const auto& candidate : localFields) {
            entries.push_back({std::make_shared<particle::ParticleFieldSnapshot>(*candidate.snapshot), candidate.snapshot->validTime});
        }
        try {
            if (submitInitialRuntimeFieldWindow(*localFields[0].snapshot, *localFields[1].snapshot) !=
                RasterArrayRuntimeFieldUpdateResult::Applied) {
                return;
            }
            timeline = std::make_shared<RasterArrayTimelineRuntime>(std::move(entries));
            timeline->setTargetFps(targetFps);
            auto updated = makeMutable<Impl>(impl());
            updated->timeline = timeline;
            baseImpl = std::move(updated);
            loaded = true;
            Log::Info(Event::Render, "ParticleTimeline: event=READY snapshots=" + std::to_string(localFields.size()));
            observer->onSourceLoaded(*this);
            observer->onSourceChanged(*this);
        } catch (const std::exception& error) {
            observer->onSourceError(*this, std::make_exception_ptr(std::runtime_error(error.what())));
        }
        return;
    }
    if (!loaded || !impl().field || !impl().nextField || runtimeFields.empty()) return;
    std::vector<RasterArrayTimelineRuntime::Entry> entries;
    entries.reserve(runtimeFields.size() + 2);
    entries.push_back({std::make_shared<particle::ParticleFieldSnapshot>(*impl().field),
                       impl().field->validTime ? impl().field->validTime : 0});
    entries.push_back({std::make_shared<particle::ParticleFieldSnapshot>(*impl().nextField),
                       impl().nextField->validTime ? impl().nextField->validTime : 40000});
    for (size_t index = 0; index < runtimeFields.size(); ++index) {
        if (!runtimeFields[index].snapshot) return;
        auto field = *runtimeFields[index].snapshot;
        entries.push_back({std::make_shared<particle::ParticleFieldSnapshot>(std::move(field)),
                           runtimeFields[index].descriptor.timestamp ? runtimeFields[index].descriptor.timestamp
                                                                       : 80000 + static_cast<uint64_t>(index) * 40000});
    }
    try {
        timeline = std::make_shared<RasterArrayTimelineRuntime>(std::move(entries));
        timeline->setTargetFps(targetFps);
        auto updated = makeMutable<Impl>(impl());
        updated->timeline = timeline;
        baseImpl = std::move(updated);
        Log::Info(Event::Render,
                  "ParticleTimeline: event=READY snapshots=" + std::to_string(runtimeFields.size() + 2) +
                      " cacheCapacity=" + std::to_string(timeline->cacheCapacity()));
        for (size_t index = 0; index < runtimeFields.size() + 2; ++index) {
            Log::Info(Event::Render,
                      "ParticleTimeline: event=TIMELINE_PRELOAD timelineId=" + getID() +
                          " index=" + std::to_string(index) + " cacheSize=" + std::to_string(runtimeFields.size() + 2) +
                          " capacity=" + std::to_string(timeline->cacheCapacity()));
        }
        observer->onSourceChanged(*this);
    } catch (const std::exception& error) {
        observer->onSourceError(*this, std::make_exception_ptr(std::runtime_error(error.what())));
    }
}

RasterArrayRuntimeFieldUpdateResult RasterArraySource::submitInitialRuntimeFieldWindow(
    particle::ParticleFieldSnapshot t0, particle::ParticleFieldSnapshot t1) {
    if (impl().field || impl().nextField || !t0.isValid() || !t1.isValid() || t0.generation == 0 ||
        t1.generation == 0 || t0.validTime >= t1.validTime || !t0.isTimeInterpolationCompatibleWith(t1)) {
        return rejectRuntimeFieldUpdate(RasterArrayRuntimeFieldUpdateResult::RejectedInvalidSnapshot,
                                        "INVALID_INITIAL_WINDOW");
    }
    baseImpl = makeMutable<Impl>(impl(), std::move(t0), std::move(t1));
    loaded = true;
    Log::Info(Event::Render, "ParticleField: event=SNAPSHOT_SUBMIT kind=INITIAL_T0_T1");
    return RasterArrayRuntimeFieldUpdateResult::Applied;
}

RasterArrayRuntimeFieldUpdateResult RasterArraySource::rejectRuntimeFieldUpdate(
    const RasterArrayRuntimeFieldUpdateResult result, const std::string& reason) const {
    Log::Warning(Event::Render,
                 "ParticleField: event=FIELD_WINDOW_UPDATE_REJECTED reason=" + reason + " sourceId=" + getID());
    return result;
}

RasterArrayRuntimeFieldUpdateResult RasterArraySource::updateRuntimeFieldWindow(particle::ParticleFieldSnapshot newT1) {
    if (!loaded || !impl().field || !impl().nextField) {
        return rejectRuntimeFieldUpdate(RasterArrayRuntimeFieldUpdateResult::RejectedNotLoaded, "SOURCE_NOT_LOADED");
    }
    if (!newT1.isValid() || newT1.generation == 0) {
        return rejectRuntimeFieldUpdate(RasterArrayRuntimeFieldUpdateResult::RejectedInvalidSnapshot,
                                        "INVALID_SNAPSHOT");
    }
    if (!impl().nextField->isTimeInterpolationCompatibleWith(newT1)) {
        return rejectRuntimeFieldUpdate(RasterArrayRuntimeFieldUpdateResult::RejectedIncompatibleSnapshot,
                                        "INCOMPATIBLE_SNAPSHOT");
    }
    if (particle::sameFieldGpuContent(*impl().nextField, newT1)) {
        Log::Info(Event::Render,
                  "ParticleField: event=FIELD_WINDOW_UPDATE_APPLIED result=NOOP sourceId=" + getID() +
                      " T0=" + impl().field->fieldID + " T1=" + impl().nextField->fieldID);
        return RasterArrayRuntimeFieldUpdateResult::Noop;
    }
    if (newT1.generation <= impl().nextField->generation) {
        return rejectRuntimeFieldUpdate(RasterArrayRuntimeFieldUpdateResult::RejectedNonMonotonicGeneration,
                                        "NON_MONOTONIC_GENERATION");
    }

    const auto oldT0 = impl().field->fieldID;
    const auto oldT1 = impl().nextField->fieldID;
    Log::Info(Event::Render,
              "ParticleField: event=FIELD_WINDOW_UPDATE_REQUEST sourceId=" + getID() +
                  " sequence=" + std::to_string(runtimeUpdateSequence + 1) + " oldT0=" + oldT0 + " oldT1=" + oldT1 +
                  " newT1=" + newT1.fieldID + " generation=" + std::to_string(newT1.generation));
    auto newT0 = *impl().nextField;
    baseImpl = makeMutable<Impl>(impl(), std::move(newT0), std::move(newT1));
    ++runtimeUpdateSequence;
    observer->onSourceChanged(*this);
    Log::Info(Event::Render,
              "ParticleField: event=FIELD_WINDOW_UPDATE_APPLIED result=APPLIED sourceId=" + getID() +
                  " sequence=" + std::to_string(runtimeUpdateSequence) + " newT0=" + impl().field->fieldID +
                  " newT1=" + impl().nextField->fieldID +
                  " generation=" + std::to_string(impl().nextField->generation));
    return RasterArrayRuntimeFieldUpdateResult::Applied;
}

RasterArrayRuntimeFieldUpdateResult RasterArraySource::advanceRuntimeFieldWindow() {
    if (timeline) {
        return timeline->step(1) ? RasterArrayRuntimeFieldUpdateResult::Applied
                                 : RasterArrayRuntimeFieldUpdateResult::RejectedNoCandidate;
    }
    if (nextRuntimeField >= runtimeFields.size()) {
        return rejectRuntimeFieldUpdate(RasterArrayRuntimeFieldUpdateResult::RejectedNoCandidate, "NO_PRELOADED_FIELD");
    }
    auto& candidate = runtimeFields[nextRuntimeField];
    if (!candidate.snapshot) {
        return rejectRuntimeFieldUpdate(RasterArrayRuntimeFieldUpdateResult::RejectedCandidateNotReady,
                                        "PRELOADED_FIELD_NOT_READY");
    }
    // A preloaded candidate is immutable content, not an active runtime
    // revision. Allocate its revision when it enters the authoritative window
    // so concurrent asset completion order cannot make it stale.
    auto newT1 = *candidate.snapshot;
    newT1.generation = ++nextRasterArrayFieldGeneration;
    const auto result = updateRuntimeFieldWindow(std::move(newT1));
    if (result == RasterArrayRuntimeFieldUpdateResult::Applied || result == RasterArrayRuntimeFieldUpdateResult::Noop) {
        ++nextRuntimeField;
    }
    return result;
}

RasterArrayRuntimeFieldUpdateResult RasterArraySource::repeatRuntimeFieldWindow() {
    if (timeline) {
        Log::Info(Event::Render, "ParticleField: event=FIELD_WINDOW_UPDATE_APPLIED result=NOOP sourceId=" + getID());
        return RasterArrayRuntimeFieldUpdateResult::Noop;
    }
    if (!loaded || !impl().nextField) {
        return rejectRuntimeFieldUpdate(RasterArrayRuntimeFieldUpdateResult::RejectedNotLoaded, "SOURCE_NOT_LOADED");
    }
    return updateRuntimeFieldWindow(*impl().nextField);
}

const char* rasterArrayHostFieldSubmitResultName(const RasterArrayHostFieldSubmitResult result) {
    switch (result) {
        case RasterArrayHostFieldSubmitResult::RejectedSourceMode:
            return "REJECTED_SOURCE_MODE";
        case RasterArrayHostFieldSubmitResult::RejectedAdapter:
            return "REJECTED_ADAPTER";
        case RasterArrayHostFieldSubmitResult::RejectedIncompatibleSnapshot:
            return "REJECTED_INCOMPATIBLE_SNAPSHOT";
        case RasterArrayHostFieldSubmitResult::RejectedNonMonotonicSnapshot:
            return "REJECTED_NON_MONOTONIC_SNAPSHOT";
        case RasterArrayHostFieldSubmitResult::RejectedCapacity:
            return "REJECTED_CAPACITY";
        case RasterArrayHostFieldSubmitResult::AcceptedPendingInitialWindow:
            return "ACCEPTED_PENDING_INITIAL_WINDOW";
        case RasterArrayHostFieldSubmitResult::InitialWindowReady:
            return "INITIAL_WINDOW_READY";
        case RasterArrayHostFieldSubmitResult::AcceptedNextSnapshot:
            return "ACCEPTED_NEXT_SNAPSHOT";
        case RasterArrayHostFieldSubmitResult::Duplicate:
            return "DUPLICATE_NOOP";
    }
    return "UNKNOWN";
}

RasterArrayHostFieldSubmitResult RasterArraySource::submitHostField(const std::string& metadataText,
                                                                    const std::vector<char>& payload) {
    Log::Info(Event::Render,
              "ParticleField: event=HOST_FIELD_SUBMIT_REQUEST sourceId=" + getID() + " metadataBytes=" +
                  std::to_string(metadataText.size()) + " payloadBytes=" + std::to_string(payload.size()));
    if (!impl().hostFieldSubmission) {
        Log::Warning(
            Event::Render,
            "ParticleField: event=HOST_FIELD_SUBMIT_REJECTED sourceId=" + getID() + " result=REJECTED_SOURCE_MODE");
        return RasterArrayHostFieldSubmitResult::RejectedSourceMode;
    }

    auto decoded = particle::LocalParticleFieldFeed::load(metadataText, payload);
    if (!decoded) {
        Log::Warning(
            Event::Render,
            "ParticleField: event=HOST_FIELD_SUBMIT_REJECTED sourceId=" + getID() +
                " result=REJECTED_ADAPTER errorCode=" + particle::particleFieldAdapterErrorName(decoded.error) +
                " payloadBytes=" + std::to_string(payload.size()) + " error=" + decoded.message);
        return RasterArrayHostFieldSubmitResult::RejectedAdapter;
    }

    Log::Info(Event::Render,
              "ParticleField: event=HOST_FIELD_LEGACY_VECTOR_ONLY sourceId=" + getID() +
                  " payloadBytes=" + std::to_string(payload.size()));
    return submitDecodedHostField(std::move(*decoded.snapshot));
}

RasterArrayHostFieldSubmitResult RasterArraySource::submitHostField(
    const std::string& metadataText,
    const particle::ParticleFieldPayloadBundle& payloads) {
    const auto bytes = [](const std::optional<std::vector<char>>& plane) {
        return plane ? std::to_string(plane->size()) : std::string("absent");
    };
    Log::Info(Event::Render,
              "ParticleField: event=HOST_FIELD_V2_SUBMIT_REQUEST sourceId=" + getID() + " metadataBytes=" +
                  std::to_string(metadataText.size()) + " scalarBytes=" + bytes(payloads.scalar) +
                  " vectorBytes=" + bytes(payloads.vector) + " validityBytes=" + bytes(payloads.validity));
    if (!impl().hostFieldSubmission) {
        return RasterArrayHostFieldSubmitResult::RejectedSourceMode;
    }
    auto decoded = particle::LocalParticleFieldFeed::load(metadataText, payloads);
    if (!decoded) {
        Log::Warning(Event::Render,
                     "ParticleField: event=HOST_FIELD_V2_SUBMIT_REJECTED sourceId=" + getID() +
                         " result=REJECTED_ADAPTER errorCode=" + particle::particleFieldAdapterErrorName(decoded.error) +
                         " error=" + decoded.message);
        return RasterArrayHostFieldSubmitResult::RejectedAdapter;
    }
    return submitDecodedHostField(std::move(*decoded.snapshot));
}

RasterArrayHostFieldSubmitResult RasterArraySource::submitDecodedHostField(particle::ParticleFieldSnapshot snapshot) {
    Log::Info(Event::Render,
              "ParticleField: event=HOST_FIELD_SNAPSHOT_READY sourceId=" + getID() + " fieldId=" + snapshot.fieldID +
                  " generation=" + std::to_string(snapshot.generation) +
                  " validTime=" + std::to_string(snapshot.validTime) + " width=" + std::to_string(snapshot.width) +
                  " height=" + std::to_string(snapshot.height) + " scalar=" +
                  (snapshot.hasScalar() ? "present" : "absent") + " vector=" +
                  (snapshot.hasVector() ? "present" : "absent") + " validity=present");

    std::shared_ptr<const particle::ParticleFieldSnapshot> previous;
    if (timeline) {
        previous = timeline->lastEntry().field;
    } else if (pendingHostField) {
        previous = std::make_shared<const particle::ParticleFieldSnapshot>(*pendingHostField);
    }
    if (previous && particle::sameFieldGpuContent(*previous, snapshot)) {
        Log::Info(Event::Render,
                  "ParticleField: event=HOST_FIELD_SUBMIT_DUPLICATE sourceId=" + getID() +
                      " fieldId=" + snapshot.fieldID + " generation=" + std::to_string(snapshot.generation) +
                      " validTime=" + std::to_string(snapshot.validTime) + " result=DUPLICATE_NOOP");
        return RasterArrayHostFieldSubmitResult::Duplicate;
    }
    if (previous && (snapshot.validTime <= previous->validTime || snapshot.generation <= previous->generation)) {
        Log::Warning(Event::Render,
                     "ParticleField: event=HOST_FIELD_SUBMIT_REJECTED sourceId=" + getID() +
                         " fieldId=" + snapshot.fieldID + " generation=" + std::to_string(snapshot.generation) +
                         " validTime=" + std::to_string(snapshot.validTime) +
                         " result=REJECTED_NON_MONOTONIC_SNAPSHOT");
        return RasterArrayHostFieldSubmitResult::RejectedNonMonotonicSnapshot;
    }
    if (previous && !previous->isTimeInterpolationCompatibleWith(snapshot)) {
        Log::Warning(Event::Render,
                     "ParticleField: event=HOST_FIELD_SUBMIT_REJECTED sourceId=" + getID() +
                         " fieldId=" + snapshot.fieldID + " generation=" + std::to_string(snapshot.generation) +
                         " validTime=" + std::to_string(snapshot.validTime) + " result=REJECTED_INCOMPATIBLE_SNAPSHOT");
        return RasterArrayHostFieldSubmitResult::RejectedIncompatibleSnapshot;
    }

    if (!pendingHostField && !timeline) {
        pendingHostField = std::move(snapshot);
        Log::Info(Event::Render,
                  "ParticleField: event=HOST_FIELD_SUBMIT_ACCEPTED sourceId=" + getID() +
                      " kind=PENDING_INITIAL_T0 fieldId=" + pendingHostField->fieldID +
                      " result=ACCEPTED_PENDING_INITIAL_WINDOW");
        return RasterArrayHostFieldSubmitResult::AcceptedPendingInitialWindow;
    }

    if (!timeline) {
        const auto initialT0 = *pendingHostField;
        const auto initialT1 = snapshot;
        if (submitInitialRuntimeFieldWindow(initialT0, initialT1) != RasterArrayRuntimeFieldUpdateResult::Applied) {
            return RasterArrayHostFieldSubmitResult::RejectedIncompatibleSnapshot;
        }
        std::vector<RasterArrayTimelineRuntime::Entry> entries;
        entries.push_back({std::make_shared<const particle::ParticleFieldSnapshot>(initialT0), initialT0.validTime});
        entries.push_back({std::make_shared<const particle::ParticleFieldSnapshot>(initialT1), initialT1.validTime});
        timeline = std::make_shared<RasterArrayTimelineRuntime>(std::move(entries));
        timeline->setTargetFps(targetFps);
        auto updated = makeMutable<Impl>(impl());
        updated->timeline = timeline;
        baseImpl = std::move(updated);
        pendingHostField.reset();
        observer->onSourceChanged(*this);
        Log::Info(Event::Render,
                  "ParticleField: event=HOST_FIELD_INITIAL_WINDOW_READY sourceId=" + getID() +
                      " T0=" + initialT0.fieldID + " T1=" + initialT1.fieldID + " result=INITIAL_WINDOW_READY");
        Log::Info(Event::Render, "ParticleTimeline: event=READY snapshots=2 source=HOST_FIELD_SUBMISSION");
        return RasterArrayHostFieldSubmitResult::InitialWindowReady;
    }

    if (timeline->size() >= timeline->cacheCapacity()) {
        Log::Warning(Event::Render,
                     "ParticleField: event=HOST_FIELD_SUBMIT_REJECTED sourceId=" + getID() +
                         " fieldId=" + snapshot.fieldID + " result=REJECTED_CAPACITY");
        return RasterArrayHostFieldSubmitResult::RejectedCapacity;
    }
    const auto fieldID = snapshot.fieldID;
    const auto generation = snapshot.generation;
    const auto validTime = snapshot.validTime;
    if (!timeline->append({std::make_shared<const particle::ParticleFieldSnapshot>(std::move(snapshot)), validTime})) {
        return RasterArrayHostFieldSubmitResult::RejectedCapacity;
    }
    Log::Info(Event::Render,
              "ParticleField: event=HOST_FIELD_NEXT_SUBMIT sourceId=" + getID() + " fieldId=" + fieldID +
                  " generation=" + std::to_string(generation) + " validTime=" + std::to_string(validTime) +
                  " snapshots=" + std::to_string(timeline->size()) + " result=ACCEPTED_NEXT_SNAPSHOT");
    return RasterArrayHostFieldSubmitResult::AcceptedNextSnapshot;
}

bool RasterArraySource::clearHostFields() {
    if (!impl().hostFieldSubmission) {
        Log::Warning(Event::Render, "WeatherRuntime: event=FIELD_WINDOW_CLEAR_REJECTED reason=SOURCE_MODE sourceId=" + getID());
        return false;
    }
    if (timeline) timeline->pause();
    Log::Info(Event::Render, "WeatherRuntime: event=TIMELINE_STOP sourceId=" + getID());
    pendingHostField.reset();
    timeline.reset();
    auto updated = makeMutable<Impl>(impl());
    updated->timeline.reset();
    updated->field.reset();
    updated->nextField.reset();
    baseImpl = std::move(updated);
    loaded = true; // An empty Host ingress is loaded; it has no drawable field.
    runtimeUpdateSequence = 0;
    observer->onSourceChanged(*this);
    Log::Info(Event::Render, "WeatherRuntime: event=FIELD_WINDOW_CLEAR sourceId=" + getID());
    Log::Info(Event::Render, "WeatherRuntime: event=SNAPSHOT_CLEAR sourceId=" + getID());
    return true;
}

bool RasterArraySource::seekRuntimeTimelineToTimestamp(const uint64_t timestamp) {
    if (!timeline) {
        Log::Warning(Event::Render, "WeatherRuntime: event=SEEK_TIMESTAMP_REJECTED reason=TIMELINE_NOT_READY sourceId=" + getID());
        return false;
    }
    const bool accepted = timeline->seekTimestamp(timestamp);
    if (!accepted) {
        Log::Warning(Event::Render, "WeatherRuntime: event=SEEK_TIMESTAMP_REJECTED reason=OUT_OF_RANGE sourceId=" + getID() +
                                     " timestamp=" + std::to_string(timestamp));
    }
    return accepted;
}

bool RasterArraySource::controlRuntimeTimeline(const int command, const float value) {
    if (!timeline) {
        Log::Warning(Event::Render, "ParticleTimeline: event=REJECTED reason=TIMELINE_NOT_READY sourceId=" + getID());
        return false;
    }
    switch (command) {
        case 0:
            return timeline->play();
        case 1:
            return timeline->pause();
        case 2:
            return timeline->setSpeed(value);
        case 3:
            return timeline->step(-1);
        case 4:
            return timeline->step(1);
        case 5:
            return timeline->seek(static_cast<size_t>(value));
        case 6:
            return timeline->injectGap();
        case 7:
            return timeline->injectInvalid();
        case 8:
            return timeline->recover();
        default:
            return false;
    }
}

bool RasterArraySource::setRuntimeTargetFps(const uint8_t requested) {
    RasterArrayTargetFps next;
    switch (requested) {
        case 0: next = RasterArrayTargetFps::Auto; break;
        case 30: next = RasterArrayTargetFps::Fps30; break;
        case 60: next = RasterArrayTargetFps::Fps60; break;
        default: return false;
    }
    targetFps = next;
    if (timeline) timeline->setTargetFps(next);
    observer->onSourceChanged(*this);
    return true;
}

uint8_t RasterArraySource::runtimeTargetFps() const {
    return static_cast<uint8_t>(targetFps);
}

bool RasterArraySource::supportsLayerType(const LayerTypeInfo* info) const {
    if (!info) return false;
    const std::string type{info->type};
    // Both displays consume the exact same canonical Source-owned snapshots.
    // WeatherColorLayer is intentionally a separate renderer, not a second
    // source, cache, timeline or field normalizer.
    return type == "raster-particle" || type == "weather-color";
}

Mutable<Source::Impl> RasterArraySource::createMutable() const noexcept {
    return staticMutableCast<Source::Impl>(makeMutable<Impl>(impl()));
}

} // namespace mbgl::style