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

false

-- Sequence used to generate a unique auto-generated codename.  nextval() is
-- outside MVCC: it advances even within a multi-row INSERT statement, so every
-- row in a batch gets a different value, 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 the codename from a sequence value: a whimsical name
-- and a four-digit number. The sequence is outside MVCC (nextval
-- always advances), so every row in a multi-row INSERT gets a
-- distinct value, and distinct values give distinct codenames
-- without a NOT EXISTS check that cannot see sibling rows.
if NEW.codename = '' or NEW.codename is null then
    NEW.codename := ores_utility_whimsical_codename_fn(
        nextval('ores_refdata_party_codename_seq'));
end if;
-- Validate the final codename.
if NEW.codename !~ '^[a-z][a-z_]+_[0-9]{4}$' then
    raise exception 'codename must match ^[a-z][a-z_]+_[0-9]{4}$ 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.
 */

/**
 * @brief Writes a party and the children a confirmed review staged, in one
 * transaction. A refusal or a failure leaves none of them behind.
 */

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

#include <string>
#include <optional>

#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 "ores.utility/codec/wire_traits_fwd.hpp"
#include <boost/uuid/uuid_io.hpp>
#include <chrono>
#include "ores.database/repository/unit_of_work.hpp"

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, tenant_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,
                                ores::utility::codec::wire_traits<put_party_response>::write_json(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;
    if (!has_permission(ctx, "refdata::parties:read")) {
        error_reply(nats_, msg, ores::service::error_code::forbidden);
        return;
    }
    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: " +
                                                         boost::uuids::to_string(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(
            boost::uuids::to_string(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(
            boost::uuids::to_string(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()});
    }
}

void put_composite(ores::nats::message msg) {
    BOOST_LOG_SEV(party_handler_lg(), debug) << "Handling " << msg.subject;
    auto req_ctx_expected = ores::service::service::make_request_context(ctx_, msg, verifier_);
    if (!req_ctx_expected) {
        error_reply(nats_, msg, req_ctx_expected.error());
        return;
    }
    const auto& req_ctx = *req_ctx_expected;
    if (!has_permission(req_ctx, "refdata::parties:write")) {
        error_reply(nats_, msg, ores::service::error_code::forbidden);
        return;
    }
    auto req = decode<put_party_composite_request>(msg);
    if (!req) {
        BOOST_LOG_SEV(party_handler_lg(), warn) << "Failed to decode: " << msg.subject;
        error_reply(nats_, msg, ores::service::error_code::bad_request);
        return;
    }
    try {
        // One transaction over every row the confirm staged. The unit of
        // work rolls back when it is destroyed uncommitted, so a refusal
        // anywhere leaves none of the writes behind. Each service is built
        // on the unit of work's context, which is what makes its repository
        // calls join the transaction rather than open their own.
        ores::database::repository::unit_of_work uow(req_ctx);
        const auto& uctx = uow.ctx();
        const auto reason =
            req->intent.reason_code.empty() ?
                std::string(ores::service::messaging::change_reasons::new_record) :
                req->intent.reason_code;
        const auto apply_intent = [&](auto& row) {
            if (row.change_reason_code.empty())
                row.change_reason_code = reason;
            row.change_commentary = req->intent.commentary;
        };

        auto party = req->party;
        apply_intent(party);
        service::party_service parties(uctx);
        parties.save_party(party);

        auto identifiers = req->identifiers;
        for (auto& row : identifiers)
            apply_intent(row);
        service::party_identifier_service identifier_service(uctx);
        identifier_service.save_party_identifiers(identifiers);

        auto contacts = req->contacts;
        for (auto& row : contacts)
            apply_intent(row);
        service::party_contact_information_service contact_service(uctx);
        contact_service.save_party_contact_informations(contacts);

        uow.commit();

        // Read the row back from the committed transaction, so the reply
        // carries the version and the audit stamps the store gave it.
        service::party_service reader(req_ctx);
        auto written = reader.find_party(party.id);
        BOOST_LOG_SEV(party_handler_lg(), debug) << "Completed " << msg.subject;
        reply(nats_,
              msg,
              put_party_composite_response{.result = {},
                                           .party = written ? std::move(*written) :
                                                              std::move(party)});
    } catch (const std::exception& e) {
        BOOST_LOG_SEV(party_handler_lg(), error) << msg.subject << " failed: " << e.what();
        put_party_composite_response failure;
        failure.result.outcome = ores::utility::domain::outcome::failed;
        failure.result.code = "internal_error";
        failure.result.message = e.what();
        reply(nats_, msg, failure);
    }
}

subs.push_back(nats.queue_subscribe(
    get_party_composite_as_of_request::nats_subject, party_queue_group, [h](ores::nats::message msg) {
        h->composite_as_of(std::move(msg));
    }));
subs.push_back(nats.queue_subscribe(
    put_party_composite_request::nats_subject, party_queue_group, [h](ores::nats::message msg) {
        h->put_composite(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;
}