feed_projector.hpp

Go to the documentation of this file.

#pragma once
 
#include <cstddef>
#include <cstdint>
#include <string_view>
 
#include "native.h"
#include <asio.hpp>
 
#include "event_projector.hpp"
#include "sqlite_hot_store.hpp"
#include "feed.hpp"
 
namespace dcn::feed
{
    class FeedProjector final : public events::IEventProjector
    {
        public:
            FeedProjector(events::SQLiteHotStore & store,
                          Feed & feed,
                          const asio::strand<asio::io_context::executor_type> & write_strand,
                          events::ProjectorRetryConfig retry = {});
 
            std::string_view id() const override;
            std::int64_t cursor() const override { return _last_change_seq; }
            asio::awaitable<std::size_t> projectBatch(std::size_t limit, std::int64_t now_ms) override;
 
        private:
            // Retries dead-lettered rows (throttled by _retry.sweep_interval_ms)
            // and clears their bit on success. Returns the number of rows resolved.
            asio::awaitable<std::size_t> _sweepDeadLetters(std::size_t limit, std::int64_t now_ms);
 
            events::SQLiteHotStore & _store;
            Feed & _feed;
            asio::strand<asio::io_context::executor_type> _write_strand;
            events::ProjectorRetryConfig _retry;
            std::int64_t _last_change_seq{0};
 
            // Consecutive-failure tracking for the row currently at the cursor head.
            std::int64_t _failing_seq{0};
            std::size_t _failure_count{0};
 
            std::int64_t _last_sweep_ms{0};
    };
}