Workflow Monitor — Design & Implementation Plan

namespace ores::workflow::eventing {

struct workflow_instance_changed_event final {
    std::chrono::system_clock::time_point timestamp;
    std::vector<std::string> ids;   // changed instance UUIDs
    std::string tenant_id;
};

}  // ores::workflow::eventing

namespace ores::eventing::domain {

template<>
struct event_traits<ores::workflow::eventing::workflow_instance_changed_event> {
    static constexpr std::string_view name =
        "ores.workflow.workflow_instance_changed";
};

}  // ores::eventing::domain

// --- List instances ---

struct workflow_instance_summary {
    std::string id;
    std::string type;              // e.g. "ore_import_workflow"
    std::string status;            // in_progress | completed | failed |
                                   // compensating | compensated
    int current_step_index = 0;
    int step_count = 0;
    std::string correlation_id;
    std::string created_by;
    std::string created_at;        // ISO-8601 UTC string
    std::optional<std::string> completed_at;
    std::string error;
};

struct list_workflow_instances_request {
    using response_type = list_workflow_instances_response;
    static constexpr std::string_view nats_subject =
        "workflow.v1.instances.list";
    int limit = 200;
    std::optional<std::string> status_filter;  // empty = all
};

struct list_workflow_instances_response {
    bool success = false;
    std::string message;
    std::vector<workflow_instance_summary> instances;
};

// --- Get steps ---

struct workflow_step_summary {
    std::string id;
    std::string name;
    std::string status;            // pending | in_progress | completed | failed
    int step_index = 0;
    std::string created_at;
    std::optional<std::string> started_at;
    std::optional<std::string> completed_at;
    std::string error;
};

struct get_workflow_steps_request {
    using response_type = get_workflow_steps_response;
    static constexpr std::string_view nats_subject =
        "workflow.v1.instances.steps";
    std::string workflow_instance_id;
};

struct get_workflow_steps_response {
    bool success = false;
    std::string message;
    std::vector<workflow_step_summary> steps;
};

void workflow_engine::publish_status_event(
    const boost::uuids::uuid& instance_id,
    const boost::uuids::uuid& tenant_id) {

    using ev = ores::eventing::domain::entity_change_event;
    ev e;
    e.entity     = "ores.workflow.workflow_instance";
    e.timestamp  = std::chrono::system_clock::now();
    e.entity_ids = { boost::uuids::to_string(instance_id) };
    e.tenant_id  = boost::uuids::to_string(tenant_id);

    const auto json = rfl::json::write(e);
    const auto data = std::as_bytes(std::span{json.data(), json.size()});
    try {
        nats_.publish("ores.workflow.workflow_instance_changed", data, {});
    } catch (const std::exception& ex) {
        BOOST_LOG_SEV(lg(), warn)
            << "Failed to publish status event: " << ex.what();
    }
}

#include <span>
#include "ores.eventing/domain/entity_change_event.hpp"

class workflow_query_handler {
public:
    workflow_query_handler(ores::nats::service::client& nats,
                           ores::database::context ctx,
                           ores::security::jwt::jwt_authenticator signer);

    void list_instances(ores::nats::message msg);
    void get_steps(ores::nats::message msg);

private:
    ores::nats::service::client& nats_;
    ores::database::context ctx_;
    ores::security::jwt::jwt_authenticator signer_;
    repository::workflow_instance_repository instance_repo_;
    repository::workflow_step_repository step_repo_;
};

auto qh = std::make_shared<messaging::workflow_query_handler>(
    nats, ctx, signer);

subs.push_back(nats.queue_subscribe(
    messaging::list_workflow_instances_request::nats_subject, qg,
    [qh](ores::nats::message msg) {
        qh->list_instances(std::move(msg));
    }));

subs.push_back(nats.queue_subscribe(
    messaging::get_workflow_steps_request::nats_subject, qg,
    [qh](ores::nats::message msg) {
        qh->get_steps(std::move(msg));
    }));

add_subdirectory(src)

set(name "ores.qt.workflow")
set(lib_binary_name ${name})
set(lib_target_name ${name}.lib)
set(ORES_QT_WORKFLOW_DIR ${CMAKE_SOURCE_DIR}/projects/ores.qt.workflow)

set(CMAKE_AUTOMOC ON)

file(GLOB_RECURSE files RELATIVE "${CMAKE_CURRENT_SOURCE_DIR}/"
    "${CMAKE_CURRENT_SOURCE_DIR}/*.cpp")
file(GLOB_RECURSE HEADERS "${ORES_QT_WORKFLOW_DIR}/include/*.hpp")

add_library(${lib_target_name} SHARED ${files} ${HEADERS})

set_target_properties(${lib_target_name} PROPERTIES
    OUTPUT_NAME          ${lib_binary_name}
    VERSION              ${PROJECT_VERSION}
    SOVERSION            ${PROJECT_VERSION_MAJOR}
    LIBRARY_OUTPUT_DIRECTORY "${ORES_PLUGIN_OUTPUT_DIRECTORY}"
    INSTALL_RPATH        "$ORIGIN/../lib:$ORIGIN"
    BUILD_RPATH          "${CMAKE_LIBRARY_OUTPUT_DIRECTORY}")

target_include_directories(${lib_target_name}
    PUBLIC  ${ORES_QT_WORKFLOW_DIR}/include
    PRIVATE "${CMAKE_CURRENT_BINARY_DIR}/${lib_target_name}_autogen/include"
            "${CMAKE_SOURCE_DIR}/projects/ores.qt/include")

target_link_libraries(${lib_target_name}
    PUBLIC
        ores.qt.api.lib
        ores.workflow.api.lib
        ores.eventing.lib
    PRIVATE
        ores.qt.lib)

install(TARGETS ${lib_target_name}
    LIBRARY DESTINATION bin
    RUNTIME DESTINATION bin)

add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/ores.qt.workflow)

// WorkflowPlugin.hpp
class WorkflowPlugin : public PluginBase {
    Q_OBJECT
    Q_PLUGIN_METADATA(IID "ores.qt.IPlugin/1.0")
    Q_INTERFACES(ores::qt::IPlugin)
public:
    explicit WorkflowPlugin(QObject* parent = nullptr);
    ~WorkflowPlugin() override;

    QString name() const override { return QStringLiteral("ores.qt.workflow"); }
    int load_order() const override { return 450; }

    void on_login(const plugin_context& ctx) override;
    QList<QMenu*> create_menus() override;
    void on_logout() override;

private:
    plugin_context ctx_;
    std::unique_ptr<WorkflowController> controller_;
};

class WorkflowController : public EntityController {
    Q_OBJECT
public:
    explicit WorkflowController(QMainWindow* mainWindow,
                                QMdiArea* mdiArea,
                                ClientManager* clientManager,
                                const QString& username,
                                QObject* parent = nullptr);
    ~WorkflowController() override;

public slots:
    void showDashboardWindow();
    void showListWindow();

private slots:
    void onNotificationReceived(const QString& eventType,
                                const QDateTime& timestamp,
                                const QStringList& entityIds,
                                const QString& tenantId);

private:
    ClientManager* clientManager_;
    QString username_;
    QPointer<WorkflowMdiWindow> mdiWindow_;
};

constexpr std::string_view workflow_event_name =
    eventing::domain::event_traits<
        workflow::eventing::workflow_instance_changed_event>::name;

// In constructor:
connect(clientManager_, &ClientManager::notificationReceived,
        this, &WorkflowController::onNotificationReceived);

auto subscribeAll = [self = QPointer<WorkflowController>(this)]() {
    if (!self) return;
    self->clientManager_->subscribeToEvent(std::string{workflow_event_name});
};
connect(clientManager_, &ClientManager::loggedIn,    this, subscribeAll);
connect(clientManager_, &ClientManager::reconnected, this, subscribeAll);
if (clientManager_->isConnected()) subscribeAll();

add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/ores.qt.workflow)