ores.cpp.eventing-integration-test.nats_integration_test

{{! GENERATED FILE — tangled from projects/ores.codegen/library/templates/ores.cpp.eventing-integration-test.nats_integration_test.org. Edit the org source. }}
{{{cpp_license}}}
{{#domain_entity}}
#include "ores.database/domain/context.hpp"
#include "ores.eventing.api/domain/entity_change_event.hpp"
#include "ores.eventing.api/domain/event_traits.hpp"
#include "ores.eventing.api/service/event_bus.hpp"
#include "ores.eventing.core/service/entity_event_publisher.hpp"
#include "ores.eventing.core/service/postgres_event_source.hpp"
#include "ores.logging/make_logger.hpp"
#include "ores.nats/domain/wire_codec.hpp"
#include "ores.nats/service/client.hpp"
#include "ores.{{component_include}}/domain/{{entity_singular}}.hpp"
#include "ores.{{component_include}}/domain/{{entity_singular}}_json_io.hpp" // IWYU pragma: keep.
#include "ores.{{component_include}}/eventing/{{entity_singular}}_changed_event.hpp"
#include "ores.{{component_include}}/{{generator_facet_name}}/{{entity_singular}}_generator.hpp"
#include "ores.{{component_core}}/repository/{{entity_singular}}_repository.hpp"
{{#sql.party_id_from_session}}
#include "ores.refdata.api/generators/party_generator.hpp"
#include "ores.refdata.core/repository/party_repository.hpp"
{{/sql.party_id_from_session}}
{{#seed_party}}
// Party seeds (mandatory party_id soft FKs, direct or via a parent's own
// mandatory party_id FK): the party generator and repository are used
// regardless of the child's generator facet, hence the fully-qualified
// refdata paths.
#include "ores.refdata.api/generators/party_generator.hpp"
#include "ores.refdata.core/repository/party_repository.hpp"
{{/seed_party}}
{{#seed_country_sentinel}}
// Country sentinel seed ({{entity_title}}'s insert trigger validates
// country_code against the countries table, and the synthetic generator
// always emits the ZZ sentinel): like the party seeds, the country
// generator and repository are used regardless of the child's generator
// facet, hence the fully-qualified refdata paths.
#include "ores.refdata.api/generators/country_generator.hpp"
#include "ores.refdata.core/repository/country_repository.hpp"
{{/seed_country_sentinel}}
{{#seed_parent_country_sentinel}}
// FK-parent sentinel seed: a seeded calendar parent's insert trigger
// validates country_code against the countries table for the write
// tenant, and the synthetic calendar generator always emits the ZZ
// sentinel -- seed it before the parent write or the parent insert is
// rejected. Like the entity-level sentinel seed, the country generator
// and repository are used regardless of the child's generator facet,
// hence the fully-qualified refdata paths.
#include "ores.refdata.api/generators/country_generator.hpp"
#include "ores.refdata.core/repository/country_repository.hpp"
{{/seed_parent_country_sentinel}}
{{#seed_currency}}
// Aggregation-currency seed ({{entity_title}}'s insert trigger validates
// aggregation_ccy against the active currencies for the write tenant,
// and the synthetic currency generator's first code -- X-0 -- is the
// code the {{entity_singular}} generator hardcodes): like the party
// seeds, the currency generator and repository are used regardless of
// the child's generator facet, hence the fully-qualified refdata paths.
#include "ores.refdata.api/generators/currency_generator.hpp"
#include "ores.refdata.core/repository/currency_repository.hpp"
{{/seed_currency}}
{{#foreign_keys}}
{{^nullable}}
{{#parent_entity_singular}}
{{^parent_is_party}}
// Soft-FK parent seeding ({{table}}): the parent's own generator and
// repository live in the same component as the child.
#include "ores.{{component_include}}/{{parent_generator_facet_name}}/{{parent_entity_singular}}_generator.hpp"
#include "ores.{{component_core}}/repository/{{parent_entity_singular}}_repository.hpp"
{{#parent_required_fks}}
// Grand-parent seeding ({{table}}): the parent's own mandatory soft FKs
// reference rows the test seeds before the parent, so their generator
// and repository headers are needed too.
#include "ores.{{component_include}}/{{parent_generator_facet_name}}/{{parent_entity_singular}}_generator.hpp"
#include "ores.{{component_core}}/repository/{{parent_entity_singular}}_repository.hpp"
{{/parent_required_fks}}
{{/parent_is_party}}
{{/parent_entity_singular}}
{{/nullable}}
{{/foreign_keys}}
#include "ores.testing/make_generation_context.hpp"
#include "ores.testing/scoped_database_helper.hpp"
#include "ores.utility/rfl/reflectors.hpp" // IWYU pragma: keep.
#include <boost/uuid/uuid_io.hpp>
#include <catch2/catch_test_macros.hpp>
#include <cstdlib>
#include <thread>

// Proves the "write an entity, observe its NATS entity-changed
// notification" pattern end to end for {{entity_singular}} -- the
// production DB-write -> pg_notify -> postgres_event_source ->
// event_bus -> NATS publish chain, assembled directly here the same
// way the production event-registrar wires it.

namespace {

const std::string_view test_suite("{{component}}.tests");
const std::string tags("[eventing][integration]");

{{#sql.party_id_from_session}}
// {{entity_title}} writes are party-scoped: the session-level
// app.current_party_id GUC must be set before writing.
ores::database::context
write_test_party_and_scope_context(ores::testing::scoped_database_helper& h,
                                   ores::utility::generation::generation_context& ctx) {
    using ores::refdata::repository::party_repository;
    party_repository party_repo;
    auto party = ores::refdata::generators::generate_synthetic_party(ctx);
    party.change_reason_code = "system.test";
    auto existing = party_repo.read_latest(h.context());
    for (const auto& e : existing) {
        if (e.tenant_id == party.tenant_id) {
            party.parent_party_id = e.id;
            break;
        }
    }
    party_repo.write(h.context(), party);
    return h.context().with_party(h.tenant_id(), party.id, {party.id}, h.db_user());
}
{{/sql.party_id_from_session}}

// Reads NATS connection settings the same way every service resolves
// them at startup -- CMake bakes every .env variable into the ctest
// process environment, so these are populated identically for both
// `compass build` local runs and CI.
ores::nats::config::nats_options test_nats_options() {
    auto env = [](const char* name) -> std::string {
        const char* v = std::getenv(name);
        return v ? std::string(v) : std::string();
    };

    ores::nats::config::nats_options opts;
    opts.url = env("ORES_NATS_URL");
    if (opts.url.empty())
        opts.url = "nats://localhost:4222";
    opts.subject_prefix = env("ORES_NATS_SUBJECT_PREFIX");
    opts.tls_ca_cert = env("ORES_NATS_TLS_CA");
    opts.tls_client_cert = env("ORES_NATS_TLS_CERT");
    opts.tls_client_key = env("ORES_NATS_TLS_KEY");
    return opts;
}

}

using namespace ores::{{component}}::{{generator_facet_name}};
using ores::{{component}}::domain::{{entity_singular}};
using ores::{{component}}::repository::{{entity_singular}}_repository;
{{#seed_country_sentinel}}
using ores::refdata::repository::country_repository;
{{/seed_country_sentinel}}
{{#seed_parent_country_sentinel}}
using ores::refdata::repository::country_repository;
{{/seed_parent_country_sentinel}}
{{#seed_currency}}
using ores::refdata::repository::currency_repository;
{{/seed_currency}}
using ores::testing::scoped_database_helper;
using namespace ores::logging;

TEST_CASE("write_{{entity_singular}}_publishes_nats_changed_event", tags) {
    auto lg(make_logger(test_suite));

    scoped_database_helper h;
    auto ctx = ores::testing::make_generation_context(h);
{{#sql.party_id_from_session}}
    auto party_ctx = write_test_party_and_scope_context(h, ctx);
{{/sql.party_id_from_session}}
{{^sql.party_id_from_session}}
    auto& party_ctx = h.context();
{{/sql.party_id_from_session}}

    // 1. Wire the same DB-notify -> event_bus -> NATS-publish chain the
    // production event-registrar wires in the live service, assembled
    // directly in the test instead of via a running process.
    namespace ev = ores::eventing;
    ev::service::event_bus bus;
    ev::service::postgres_event_source event_source(party_ctx, bus);

    ores::nats::service::client nats(test_nats_options());
    nats.connect();
    REQUIRE(nats.is_connected());

    auto sub = bus.subscribe<ores::{{component}}::eventing::{{entity_singular}}_changed_event>(
        [&nats](const ores::{{component}}::eventing::{{entity_singular}}_changed_event& e) {
            ev::service::publish_entity_event(
                nats,
                std::string(ev::domain::event_traits<
                            ores::{{component}}::eventing::{{entity_singular}}_changed_event>::name),
                ev::domain::entity_change_event{.entity = "ores.{{component}}.{{entity_singular}}",
                                                .timestamp = e.timestamp,
{{#primary_key.is_text}}
                                                .entity_ids = e.{{primary_key.column}}s,
{{/primary_key.is_text}}
{{#primary_key.is_uuid}}
                                                .entity_ids = e.{{entity_singular_short}}_ids,
{{/primary_key.is_uuid}}
                                                .tenant_id = e.tenant_id});
        });

    event_source.register_mapping<ores::{{component}}::eventing::{{entity_singular}}_changed_event>(
        "ores.{{component}}.{{entity_singular}}", "{{sql_name_base}}");

    // 2. Subscribe as an external observer would, on the relative subject --
    // client::subscribe() prepends the subject_prefix itself.
    auto observer = nats.subscribe_buffered(
        std::string(ev::domain::event_traits<
                    ores::{{component}}::eventing::{{entity_singular}}_changed_event>::name),
        10);

    // The listener thread issues LISTEN asynchronously on its own
    // dedicated connection. Block until it has actually done so before
    // writing -- Postgres does not queue NOTIFYs sent before a matching
    // LISTEN is registered.
    event_source.start();
    REQUIRE(event_source.wait_until_ready());

    // 3. Write -- triggers the entity's notify trigger -> pg_notify ->
    // the chain wired above -> NATS.
    auto v = generate_synthetic_{{entity_singular}}(ctx);
{{#has_audit_group}}
    v.audit.change_reason_code = "system.test";
{{/has_audit_group}}
{{^has_audit_group}}
    v.change_reason_code = "system.test";
{{/has_audit_group}}
{{#sql.party_id_from_session}}
{{#has_identity_group}}
    v.identity.party_id = *party_ctx.party_id();
{{/has_identity_group}}
{{^has_identity_group}}
    v.party_id = *party_ctx.party_id();
{{/has_identity_group}}
{{/sql.party_id_from_session}}
{{#foreign_keys}}
{{^nullable}}
{{#parent_entity_singular}}
    // Seed the active {{parent_entity_singular}} row {{table}} references:
    // the insert trigger's existence check rejects a synthetic key that
    // matches no active row, so the parent must be written first.
    auto {{column}}_parent = ores::{{component}}::{{parent_generator_facet_name}}::generate_synthetic_{{parent_entity_singular}}(ctx);
{{#parent_has_audit_group}}
    {{column}}_parent.audit.change_reason_code = "system.test";
{{/parent_has_audit_group}}
{{^parent_has_audit_group}}
    {{column}}_parent.change_reason_code = "system.test";
{{/parent_has_audit_group}}
{{#parent_requires_party}}
    // {{parent_entity_singular}}'s own mandatory party_id FK (session-set in
    // production) needs an active party too: seed one, attached under the
    // tenant's root party like the direct-party branch below.
    auto {{column}}_party = ores::{{component}}::generators::generate_synthetic_party(ctx);
    {{column}}_party.change_reason_code = "system.test";
    auto {{column}}_party_existing = ores::{{component}}::repository::party_repository().read_latest(party_ctx);
    for (const auto& e : {{column}}_party_existing) {
        if (e.tenant_id == {{column}}_party.tenant_id) {
            {{column}}_party.parent_party_id = e.id;
            break;
        }
    }
    ores::{{component}}::repository::party_repository {{column}}_party_repo;
    {{column}}_party_repo.write(party_ctx, {{column}}_party);
    {{column}}_parent.{{#parent_has_identity_group}}identity.{{/parent_has_identity_group}}party_id = {{column}}_party.id;
{{/parent_requires_party}}
{{#parent_is_party}}
    // Only one root party (parent_party_id null) is allowed per tenant:
    // attach to the existing root party instead of creating a second one.
    auto {{column}}_existing = ores::{{component}}::repository::party_repository().read_latest(party_ctx);
    for (const auto& e : {{column}}_existing) {
        if (e.tenant_id == {{column}}_parent.tenant_id) {
            {{column}}_parent.parent_party_id = e.id;
            break;
        }
    }
{{/parent_is_party}}
{{#parent_required_fks}}
    // Seed the active {{parent_entity_singular}} row {{table}} references:
    // the parent's own existence check rejects a synthetic key that
    // matches no active row, so the grandparent must be written first.
    auto {{var}} = ores::{{component}}::{{parent_generator_facet_name}}::generate_synthetic_{{parent_entity_singular}}(ctx);
{{#parent_has_audit_group}}
    {{var}}.audit.change_reason_code = "system.test";
{{/parent_has_audit_group}}
{{^parent_has_audit_group}}
    {{var}}.change_reason_code = "system.test";
{{/parent_has_audit_group}}
    ores::{{component}}::repository::{{parent_entity_singular}}_repository {{var}}_repo;
    {{var}}_repo.write(party_ctx, {{var}});
    {{parent_var}}.{{column}} = {{var}}.{{target_column}};
{{/parent_required_fks}}
{{#parent_seed_country_sentinel}}
    // The parent calendar's insert trigger validates country_code
    // against the countries table for the write tenant, and the
    // synthetic calendar generator always emits the ZZ sentinel --
    // seed it before the parent write or the parent insert is
    // rejected. Distinct name from the entity-level sentinel seed
    // block: both are in scope when the entity also carries the
    // seed_country_sentinel flag.
    country_repository parent_cty_repo;
    parent_cty_repo.write(party_ctx, {generate_country_sentinel(ctx)});
{{/parent_seed_country_sentinel}}
    ores::{{component}}::repository::{{parent_entity_singular}}_repository {{column}}_repo;
    {{column}}_repo.write(party_ctx, {{column}}_parent);
    v.{{#is_identity_group_column}}identity.{{/is_identity_group_column}}{{column}} = {{column}}_parent.{{target_column}};
{{/parent_entity_singular}}
{{/nullable}}
{{/foreign_keys}}
{{#has_identity_group}}
{{#primary_key.is_uuid}}
    const auto id_str = boost::uuids::to_string(v.identity.{{primary_key.column}});
{{/primary_key.is_uuid}}
{{^primary_key.is_uuid}}
    const auto id_str = v.identity.{{primary_key.column}};
{{/primary_key.is_uuid}}
{{/has_identity_group}}
{{^has_identity_group}}
{{#primary_key.is_uuid}}
    const auto id_str = boost::uuids::to_string(v.{{primary_key.column}});
{{/primary_key.is_uuid}}
{{^primary_key.is_uuid}}
    const auto id_str = v.{{primary_key.column}};
{{/primary_key.is_uuid}}
{{/has_identity_group}}
    BOOST_LOG_SEV(lg, debug) << "{{entity_title}}: " << v;
{{#seed_country_sentinel}}
    // {{entity_title}}'s insert trigger validates country_code against
    // the countries table for the write tenant, and the synthetic
    // generator always uses the ZZ sentinel -- seed it first (as the
    // materialisation tests do) or the insert is rejected.
    country_repository cty_repo;
    cty_repo.write(party_ctx, {generate_country_sentinel(ctx)});
{{/seed_country_sentinel}}
{{#seed_currency}}
    // Seed the active aggregation-currency row the insert trigger
    // references: the currency check is strict and tenant-scoped, and
    // this process's synthetic-currency counter has moved past the
    // X-0 code the {{entity_singular}} generator hardcodes -- so stamp
    // it explicitly, mirroring write_synthetic_pairs' pattern.
    auto aggregation_ccy = generate_synthetic_currency(ctx);
    aggregation_ccy.iso_code = "X-0";
    currency_repository ccy_repo;
    ccy_repo.write(party_ctx, {aggregation_ccy});
{{/seed_currency}}

    {{entity_singular}}_repository repo;
    repo.write(party_ctx, v);

    // 4. Poll the observer's buffer for the notification. Generous
    // timeout: trigger -> pg_notify -> 100ms listener poll -> event_bus
    // -> NATS round trip, all real, no mocks.
    std::vector<ores::nats::message> received;
    for (int i = 0; i < 50 && received.empty(); ++i) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        auto snap = observer.snapshot();
        for (const auto& msg : snap) {
            auto decoded =
                ores::nats::default_wire_codec().decode<ev::domain::entity_change_event>(
                    msg.data);
            if (decoded && decoded->entity == "ores.{{component}}.{{entity_singular}}") {
                for (const auto& changed_id : decoded->entity_ids) {
                    if (changed_id == id_str)
                        received.push_back(msg);
                }
            }
        }
    }

    event_source.stop();

    REQUIRE_FALSE(received.empty());
    BOOST_LOG_SEV(lg, info) << "Received " << received.size()
                            << " matching NATS notification(s) for {{entity_singular}} " << id_str;
}
{{/domain_entity}}