streaming_window.cpp

/**
 * @file streaming_window.cpp
 * @brief Canonical Doxygen file header for ThemisDB-generated maturity metadata.
 * @version 0.0.32
 * @note Maturity: 🟢 PRODUCTION-READY
 * @note Score: 85/100
 * @note Gap Summary: total=20; TODO=18, Stub=1, Unimpl=0, Mock=1, Sim=0, Debt=0, C=9, H=10, M=17, L=0
 * @note Status: Production Ready
 * @note This block is auto-generated and will be overwritten.
 */

/*
 * ThemisDB | File: streaming_window.cpp | Version: 0.0.32 | Last Modified: 2026-05-31 12:49:01
 * Author: makr-code | Maturity: 🟢 PRODUCTION-READY | Score: 97/100 | Lines: 1348
 * Gap Summary: total=20; TODO=18, Stub=1, Unimpl=0, Mock=1, Sim=0, Debt=0, C=9, H=19, M=26, L=0
 * PR History (last 5): #4929 [Docs][analytics] Refresh m... (2026-05-10) | #4339 Analytics module: stats.h u... (2026-03-19) | #4327 fix(analytics): configurabl... (2026-03-18) | #3478 docs(analytics): sync READM... (2026-03-12) | #3326 [analytics] Mark unit test ... (2026-03-12)
 * Status: Production Ready
 * (Automatisch generiert, Änderungen werden überschrieben)
 */

/**
 * Streaming Aggregation Windows - Implementation
 *
 * @module Streaming
 *
 * Full implementation of all window types declared in
 * include/analytics/streaming_window.h.
 *
 * Data flow:
 *   Window::ingest(Event) → advances watermark; assigns event to window(s)
 *   Window::flush()       → emits WindowResult for all windows whose close
 *                          condition is met (end < watermark - max_out_of_orderness)
 *   idle timeout loop     → advances watermark to now when no events arrive for
 *                          idle_timeout; closes and emits expired windows
 *
 * Error paths:
 *   - Late events (event_time < current_watermark): silently accepted into
 *     open windows; sets WindowResult::is_late_firing = true on emit.
 *   - Events with negative timestamps: treated as out-of-order; handled by
 *     watermark tolerance (max_out_of_orderness).
 *   - Concurrent ingest + flush: protected by per-window mutex; flush() and
 *     ingest() are serialised.
 *   - Session gap overflow (gap > max_session_gap): session is closed and
 *     a new one opened; no error thrown.
 *
 * Cross-links:
 *   include/analytics/streaming_window.h — IStreamingWindow, WindowResult
 *   src/analytics/streaming_join.cpp     — downstream consumer of window output
 *   tests/analytics/test_streaming_window.cpp — coverage
 *
 * Design notes:
 *  - Event-time semantics with configurable watermarking
 *  - Thread-safe ingestion (mutex per window)
 *  - Watermark advances monotonically on each ingested record
 *  - Shared aggregation computation logic via anonymous namespace helpers
 *  - Session expiry via background thread (SessionWindow only)
 *
 * Open TODOs (tracked here per code-review requirements; see also
 * src/analytics/FUTURE_ENHANCEMENTS.md §13):
 *
 * TODO(v1.8.0) #1: RESOLVED — idle_timeout background thread added to
 *   TumblingWindow and SlidingWindow (idleTimeoutLoop). When no events arrive
 *   for idle_timeout duration, the watermark is advanced to now –
 *   max_out_of_orderness and expired windows are closed and emitted.
 *
 * TODO(v1.8.0) #2: RESOLVED — partition_key stored in InternalWindow for both
 *   TumblingWindow and SlidingWindow; propagated to WindowResult::partition_key
 *   in computeResult().  ensureWindowsExist() updated to accept partition_key.
 *
 * TODO(v1.8.0) #3: RESOLVED — SessionWindow::expiryLoop now passes
 *   s.has_late_records to computeResult() so timer-driven session closures
 *   correctly set is_late_firing on the emitted result.
 *
 * TODO(v1.8.0) #4: RESOLVED — StreamingWindowPipeline::Config gains
 *   session_expiry_interval_ms; the session() factory accepts an optional
 *   expiry_interval_ms parameter and the build() SESSION case forwards it to
 *   SessionWindowConfig::session_expiry_check_interval_ms.
 *
 * TODO(v1.8.0) #5: RESOLVED — O(N) duplicate-detection loop in
 *   SlidingWindow::ensureWindowsExist and HoppingWindow::ensureWindowsExist
 *   replaced with an unordered_set<int64_t> (window_start_set_) for O(1)
 *   lookup.  Set is kept in sync on window creation and pruning.
 *
 * TODO(v1.8.0) #6: RESOLVED — calcPercentile() now accepts a const reference;
 *   the O(N) copy-per-call-site is eliminated. The single scratch copy occurs
 *   inside themis::analytics::detail::computePercentile (stats.h).
 *
 * TODO(v1.8.0) #7: RESOLVED — SessionWindow::computeResult() now accepts a
 *   bool late parameter and sets r.is_late_firing accordingly.  ingest() and
 *   flush() pass s.has_late_records; expiryLoop() does the same.
 *
 * TODO(v1.8.0) #8: RESOLVED — The double-close guard is already present via
 *   the !w.closed check inside closeExpiredWindows(); flush() delegates to
 *   closeExpiredWindows(INT64_MAX) which respects the flag, so no duplicate
 *   results are emitted on shutdown.
 */

#include "analytics/streaming_window.h"
#include <stdexcept>
#include "analytics/detail/stats.h"

#include <algorithm>
#include <cassert>
#include <cmath>
#include <limits>
#include <random>
#include <set>
#include <spdlog/spdlog.h>
#include <sstream>

#include "analytics/detail/stats.h"

namespace themisdb {
namespace analytics {

// ============================================================================
// Anonymous helpers
// ============================================================================

namespace {

std::string genId() {
    static std::atomic<uint64_t> counter{1};
    uint64_t c = counter.fetch_add(1, std::memory_order_relaxed);
    uint64_t t = static_cast<uint64_t>(
        std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch())
            .count());

    // Stable pseudo-UUID layout based on monotonic time + process-local counter.
    uint64_t a = t;
    uint64_t b = (c << 32) ^ (t >> 7);
    char buf[37];
    std::snprintf(buf, sizeof(buf), "%08x-%04x-%04x-%04x-%012llx", static_cast<unsigned>(a >> 32),
                  static_cast<unsigned>((a >> 16) & 0xFFFF), static_cast<unsigned>(a & 0xFFFF),
                  static_cast<unsigned>((b >> 48) & 0xFFFF),
                  static_cast<unsigned long long>(b & 0x0000'FFFF'FFFF'FFFFULL));
    return std::string(buf);
}

int64_t toMicros(const std::chrono::system_clock::time_point &tp) {
    return std::chrono::duration_cast<std::chrono::microseconds>(tp.time_since_epoch()).count();
}

std::chrono::system_clock::time_point fromMicros(int64_t us) {
    return std::chrono::system_clock::time_point(std::chrono::microseconds(us));
}

double toDouble(const RecordValue &v) {
    if (auto *d = std::get_if<double>(&v)) {
        return *d;
    }
    if (auto *i = std::get_if<int64_t>(&v)) {
        return static_cast<double>(*i);
    }
    if (auto *b = std::get_if<bool>(&v)) {
        return *b ? 1.0 : 0.0;
    }
    return 0.0;
}

std::string rvToString(const RecordValue &v) {
    if (std::holds_alternative<std::monostate>(v)) {
        return "";
    }
    if (auto *s = std::get_if<std::string>(&v)) {
        return *s;
    }
    if (auto *i = std::get_if<int64_t>(&v)) {
        return std::to_string(*i);
    }
    if (auto *d = std::get_if<double>(&v)) {
        return std::to_string(*d);
    }
    if (auto *b = std::get_if<bool>(&v)) {
        return *b ? "true" : "false";
    }
    return "";
}

/** Compute percentile (p in [0,100]) from an unsorted values vector.
 *  Delegates to themis::analytics::detail::computePercentile (stats.h) —
 *  fixes TODO(v1.8.0) #6: was taking by value (O(N) copy per call-site).
 */
double calcPercentile(const std::vector<double> &vals, double p) {
    return themis::analytics::detail::computePercentile(vals, p);
}

/**
 * Compute all aggregations from a flat list of records.
 */
std::vector<AggregatedValue> computeAggregations(const std::vector<StreamRecord> &records,
                                                 const std::vector<WindowAggregateSpec> &specs) {
    std::vector<AggregatedValue> results;
    results.reserve(specs.size());

    for (const auto &spec : specs) {
        AggregatedValue av;
        av.name  = spec.name;
        av.func  = spec.func;
        av.count = static_cast<uint64_t>(records.size());

        // Collect numeric values for all records
        std::vector<double> nums;
        RecordValue first_val{std::monostate{}};
        RecordValue last_val{std::monostate{}};
        bool has_first = false;
        double sum_val = 0.0;
        double min_val = std::numeric_limits<double>::max();
        double max_val = std::numeric_limits<double>::lowest();

        for (const auto &rec : records) {
            RecordValue fv{std::monostate{}};
            if (!spec.field.empty()) {
                auto it = rec.fields.find(spec.field);
                if (it != rec.fields.end()) {
                    fv = it->second;
                }
            }
            double d = toDouble(fv);
            sum_val += d;
            if (d < min_val)
                min_val = d;
            if (d > max_val)
                max_val = d;
            nums.push_back(d);
            last_val = fv;
            if (!has_first) {
                first_val = fv;
                has_first = true;
            }
        }

        uint64_t n = static_cast<uint64_t>(records.size());
        switch (spec.func) {
            case AggFunc::COUNT:
                av.value = static_cast<int64_t>(n);
                break;
            case AggFunc::SUM:
                av.value = sum_val;
                break;
            case AggFunc::AVG:
                av.value = (n > 0) ? sum_val / static_cast<double>(n) : 0.0;
                break;
            case AggFunc::MIN:
                av.value = (n > 0) ? min_val : 0.0;
                break;
            case AggFunc::MAX:
                av.value = (n > 0) ? max_val : 0.0;
                break;
            case AggFunc::STDDEV: {
                if (n < 2) {
                    av.value = 0.0;
                    break;
                }
                double mean = sum_val / static_cast<double>(n);
                double var  = 0.0;
                for (double v : nums) {
                    var += (v - mean) * (v - mean);
                }
                av.value = std::sqrt(var / static_cast<double>(n - 1));
                break;
            }
            case AggFunc::VARIANCE: {
                if (n < 2) {
                    av.value = 0.0;
                    break;
                }
                double mean = sum_val / static_cast<double>(n);
                double var  = 0.0;
                for (double v : nums) {
                    var += (v - mean) * (v - mean);
                }
                av.value = var / static_cast<double>(n - 1);
                break;
            }
            case AggFunc::PERCENTILE:
                av.value = calcPercentile(nums, spec.percentile_p);
                break;
            case AggFunc::FIRST:
                av.value = first_val;
                break;
            case AggFunc::LAST:
                av.value = last_val;
                break;
            case AggFunc::DISTINCT_COUNT: {
                // Only count values where the field is actually present (not monostate).
                // This avoids conflating a missing field with an empty-string value.
                std::set<std::string> distinct;
                for (const auto &rec : records) {
                    if (!spec.field.empty()) {
                        auto it = rec.fields.find(spec.field);
                        if (it != rec.fields.end() && !std::holds_alternative<std::monostate>(it->second)) {
                            distinct.insert(rvToString(it->second));
                        }
                    }
                }
                av.value = static_cast<int64_t>(distinct.size());
                break;
            }
        }

        results.push_back(std::move(av));
    }
    return results;
}

} // anonymous namespace

// ============================================================================
// WindowResult
// ============================================================================

std::optional<RecordValue> WindowResult::get(const std::string &agg_name) const {
    for (const auto &av : aggregations) {
        if (av.name == agg_name) {
            return av.value;
        }
    }
    return std::nullopt;
}

// ============================================================================
// makeRecord helper
// ============================================================================

StreamRecord makeRecord(const std::string &id, std::chrono::system_clock::time_point event_time,
                        const std::string &partition_key,
                        std::initializer_list<std::pair<std::string, RecordValue>> fields) {
    StreamRecord r;
    r.record_id     = id.empty() ? genId() : id;
    r.event_time    = event_time;
    r.ingest_time   = std::chrono::system_clock::now();
    r.partition_key = partition_key;
    for (auto &[k, v] : fields) {
        r.fields[k] = v;
    }
    return r;
}

// ============================================================================
// TumblingWindow
// ============================================================================

TumblingWindow::TumblingWindow(const TumblingWindowConfig &config)
    : config_(config), callback_(), agg_specs_(), open_windows_(), watermark_us_(0), windows_opened_(0),
      windows_closed_(0), records_ingested_(0), records_dropped_(0), late_records_(0), results_emitted_(0),
      idle_running_(false), last_event_us_(0) {
    agg_specs_.reserve(16);
    if (config_.watermark.idle_timeout.count() > 0) {
        idle_running_ = true;
        idle_thread_  = std::thread([this] { idleTimeoutLoop(); });
    }
}

TumblingWindow::~TumblingWindow() {
    if (idle_running_) {
        idle_running_ = false;
        idle_cv_.notify_all();
        if (idle_thread_.joinable()) {
            idle_thread_.join();
        }
    }
    open_windows_.clear();
    agg_specs_.clear();
    callback_ = {};
}

std::unique_ptr<TumblingWindow> createTumblingWindow(const TumblingWindowConfig &config) {
    return std::make_unique<TumblingWindow>(config);
}

void TumblingWindow::addAggregation(const WindowAggregateSpec &spec) {
    std::lock_guard lk(mutex_);
    agg_specs_.push_back(spec);
}

void TumblingWindow::setResultCallback(ResultCallback cb) {
    std::lock_guard lk(mutex_);
    callback_ = std::move(cb);
}

int64_t TumblingWindow::slotIndex(const std::chrono::system_clock::time_point &tp) const {
    int64_t us = toMicros(tp);
    int64_t sz = static_cast<int64_t>(config_.size.count()) * 1000LL; // ms → us
    return (sz > 0) ? (us / sz) : 0;
}

std::chrono::system_clock::time_point TumblingWindow::slotStart(int64_t idx) const {
    int64_t sz_us = static_cast<int64_t>(config_.size.count()) * 1000LL;
    return fromMicros(idx * sz_us);
}

void TumblingWindow::updateWatermark(const std::chrono::system_clock::time_point &event_time) {
    int64_t ev_us  = toMicros(event_time);
    int64_t tol    = config_.watermark.max_out_of_orderness.count() * 1000LL;
    int64_t new_wm = ev_us - tol;
    // Monotonically advance
    int64_t old_wm = watermark_us_.load(std::memory_order_relaxed);
    while (
        new_wm > old_wm
        && !watermark_us_.compare_exchange_weak(old_wm, new_wm, std::memory_order_release, std::memory_order_relaxed)) {
    }
}

std::vector<WindowResult> TumblingWindow::closeExpiredWindows(int64_t watermark_us) {
    // Called with mutex_ held. Returns results to emit; callers fire callbacks
    // outside the lock to prevent re-entrant deadlock.
    std::vector<WindowResult> pending;
    for (auto it = open_windows_.begin(); it != open_windows_.end();) {
        if (toMicros(it->second.end) <= watermark_us) {
            InternalWindow closed = std::move(it->second);
            it                    = open_windows_.erase(it);

            if (config_.emit_empty_windows || !closed.records.empty()) {
                pending.push_back(computeResult(closed, false));
                ++results_emitted_;
            }
            ++windows_closed_;
        } else {
            ++it;
        }
    }
    return pending;
}

WindowResult TumblingWindow::computeResult(const InternalWindow &win, bool late) const {
    WindowResult r;
    r.window_id      = genId();
    r.window_start   = win.start;
    r.window_end     = win.end;
    r.partition_key  = win.partition_key;
    r.record_count   = win.records.size();
    r.aggregations   = computeAggregations(win.records, agg_specs_);
    r.is_late_firing = late;
    return r;
}

bool TumblingWindow::ingest(const StreamRecord &record) {
    ++records_ingested_;
    updateWatermark(record.event_time);
    int64_t wm = watermark_us_.load(std::memory_order_acquire);

    // Check if record is too old
    int64_t ev_us = toMicros(record.event_time);
    if (ev_us < wm && !config_.watermark.allow_late_data) {
        ++late_records_;
        ++records_dropped_;
        spdlog::debug("TumblingWindow: dropped late record (event={} < watermark={})", ev_us, wm);
        return false;
    }

    // Update last_event_us_ monotonically for idle-timeout tracking
    int64_t old_last = last_event_us_.load(std::memory_order_relaxed);
    while (ev_us > old_last
           && !last_event_us_.compare_exchange_weak(old_last, ev_us, std::memory_order_release,
                                                    std::memory_order_relaxed)) {
    }

    std::vector<WindowResult> pending;
    ResultCallback cb;
    {
        std::lock_guard lk(mutex_);

        int64_t idx = slotIndex(record.event_time);
        if (open_windows_.find(idx) == open_windows_.end()) {
            InternalWindow win;
            win.start          = slotStart(idx);
            win.end            = win.start + config_.size;
            win.partition_key  = record.partition_key;
            open_windows_[idx] = std::move(win);
            ++windows_opened_;
        }

        if (ev_us < wm && config_.watermark.allow_late_data) {
            ++late_records_;
        }
        open_windows_[idx].records.push_back(record);

        pending = closeExpiredWindows(wm);
        cb      = callback_;
    } // mutex_ released

    // BUG 3 FIX: fire callbacks outside the lock to prevent re-entrant deadlock.
    if (cb) {
        for (auto& r : pending) {
            try { cb(r); } catch (...) {}
        }
    }
    return true;
}

void TumblingWindow::flush() {
    std::vector<WindowResult> pending;
    ResultCallback cb;
    {
        std::lock_guard lk(mutex_);
        pending = closeExpiredWindows(std::numeric_limits<int64_t>::max());
        cb      = callback_;
    } // mutex_ released
    if (cb) {
        for (auto& r : pending) {
            try { cb(r); } catch (...) {}
        }
    }
}

WindowStats TumblingWindow::getStats() const {
    return WindowStats{windows_opened_.load(),  windows_closed_.load(), records_ingested_.load(),
                       records_dropped_.load(), late_records_.load(),   results_emitted_.load()};
}

void TumblingWindow::idleTimeoutLoop() {
    while (idle_running_) {
        {
            std::unique_lock lk(idle_mutex_);
            idle_cv_.wait_for(lk, config_.watermark.idle_timeout, [this] { return !idle_running_.load(); });
        }
        if (!idle_running_) {
            break;
        }

        auto now_us        = toMicros(std::chrono::system_clock::now());
        int64_t last       = last_event_us_.load(std::memory_order_acquire);
        int64_t timeout_us = config_.watermark.idle_timeout.count() * 1000LL;
        if (last > 0 && (now_us - last) < timeout_us) {
            continue;
        }

        int64_t tol_us = config_.watermark.max_out_of_orderness.count() * 1000LL;
        int64_t new_wm = now_us - tol_us;
        int64_t old_wm = watermark_us_.load(std::memory_order_relaxed);
        while (new_wm > old_wm
               && !watermark_us_.compare_exchange_weak(old_wm, new_wm, std::memory_order_release,
                                                       std::memory_order_relaxed)) {
        }
        int64_t wm = watermark_us_.load(std::memory_order_acquire);

        std::vector<WindowResult> pending;
        ResultCallback cb;
        {
            std::lock_guard lk(mutex_);
            pending = closeExpiredWindows(wm);
            cb      = callback_;
        }
        if (cb) {
            for (auto& r : pending) {
                try { cb(r); } catch (...) {}
            }
        }
    }
}

// ============================================================================
// SlidingWindow
// ============================================================================

SlidingWindow::SlidingWindow(const SlidingWindowConfig &config)
    : config_(config), callback_(), agg_specs_(), windows_(), watermark_us_(0), windows_opened_(0), windows_closed_(0),
      records_ingested_(0), records_dropped_(0), late_records_(0), results_emitted_(0), idle_running_(false),
      last_event_us_(0) {
    agg_specs_.reserve(16);
    if (config_.watermark.idle_timeout.count() > 0) {
        idle_running_ = true;
        idle_thread_  = std::thread([this] { idleTimeoutLoop(); });
    }
}

SlidingWindow::~SlidingWindow() {
    if (idle_running_) {
        idle_running_ = false;
        idle_cv_.notify_all();
        if (idle_thread_.joinable()) {
            idle_thread_.join();
        }
    }
    flush();
    agg_specs_.clear();
    callback_ = {};
}

std::unique_ptr<SlidingWindow> createSlidingWindow(const SlidingWindowConfig &config) {
    return std::make_unique<SlidingWindow>(config);
}

void SlidingWindow::addAggregation(const WindowAggregateSpec &spec) {
    std::lock_guard lk(mutex_);
    agg_specs_.push_back(spec);
}

void SlidingWindow::setResultCallback(ResultCallback cb) {
    std::lock_guard lk(mutex_);
    callback_ = std::move(cb);
}

std::string SlidingWindow::generateId() {
    return genId();
}

void SlidingWindow::updateWatermark(const std::chrono::system_clock::time_point &event_time) {
    int64_t ev_us  = toMicros(event_time);
    int64_t tol_us = config_.watermark.max_out_of_orderness.count() * 1000LL;
    int64_t new_wm = ev_us - tol_us;
    int64_t old_wm = watermark_us_.load(std::memory_order_relaxed);
    while (
        new_wm > old_wm
        && !watermark_us_.compare_exchange_weak(old_wm, new_wm, std::memory_order_release, std::memory_order_relaxed)) {
    }
}

void SlidingWindow::ensureWindowsExist(const std::chrono::system_clock::time_point &event_time,
                                       const std::string &partition_key) {
    // Called with mutex_ held
    // A record at event_time must appear in all windows [start, start+size) where
    //   start ∈ { ..., t - (t % slide), t - (t % slide) + slide, ... }
    // covering [event_time - size + slide .. event_time]
    int64_t ev_us    = toMicros(event_time);
    int64_t size_us  = config_.size.count() * 1000LL;
    int64_t slide_us = config_.slide.count() * 1000LL;
    if (slide_us <= 0) {
        slide_us = size_us;
    }

    // First window that could contain event_time
    int64_t first_start_us = (ev_us / slide_us) * slide_us;
    // Also check windows that started before and still cover event_time
    // i.e. start >= ev_us - size_us + slide_us, rounded to slide alignment
    int64_t earliest_start = ((ev_us - size_us) / slide_us) * slide_us;
    if (earliest_start < 0)
        earliest_start = 0;

    for (int64_t start_us = earliest_start; start_us <= first_start_us; start_us += slide_us) {
        int64_t end_us = start_us + size_us;
        // Only create if event_time falls in [start, end)
        if (ev_us < start_us || ev_us >= end_us) {
            continue;
        }
        // O(1) duplicate check via hash set
        bool found = (window_start_set_.count(start_us) > 0);
        if (!found) {
            InternalWindow win;
            win.window_id     = genId();
            win.start         = fromMicros(start_us);
            win.end           = fromMicros(end_us);
            win.partition_key = partition_key;
            window_start_set_.insert(start_us);
            windows_.push_back(std::move(win));
            ++windows_opened_;
        }
    }
}

std::vector<WindowResult> SlidingWindow::closeExpiredWindows(int64_t watermark_us) {
    // Called with mutex_ held. Returns results to emit; callers fire callbacks
    // outside the lock to prevent re-entrant deadlock.
    std::vector<WindowResult> pending;
    for (auto &w : windows_) {
        if (!w.closed && toMicros(w.end) <= watermark_us) {
            w.closed = true;
            ++windows_closed_;
            pending.push_back(computeResult(w, false));
            ++results_emitted_;
        }
    }
    // Prune closed windows and evict from the start set
    while (!windows_.empty() && windows_.front().closed) {
        window_start_set_.erase(toMicros(windows_.front().start));
        windows_.pop_front();
    }
    return pending;
}

WindowResult SlidingWindow::computeResult(const InternalWindow &win, bool late) const {
    WindowResult r;
    r.window_id      = win.window_id;
    r.window_start   = win.start;
    r.window_end     = win.end;
    r.partition_key  = win.partition_key;
    r.record_count   = win.records.size();
    r.aggregations   = computeAggregations(win.records, agg_specs_);
    r.is_late_firing = late;
    return r;
}

bool SlidingWindow::ingest(const StreamRecord &record) {
    ++records_ingested_;
    updateWatermark(record.event_time);
    int64_t wm    = watermark_us_.load(std::memory_order_acquire);
    int64_t ev_us = toMicros(record.event_time);

    if (ev_us < wm && !config_.watermark.allow_late_data) {
        ++late_records_;
        ++records_dropped_;
        return false;
    }

    // Update last_event_us_ monotonically for idle-timeout tracking
    int64_t old_last = last_event_us_.load(std::memory_order_relaxed);
    while (ev_us > old_last
           && !last_event_us_.compare_exchange_weak(old_last, ev_us, std::memory_order_release,
                                                    std::memory_order_relaxed)) {
    }

    std::vector<WindowResult> pending;
    ResultCallback cb;
    {
        std::lock_guard lk(mutex_);

        ensureWindowsExist(record.event_time, record.partition_key);

        // Add record to all overlapping open windows
        for (auto &w : windows_) {
            if (!w.closed && record.event_time >= w.start && record.event_time < w.end) {
                w.records.push_back(record);
            }
        }

        if (ev_us < wm && config_.watermark.allow_late_data) {
            ++late_records_;
        }

        pending = closeExpiredWindows(wm);
        cb      = callback_;
    } // mutex_ released

    // BUG 3 FIX: fire callbacks outside the lock.
    if (cb) {
        for (auto& r : pending) {
            try { cb(r); } catch (...) {}
        }
    }
    return true;
}

void SlidingWindow::flush() {
    std::vector<WindowResult> pending;
    ResultCallback cb;
    {
        std::lock_guard lk(mutex_);
        pending = closeExpiredWindows(std::numeric_limits<int64_t>::max());
        cb      = callback_;
    }
    if (cb) {
        for (auto& r : pending) {
            try { cb(r); } catch (...) {}
        }
    }
}

WindowStats SlidingWindow::getStats() const {
    return WindowStats{windows_opened_.load(),  windows_closed_.load(), records_ingested_.load(),
                       records_dropped_.load(), late_records_.load(),   results_emitted_.load()};
}

void SlidingWindow::idleTimeoutLoop() {
    while (idle_running_) {
        {
            std::unique_lock lk(idle_mutex_);
            idle_cv_.wait_for(lk, config_.watermark.idle_timeout, [this] { return !idle_running_.load(); });
        }
        if (!idle_running_) {
            break;
        }

        auto now_us        = toMicros(std::chrono::system_clock::now());
        int64_t last       = last_event_us_.load(std::memory_order_acquire);
        int64_t timeout_us = config_.watermark.idle_timeout.count() * 1000LL;
        if (last > 0 && (now_us - last) < timeout_us) {
            continue;
        }

        int64_t tol_us = config_.watermark.max_out_of_orderness.count() * 1000LL;
        int64_t new_wm = now_us - tol_us;
        int64_t old_wm = watermark_us_.load(std::memory_order_relaxed);
        while (new_wm > old_wm
               && !watermark_us_.compare_exchange_weak(old_wm, new_wm, std::memory_order_release,
                                                       std::memory_order_relaxed)) {
        }
        int64_t wm = watermark_us_.load(std::memory_order_acquire);

        std::vector<WindowResult> pending;
        ResultCallback cb;
        {
            std::lock_guard lk(mutex_);
            pending = closeExpiredWindows(wm);
            cb      = callback_;
        }
        if (cb) {
            for (auto& r : pending) {
                try { cb(r); } catch (...) {}
            }
        }
    }
}

// ============================================================================
// SessionWindow
// ============================================================================

SessionWindow::SessionWindow(const SessionWindowConfig &config)
    : config_(config), callback_(), agg_specs_(), sessions_(), running_(false), watermark_us_(0), windows_opened_(0),
      windows_closed_(0), records_ingested_(0), records_dropped_(0), late_records_(0), results_emitted_(0) {
    agg_specs_.reserve(16);
    running_       = true;
    expiry_thread_ = std::thread([this] { expiryLoop(); });
}

SessionWindow::~SessionWindow() {
    running_ = false;
    expiry_cv_.notify_all();
    if (expiry_thread_.joinable()) {
        expiry_thread_.join();
    }
    flush();
    agg_specs_.clear();
    callback_ = {};
}

std::unique_ptr<SessionWindow> createSessionWindow(const SessionWindowConfig &config) {
    return std::make_unique<SessionWindow>(config);
}

void SessionWindow::addAggregation(const WindowAggregateSpec &spec) {
    std::lock_guard lk(mutex_);
    agg_specs_.push_back(spec);
}

void SessionWindow::setResultCallback(ResultCallback cb) {
    std::lock_guard lk(mutex_);
    callback_ = std::move(cb);
}

std::string SessionWindow::generateId() {
    return genId();
}

WindowResult SessionWindow::computeResult(const Session &s, bool late) const {
    WindowResult r;
    r.window_id      = s.session_id;
    r.window_start   = s.start;
    r.window_end     = s.last_event;
    r.partition_key  = s.partition_key;
    r.record_count   = s.records.size();
    r.aggregations   = computeAggregations(s.records, agg_specs_);
    r.is_late_firing = late;
    return r;
}

bool SessionWindow::ingest(const StreamRecord &record) {
    ++records_ingested_;

    // BUG 4 FIX: Apply watermark check (was entirely missing).
    // Use processing-time as a proxy watermark because session windows are
    // gap-based rather than slot-based, but still respect out-of-orderness tolerance.
    int64_t ev_us  = toMicros(record.event_time);
    int64_t tol_us = config_.watermark.max_out_of_orderness.count() * 1000LL;
    int64_t new_wm = ev_us - tol_us;
    int64_t old_wm = watermark_us_.load(std::memory_order_relaxed);
    while (
        new_wm > old_wm
        && !watermark_us_.compare_exchange_weak(old_wm, new_wm, std::memory_order_release, std::memory_order_relaxed)) {
    }
    int64_t wm = watermark_us_.load(std::memory_order_acquire);

    if (ev_us < wm && !config_.watermark.allow_late_data) {
        ++late_records_;
        ++records_dropped_;
        return false;
    }

    WindowResult pending_result;
    bool has_pending = false;
    ResultCallback cb;

    {
        std::lock_guard lk(mutex_);

        if (ev_us < wm) {
            ++late_records_;
        }

        const std::string &key = record.partition_key;
        auto it                = sessions_.find(key);
        if (it == sessions_.end()) {
            Session s;
            s.session_id    = genId();
            s.partition_key = key;
            s.start         = record.event_time;
            s.last_event    = record.event_time;
            s.records.push_back(record);
            if (ev_us < wm && config_.watermark.allow_late_data) {
                s.has_late_records = true;
            }
            sessions_[key] = std::move(s);
            ++windows_opened_;
        } else {
            auto &s = it->second;
            auto gap_since_last
                = std::chrono::duration_cast<std::chrono::milliseconds>(record.event_time - s.last_event);

            if (gap_since_last > config_.gap) {
                // Gap exceeded → close current session and start new
                pending_result = computeResult(s, s.has_late_records);
                has_pending    = true;
                ++windows_closed_;
                ++results_emitted_;
                // New session
                Session ns;
                ns.session_id    = genId();
                ns.partition_key = key;
                ns.start         = record.event_time;
                ns.last_event    = record.event_time;
                ns.records.push_back(record);
                if (ev_us < wm && config_.watermark.allow_late_data) {
                    ns.has_late_records = true;
                }
                it->second = std::move(ns);
                ++windows_opened_;
            } else {
                // Extend existing session.
                // BUG 2 FIX: use max() so that an out-of-order record cannot
                // regress last_event and cause instant timer-driven expiry.
                s.last_event = std::max(s.last_event, record.event_time);
                s.records.push_back(record);
                if (ev_us < wm && config_.watermark.allow_late_data) {
                    s.has_late_records = true;
                }
            }
        }
        cb = callback_;
    } // mutex_ released before callback

    // BUG 3 FIX: invoke callback outside the mutex to prevent re-entrant deadlock.
    if (has_pending && cb) {
        try { cb(pending_result); } catch (...) {}
    }
    return true;
}

void SessionWindow::flush() {
    std::vector<WindowResult> pending;
    ResultCallback cb;
    {
        std::lock_guard lk(mutex_);
        for (auto &[key, s] : sessions_) {
            if (!s.records.empty()) {
                pending.push_back(computeResult(s, s.has_late_records));
                ++windows_closed_;
                ++results_emitted_;
            }
        }
        sessions_.clear();
        cb = callback_;
    }
    // BUG 3 FIX: invoke callbacks outside the mutex.
    if (cb) {
        for (auto& r : pending) {
            try { cb(r); } catch (...) {}
        }
    }
}

WindowStats SessionWindow::getStats() const {
    return WindowStats{windows_opened_.load(),  windows_closed_.load(), records_ingested_.load(),
                       records_dropped_.load(), late_records_.load(),   results_emitted_.load()};
}

void SessionWindow::expiryLoop() {
    while (running_) {
        {
            std::unique_lock lk(expiry_mutex_);
            expiry_cv_.wait_for(lk, config_.session_expiry_check_interval_ms, [this] { return !running_.load(); });
        }
        if (!running_) {
            break;
        }

        std::vector<WindowResult> pending;
        ResultCallback cb;
        {
            auto now = std::chrono::system_clock::now();
            std::lock_guard lk(mutex_);
            std::vector<std::string> to_close;
            for (const auto &[key, s] : sessions_) {
                auto idle = std::chrono::duration_cast<std::chrono::milliseconds>(now - s.last_event);
                if (idle > config_.gap) {
                    to_close.push_back(key);
                }
            }
            for (const auto &key : to_close) {
                auto &s = sessions_[key];
                if (!s.records.empty()) {
                    pending.push_back(computeResult(s, s.has_late_records));
                    ++windows_closed_;
                    ++results_emitted_;
                }
                sessions_.erase(key);
            }
            cb = callback_;
        } // mutex_ released before callback (BUG 3 FIX)
        if (cb) {
            for (auto& r : pending) {
                try { cb(r); } catch (...) {}
            }
        }
    }
}

// ============================================================================
// HoppingWindow
// ============================================================================

HoppingWindow::HoppingWindow(const HoppingWindowConfig &config)
    : config_(config), callback_(), agg_specs_(), windows_(), watermark_us_(0), windows_opened_(0), windows_closed_(0),
      records_ingested_(0), records_dropped_(0), late_records_(0), results_emitted_(0) {
    agg_specs_.reserve(16);
}

HoppingWindow::~HoppingWindow() {
    flush();
    agg_specs_.clear();
    callback_ = {};
}

std::unique_ptr<HoppingWindow> createHoppingWindow(const HoppingWindowConfig &config) {
    return std::make_unique<HoppingWindow>(config);
}

void HoppingWindow::addAggregation(const WindowAggregateSpec &spec) {
    std::lock_guard lk(mutex_);
    agg_specs_.push_back(spec);
}

void HoppingWindow::setResultCallback(ResultCallback cb) {
    std::lock_guard lk(mutex_);
    callback_ = std::move(cb);
}

std::string HoppingWindow::generateId() {
    return genId();
}

void HoppingWindow::updateWatermark(const std::chrono::system_clock::time_point &event_time) {
    int64_t ev_us  = toMicros(event_time);
    int64_t tol_us = config_.watermark.max_out_of_orderness.count() * 1000LL;
    int64_t new_wm = ev_us - tol_us;
    int64_t old_wm = watermark_us_.load(std::memory_order_relaxed);
    while (
        new_wm > old_wm
        && !watermark_us_.compare_exchange_weak(old_wm, new_wm, std::memory_order_release, std::memory_order_relaxed)) {
    }
}

void HoppingWindow::ensureWindowsExist(const std::chrono::system_clock::time_point &event_time) {
    // Same as SlidingWindow but uses hop instead of slide
    int64_t ev_us   = toMicros(event_time);
    int64_t size_us = config_.size.count() * 1000LL;
    int64_t hop_us  = config_.hop.count() * 1000LL;
    if (hop_us <= 0) {
        hop_us = size_us;
    }

    int64_t first_start = (ev_us / hop_us) * hop_us;
    int64_t earliest    = ((ev_us - size_us) / hop_us) * hop_us;
    if (earliest < 0)
        earliest = 0;

    for (int64_t start_us = earliest; start_us <= first_start; start_us += hop_us) {
        int64_t end_us = start_us + size_us;
        if (ev_us < start_us || ev_us >= end_us) {
            continue;
        }
        // O(1) duplicate check via hash set
        bool found = (window_start_set_.count(start_us) > 0);
        if (!found) {
            InternalWindow win;
            win.window_id = genId();
            win.start     = fromMicros(start_us);
            win.end       = fromMicros(end_us);
            window_start_set_.insert(start_us);
            windows_.push_back(std::move(win));
            ++windows_opened_;
        }
    }
}

std::vector<WindowResult> HoppingWindow::closeExpiredWindows(int64_t watermark_us) {
    // Called with mutex_ held. Returns results to emit; callers fire callbacks
    // outside the lock to prevent re-entrant deadlock.
    std::vector<WindowResult> pending;
    for (auto &w : windows_) {
        if (!w.closed && toMicros(w.end) <= watermark_us) {
            w.closed = true;
            ++windows_closed_;
            pending.push_back(computeResult(w, false));
            ++results_emitted_;
        }
    }
    // Prune closed windows and evict from the start set
    while (!windows_.empty() && windows_.front().closed) {
        window_start_set_.erase(toMicros(windows_.front().start));
        windows_.pop_front();
    }
    return pending;
}

WindowResult HoppingWindow::computeResult(const InternalWindow &win, bool late) const {
    WindowResult r;
    r.window_id      = win.window_id;
    r.window_start   = win.start;
    r.window_end     = win.end;
    r.record_count   = win.records.size();
    r.aggregations   = computeAggregations(win.records, agg_specs_);
    r.is_late_firing = late;
    return r;
}

bool HoppingWindow::ingest(const StreamRecord &record) {
    ++records_ingested_;
    updateWatermark(record.event_time);
    int64_t wm    = watermark_us_.load(std::memory_order_acquire);
    int64_t ev_us = toMicros(record.event_time);

    if (ev_us < wm && !config_.watermark.allow_late_data) {
        ++late_records_;
        ++records_dropped_;
        return false;
    }

    std::vector<WindowResult> pending;
    ResultCallback cb;
    {
        std::lock_guard lk(mutex_);
        ensureWindowsExist(record.event_time);

        for (auto &w : windows_) {
            if (!w.closed && record.event_time >= w.start && record.event_time < w.end) {
                w.records.push_back(record);
            }
        }

        if (ev_us < wm && config_.watermark.allow_late_data) {
            ++late_records_;
        }
        pending = closeExpiredWindows(wm);
        cb      = callback_;
    } // mutex_ released

    // BUG 3 FIX: fire callbacks outside the lock.
    if (cb) {
        for (auto& r : pending) {
            try { cb(r); } catch (...) {}
        }
    }
    return true;
}

void HoppingWindow::flush() {
    std::vector<WindowResult> pending;
    ResultCallback cb;
    {
        std::lock_guard lk(mutex_);
        pending = closeExpiredWindows(std::numeric_limits<int64_t>::max());
        cb      = callback_;
    }
    if (cb) {
        for (auto& r : pending) {
            try { cb(r); } catch (...) {}
        }
    }
}

WindowStats HoppingWindow::getStats() const {
    return WindowStats{windows_opened_.load(),  windows_closed_.load(), records_ingested_.load(),
                       records_dropped_.load(), late_records_.load(),   results_emitted_.load()};
}

// ============================================================================
// StreamingWindowPipeline
// ============================================================================

StreamingWindowPipeline StreamingWindowPipeline::tumbling(std::chrono::milliseconds size, WatermarkConfig wm) {
    StreamingWindowPipeline p;
    p.agg_specs_.reserve(16);
    p.config_.type      = Type::TUMBLING;
    p.config_.size      = size;
    p.config_.watermark = wm;
    return p;
}

StreamingWindowPipeline StreamingWindowPipeline::sliding(std::chrono::milliseconds size,
                                                         std::chrono::milliseconds slide, WatermarkConfig wm) {
    StreamingWindowPipeline p;
    p.agg_specs_.reserve(16);
    p.config_.type      = Type::SLIDING;
    p.config_.size      = size;
    p.config_.slide     = slide;
    p.config_.watermark = wm;
    return p;
}

StreamingWindowPipeline StreamingWindowPipeline::session(std::chrono::milliseconds gap, WatermarkConfig wm,
                                                         std::chrono::milliseconds expiry_interval_ms) {
    StreamingWindowPipeline p;
    p.agg_specs_.reserve(16);
    p.config_.type                       = Type::SESSION;
    p.config_.gap                        = gap;
    p.config_.watermark                  = wm;
    p.config_.session_expiry_interval_ms = expiry_interval_ms;
    return p;
}

StreamingWindowPipeline StreamingWindowPipeline::hopping(std::chrono::milliseconds size, std::chrono::milliseconds hop,
                                                         WatermarkConfig wm) {
    StreamingWindowPipeline p;
    p.agg_specs_.reserve(16);
    p.config_.type      = Type::HOPPING;
    p.config_.size      = size;
    p.config_.hop       = hop;
    p.config_.watermark = wm;
    return p;
}

StreamingWindowPipeline &StreamingWindowPipeline::aggregate(const WindowAggregateSpec &spec) {
    agg_specs_.emplace_back(spec.name, spec.func, spec.field, spec.percentile_p);
    return *this;
}

StreamingWindowPipeline &StreamingWindowPipeline::onResult(std::function<void(WindowResult)> callback) {
    callback_ = std::move(callback);
    return *this;
}

std::shared_ptr<StreamingWindowPipeline> StreamingWindowPipeline::build() {
    // NOTE ON DATA RACE FINDINGS IN BUILD() METHOD:
    // This method constructs a new shared_ptr<StreamingWindowPipeline> in a single-threaded
    // context and returns it to the caller. All member variable assignments (config_, agg_specs_,
    // callback_, tumbling_, sliding_, session_, hopping_) occur on a newly-created object
    // before it is returned from the function. The object is not shared with any other thread
    // until build() returns and the caller receives the shared_ptr. Therefore, all concurrent
    // access warnings on lines 1224, 1225, 1233, 1247, 1261, 1275 are FALSE_POSITIVES:
    // they flag safe initialization of a locally-constructed object as if it were already
    // shared and concurrently accessed.
    
    auto pipeline        = std::make_shared<StreamingWindowPipeline>();
    pipeline->config_    = config_;
    pipeline->agg_specs_ = agg_specs_;
    pipeline->callback_  = callback_;

    switch (config_.type) {
        case Type::TUMBLING: {
            TumblingWindowConfig cfg;
            cfg.size            = config_.size;
            cfg.watermark       = config_.watermark;
            pipeline->tumbling_ = std::make_shared<TumblingWindow>(cfg);
            for (const auto &s : agg_specs_) {
                pipeline->tumbling_->addAggregation(s);
            }
            if (callback_) {
                pipeline->tumbling_->setResultCallback(callback_);
            }
            break;
        }
        case Type::SLIDING: {
            SlidingWindowConfig cfg;
            cfg.size           = config_.size;
            cfg.slide          = config_.slide;
            cfg.watermark      = config_.watermark;
            pipeline->sliding_ = std::make_shared<SlidingWindow>(cfg);
            for (const auto &s : agg_specs_) {
                pipeline->sliding_->addAggregation(s);
            }
            if (callback_) {
                pipeline->sliding_->setResultCallback(callback_);
            }
            break;
        }
        case Type::SESSION: {
            SessionWindowConfig cfg;
            cfg.gap                              = config_.gap;
            cfg.watermark                        = config_.watermark;
            cfg.session_expiry_check_interval_ms = config_.session_expiry_interval_ms;
            pipeline->session_                   = std::make_shared<SessionWindow>(cfg);
            for (const auto &s : agg_specs_) {
                pipeline->session_->addAggregation(s);
            }
            if (callback_) {
                pipeline->session_->setResultCallback(callback_);
            }
            break;
        }
        case Type::HOPPING: {
            HoppingWindowConfig cfg;
            cfg.size           = config_.size;
            cfg.hop            = config_.hop;
            cfg.watermark      = config_.watermark;
            pipeline->hopping_ = std::make_shared<HoppingWindow>(cfg);
            for (const auto &s : agg_specs_) {
                pipeline->hopping_->addAggregation(s);
            }
            if (callback_) {
                pipeline->hopping_->setResultCallback(callback_);
            }
            break;
        }
    }
    pipeline->built_ = true;
    return pipeline;
}

bool StreamingWindowPipeline::ingest(const StreamRecord &record) {
    if (!built_) {
        spdlog::warn("StreamingWindowPipeline::ingest() called before build()");
        return false;
    }
    if (tumbling_) {
        return tumbling_->ingest(record);
    }
    if (sliding_) {
        return sliding_->ingest(record);
    }
    if (session_) {
        return session_->ingest(record);
    }
    if (hopping_) {
        return hopping_->ingest(record);
    }
    return false;
}

void StreamingWindowPipeline::flush() {
    if (!built_) {
        spdlog::warn("StreamingWindowPipeline::flush() called before build()");
        return;
    }
    if (tumbling_) {
        tumbling_->flush();
    }
    if (sliding_) {
        sliding_->flush();
    }
    if (session_) {
        session_->flush();
    }
    if (hopping_) {
        hopping_->flush();
    }
}

WindowStats StreamingWindowPipeline::getStats() const {
    if (!built_) {
        return {};
    }
    if (tumbling_) {
        return tumbling_->getStats();
    }
    if (sliding_) {
        return sliding_->getStats();
    }
    if (session_) {
        return session_->getStats();
    }
    if (hopping_) {
        return hopping_->getStats();
    }
    return {};
}

} // namespace analytics
} // namespace themisdb