ores.refdata.party

std::string(faker::company::companyName())

std::string(faker::string::alpha(6))

""

std::nullopt

std::string("Operational")

std::string("Corporate")

std::nullopt

std::string("WRLD")

std::string("Active")

std::nullopt

-- Sequence used to generate a unique base-26 suffix for auto-generated
-- codenames.  nextval() is outside MVCC: it advances even within a
-- multi-row INSERT statement, so every row in a batch gets a different
-- suffix, avoiding duplicate-key collisions that a NOT EXISTS loop cannot
-- prevent due to the statement-level snapshot in READ COMMITTED.
create sequence if not exists ores_refdata_party_codename_seq;

drop sequence if exists ores_refdata_party_codename_seq;

-- Codename is immutable: restore from active row on every update.
select codename into NEW.codename
from "ores_refdata_parties_tbl"
where tenant_id = NEW.tenant_id
  and id = NEW.id
  and valid_to = ores_utility_infinity_timestamp_fn();

-- Auto-generate codename using whimsical name + sequence suffix.
-- The sequence suffix is outside MVCC (nextval always advances),
-- so every row in a multi-row INSERT gets a distinct suffix, making
-- the codename globally unique without relying on a NOT EXISTS
-- check that cannot see sibling rows in the same statement.
if NEW.codename = '' or NEW.codename is null then
    NEW.codename := ores_utility_generate_whimsical_name_fn() || '_' ||
        ores_utility_to_base26_fn(nextval('ores_refdata_party_codename_seq'));
end if;
-- Validate the final codename.
if NEW.codename !~ '^[a-z][a-z_]+$' then
    raise exception 'codename must match ^[a-z][a-z_]+$ got: %', NEW.codename
        using errcode = '23514';
end if;
-- Provision the per-party report event queue.
perform ores_mq_queues_create_fn(
    NEW.tenant_id, NEW.id, 'party', 'task',
    'report_events', 'Per-party report scheduling queue', current_user);

/**
 * @brief Reads a party as it stood at a specific version, together with its
 * identifiers and contact information as they stood during that same
 * version's [valid_from, valid_to) window. See the "Temporal composite
 * entity versioning" architecture doc.
 */

#include <chrono>
#include <string>
#include <optional>
#include <boost/uuid/uuid.hpp>

#include <string>
#include <optional>
#include "sqlgen/Timestamp.hpp"
#include "sqlgen/PrimaryKey.hpp"

#include "ores.refdata.api/domain/party_contact_information.hpp"
#include "ores.refdata.api/domain/party_identifier.hpp"

#include "ores.refdata.core/service/party_contact_information_service.hpp"
#include "ores.refdata.core/service/party_identifier_service.hpp"
#include "ores.database/service/tenant_context.hpp"
#include "ores.service/messaging/workflow_helpers.hpp"
#include <chrono>

using ores::service::messaging::is_workflow_command;
using ores::service::messaging::extract_workflow_header;
using ores::service::messaging::publish_step_completion;
using ores::service::messaging::check_step_idempotency;
using ores::workflow::messaging::step_id_header;
using ores::workflow::messaging::instance_id_header;
using ores::workflow::messaging::tenant_id_header;

// Workflow step command: bypass JWT auth; use X-Tenant-Id for context.
if (is_workflow_command(msg)) {
    const auto step_id = extract_workflow_header(msg, step_id_header);
    const auto inst_id = extract_workflow_header(msg, instance_id_header);
    const auto tenant_id = extract_workflow_header(msg, tenant_id_header);

    // Idempotency guard: replay cached result if this step already completed.
    if (auto cached = check_step_idempotency(nats_, step_id)) {
        publish_step_completion(nats_,
                                step_id,
                                inst_id,
                                cached->outcome,
                                cached->result_json,
                                cached->error_message,
                                cached->log);
        return;
    }

    auto req = decode<put_party_request>(msg);
    if (!req) {
        publish_step_completion(nats_,
                                step_id,
                                inst_id,
                                ores::workflow::messaging::step_outcome::failed,
                                "",
                                "Failed to decode put_party_request");
        return;
    }
    try {
        using ores::database::service::tenant_context;
        auto wf_ctx = tenant_context::with_tenant(ctx_, tenant_id);
        service::party_service svc(wf_ctx);
        // The service answers the canonical response, whose result
        // states the outcome. A refusal is the step failing.
        const auto resp = svc.put_party(*req);
        if (resp.result.outcome != ores::utility::domain::outcome::ok) {
            publish_step_completion(nats_,
                                    step_id,
                                    inst_id,
                                    ores::workflow::messaging::step_outcome::failed,
                                    "",
                                    resp.result.message);
            return;
        }
        BOOST_LOG_SEV(party_handler_lg(), debug)
            << "Workflow step completed: " << msg.subject;
        publish_step_completion(nats_,
                                step_id,
                                inst_id,
                                ores::workflow::messaging::step_outcome::completed,
                                rfl::json::write(resp),
                                "");
    } catch (const std::exception& e) {
        BOOST_LOG_SEV(party_handler_lg(), error)
            << "Workflow step failed: " << msg.subject << " — " << e.what();
        publish_step_completion(nats_,
                                step_id,
                                inst_id,
                                ores::workflow::messaging::step_outcome::failed,
                                "",
                                e.what());
    }
    return;
}

void composite_as_of(ores::nats::message msg) {
    BOOST_LOG_SEV(party_handler_lg(), debug) << "Handling " << msg.subject;
    auto ctx_expected = ores::service::service::make_request_context(ctx_, msg, verifier_);
    if (!ctx_expected) {
        error_reply(nats_, msg, ctx_expected.error());
        return;
    }
    const auto& ctx = *ctx_expected;
    auto req = decode<get_party_composite_as_of_request>(msg);
    if (!req) {
        BOOST_LOG_SEV(party_handler_lg(), warn) << "Failed to decode: " << msg.subject;
        reply(nats_,
              msg,
              get_party_composite_as_of_response{.success = false,
                                                 .message = "Failed to decode request"});
        return;
    }
    try {
        service::party_service party_svc(ctx);
        const auto version = static_cast<std::uint32_t>(req->version);
        auto current = party_svc.get_party_at_version(req->id, version);
        if (!current) {
            reply(nats_,
                  msg,
                  get_party_composite_as_of_response{.success = false,
                                                     .message =
                                                         "No such party version: " + req->id +
                                                         " v" + std::to_string(version)});
            return;
        }

        // Windows are contiguous by construction: the next version's
        // valid_from is this version's valid_to. If there is no next
        // version, this is the current one — its window is still open,
        // so bound it with a safely-far-future instant instead (see the
        // "Temporal composite entity versioning" architecture doc; the
        // domain object does not surface valid_to directly).
        auto next = party_svc.get_party_at_version(req->id, version + 1);
        const auto window_end =
            next ? next->recorded_at :
                   std::chrono::system_clock::now() + std::chrono::hours(24 * 365 * 100);

        service::party_identifier_service identifier_svc(ctx);
        auto identifiers = identifier_svc.list_party_identifiers_by_party_id_as_of(
            req->id, current->recorded_at, window_end);

        service::party_contact_information_service contact_svc(ctx);
        auto contacts = contact_svc.list_party_contact_informations_by_party_id_as_of(
            req->id, current->recorded_at, window_end);

        BOOST_LOG_SEV(party_handler_lg(), debug) << "Completed " << msg.subject;
        reply(nats_,
              msg,
              get_party_composite_as_of_response{.success = true,
                                                 .party = std::move(*current),
                                                 .identifiers = std::move(identifiers),
                                                 .contacts = std::move(contacts)});
    } catch (const std::exception& e) {
        BOOST_LOG_SEV(party_handler_lg(), error) << msg.subject << " failed: " << e.what();
        reply(nats_,
              msg,
              get_party_composite_as_of_response{.success = false, .message = e.what()});
    }
}

subs.push_back(nats.queue_subscribe(
    get_party_composite_as_of_request::nats_subject, queue_group, [h](ores::nats::message msg) {
        h->composite_as_of(std::move(msg));
    }));

using children_map = immer::map<boost::uuids::uuid, std::vector<boost::uuids::uuid>, key_hash>;

auto children_t = children_map{}.transient();
for (const auto& [id, p] : entries) {
    if (p.parent_party_id) {
        const auto* existing = children_t.find(*p.parent_party_id);
        auto siblings = existing ? *existing : std::vector<boost::uuids::uuid>{};
        siblings.push_back(id);
        children_t.set(*p.parent_party_id, std::move(siblings));
    }
}
auto aux = children_t.persistent();

std::vector<boost::uuids::uuid>
compute_visible_party_ids(const std::string& tenant_id, const boost::uuids::uuid& root_id) const {
    const auto snap = cache_.snapshot(tenant_id);
    if (!snap)
        return {root_id};
    std::vector<boost::uuids::uuid> result;
    std::vector<boost::uuids::uuid> stack{root_id};
    while (!stack.empty()) {
        const auto node = stack.back();
        stack.pop_back();
        result.push_back(node);
        const auto* siblings = snap->aux.find(node);
        if (siblings)
            for (const auto& child : *siblings)
                stack.push_back(child);
    }
    return result;
}