postgres_event_source.hpp

/* -*- mode: c++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*-
 *
 * Copyright (C) 2025 Marco Craveiro <marco.craveiro@gmail.com>
 *
 * This program is free software; you can redistribute it and/or modify it under
 * the terms of the GNU General Public License as published by the Free Software
 * Foundation; either version 3 of the License, or (at your option) any later
 * version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 *
 */
#ifndef ORES_EVENTING_SERVICE_POSTGRES_EVENT_SOURCE_HPP
#define ORES_EVENTING_SERVICE_POSTGRES_EVENT_SOURCE_HPP
 
#include <memory>
#include <functional>
#include <unordered_map>
#include "ores.logging/make_logger.hpp"
#include "ores.database/domain/context.hpp"
#include "ores.database/service/postgres_listener_service.hpp"
#include "ores.eventing/domain/entity_change_event.hpp"
#include "ores.eventing/service/event_bus.hpp"
 
namespace ores::eventing::service {
 
class postgres_event_source final {
private:
    [[nodiscard]] static auto& lg() {
        using namespace ores::logging;
        static auto instance = make_logger(
            "ores.eventing.service.postgres_event_source");
        return instance;
    }
 
    using publisher_fn = std::function<void(
        std::chrono::system_clock::time_point,
        const std::vector<std::string>&,
        const std::string&)>;
 
    struct entity_mapping {
        std::string channel_name;
        publisher_fn publisher;
    };
 
public:
    postgres_event_source(database::context ctx, event_bus& bus);
 
    ~postgres_event_source();
 
    postgres_event_source(const postgres_event_source&) = delete;
    postgres_event_source& operator=(const postgres_event_source&) = delete;
 
    template<typename Event>
    void register_mapping(const std::string& entity_name,
                          const std::string& channel_name) {
        using namespace ores::logging;
        BOOST_LOG_SEV(lg(), info)
            << "Registering entity-to-event mapping: entity='" << entity_name
            << "', channel='" << channel_name << "'";
 
        entity_mappings_[entity_name] = entity_mapping{
            .channel_name = channel_name,
            .publisher = [this, entity_name](std::chrono::system_clock::time_point ts,
                                              const std::vector<std::string>& entity_ids,
                                              const std::string& tenant_id) {
                using namespace ores::logging;
                BOOST_LOG_SEV(lg(), info)
                    << "Publishing domain event for entity: " << entity_name
                    << " with " << entity_ids.size() << " entity IDs"
                    << " (tenant: " << tenant_id << ")";
                bus_.publish(Event{ts, entity_ids, tenant_id});
            }
        };
 
        listener_.subscribe(channel_name);
        BOOST_LOG_SEV(lg(), debug)
            << "Subscribed to PostgreSQL channel: " << channel_name;
    }
 
    void start();
 
    void stop();
 
private:
    void on_entity_change(const domain::entity_change_event& e);
 
    event_bus& bus_;
    ores::database::service::postgres_listener_service listener_;
    std::unordered_map<std::string, entity_mapping> entity_mappings_;
};
 
}
 
#endif