|
|
@@ -1,6 +1,8 @@
|
|
|
#include "execution_controller.hpp"
|
|
|
#include "../webserver_service.hpp"
|
|
|
#include "logging/logger.hpp"
|
|
|
+#include "proto/runner.grpc.pb.h"
|
|
|
+#include <grpcpp/grpcpp.h>
|
|
|
|
|
|
namespace smartbotic::webserver::api {
|
|
|
|
|
|
@@ -8,9 +10,10 @@ ExecutionController::ExecutionController(storage::StorageClient& storage,
|
|
|
auth::AuthMiddleware& middleware,
|
|
|
WebSocketServer& ws_server,
|
|
|
WorkflowScheduler& scheduler,
|
|
|
+ runners::LoadBalancer& load_balancer,
|
|
|
FailureHandler on_failure)
|
|
|
: storage_(storage), middleware_(middleware), ws_server_(ws_server),
|
|
|
- scheduler_(scheduler), on_failure_(std::move(on_failure)) {}
|
|
|
+ scheduler_(scheduler), load_balancer_(load_balancer), on_failure_(std::move(on_failure)) {}
|
|
|
|
|
|
void ExecutionController::registerRoutes(httplib::Server& server) {
|
|
|
server.Get("/api/v1/executions", [this](const httplib::Request& req, httplib::Response& res) {
|
|
|
@@ -19,6 +22,15 @@ void ExecutionController::registerRoutes(httplib::Server& server) {
|
|
|
});
|
|
|
});
|
|
|
|
|
|
+ // Registered before the generic "/api/v1/executions/([^/]+)" GET route
|
|
|
+ // below, otherwise that regex captures "pending" as an execution id and
|
|
|
+ // this listing 404s.
|
|
|
+ server.Get("/api/v1/executions/pending", [this](const httplib::Request& req, httplib::Response& res) {
|
|
|
+ middleware_.requireAuth(req, res, [this](auto& req, auto& res, auto& ctx) {
|
|
|
+ listPending(req, res, ctx);
|
|
|
+ });
|
|
|
+ });
|
|
|
+
|
|
|
server.Get(R"(/api/v1/executions/([^/]+))", [this](const httplib::Request& req, httplib::Response& res) {
|
|
|
middleware_.requireAuth(req, res, [this](auto& req, auto& res, auto& ctx) {
|
|
|
getExecution(req, res, ctx);
|
|
|
@@ -37,6 +49,12 @@ void ExecutionController::registerRoutes(httplib::Server& server) {
|
|
|
});
|
|
|
});
|
|
|
|
|
|
+ server.Post(R"(/api/v1/executions/([^/]+)/resume)", [this](const httplib::Request& req, httplib::Response& res) {
|
|
|
+ middleware_.requireAuth(req, res, [this](auto& req, auto& res, auto& ctx) {
|
|
|
+ resumeExecution(req, res, ctx);
|
|
|
+ });
|
|
|
+ });
|
|
|
+
|
|
|
// Internal endpoint for runner to send execution events (no auth required from internal network)
|
|
|
server.Post("/api/v1/internal/execution-event", [this](const httplib::Request& req, httplib::Response& res) {
|
|
|
receiveExecutionEvent(req, res);
|
|
|
@@ -131,6 +149,94 @@ void ExecutionController::retryExecution(const httplib::Request& req, httplib::R
|
|
|
sendJson(res, {{"success", true}, {"message", "Retry queued"}});
|
|
|
}
|
|
|
|
|
|
+void ExecutionController::resumeExecution(const httplib::Request& req, httplib::Response& res,
|
|
|
+ const auth::AuthContext& ctx) {
|
|
|
+ std::string execution_id = req.matches[1];
|
|
|
+
|
|
|
+ nlohmann::json body;
|
|
|
+ try {
|
|
|
+ body = req.body.empty() ? nlohmann::json::object() : nlohmann::json::parse(req.body);
|
|
|
+ } catch (...) {
|
|
|
+ sendError(res, "Body is not valid JSON", 400);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ const std::string token = body.value("token", "");
|
|
|
+ if (token.empty()) {
|
|
|
+ sendError(res, "A token is required to answer a paused execution", 400);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ nlohmann::json payload = body.value("data", nlohmann::json::object());
|
|
|
+ payload["approved"] = body.value("approved", true);
|
|
|
+ payload["answeredBy"] = ctx.user_id;
|
|
|
+
|
|
|
+ auto runner = load_balancer_.selectRunner();
|
|
|
+ if (!runner) {
|
|
|
+ sendError(res, "No runners available", 503);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ // Qualified with the leading "::" because smartbotic::webserver::grpc (a
|
|
|
+ // forward-declared namespace for the node sync / credential gRPC servers,
|
|
|
+ // pulled in via webserver_service.hpp) would otherwise shadow the real
|
|
|
+ // ::grpc namespace here.
|
|
|
+ auto channel = ::grpc::CreateChannel(runner->address, ::grpc::InsecureChannelCredentials());
|
|
|
+ auto stub = proto::RunnerService::NewStub(channel);
|
|
|
+
|
|
|
+ proto::ResumeExecutionRequest grpc_req;
|
|
|
+ grpc_req.set_execution_id(execution_id);
|
|
|
+ grpc_req.set_token(token);
|
|
|
+ grpc_req.set_payload(payload.dump());
|
|
|
+
|
|
|
+ proto::ExecuteWorkflowResponse grpc_res;
|
|
|
+ ::grpc::ClientContext grpc_ctx;
|
|
|
+ grpc_ctx.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(60));
|
|
|
+
|
|
|
+ auto status = stub->ResumeExecution(&grpc_ctx, grpc_req, &grpc_res);
|
|
|
+ if (!status.ok()) {
|
|
|
+ sendError(res, "Could not resume: " + status.error_message(), 400);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ ws_server_.broadcast("executions." + execution_id + ".resumed", {
|
|
|
+ {"executionId", execution_id},
|
|
|
+ {"answeredBy", ctx.user_id}
|
|
|
+ });
|
|
|
+
|
|
|
+ sendJson(res, {{"executionId", execution_id}, {"status", "resumed"}});
|
|
|
+}
|
|
|
+
|
|
|
+void ExecutionController::listPending(const httplib::Request& req, httplib::Response& res,
|
|
|
+ const auth::AuthContext& ctx) {
|
|
|
+ storage::QueryOptions options;
|
|
|
+ options.filters.push_back({"status", "waiting"});
|
|
|
+
|
|
|
+ auto result = storage_.query("executions", options);
|
|
|
+ if (result.failed()) {
|
|
|
+ sendError(res, "Could not list pending approvals", 500);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ // Deliberately no pause token here: listing is a weaker permission than
|
|
|
+ // answering, and anyone who could list every pending approval would
|
|
|
+ // otherwise be able to answer all of them. The token reaches an approver
|
|
|
+ // through the node's own output on the execution detail.
|
|
|
+ nlohmann::json pending = nlohmann::json::array();
|
|
|
+ for (const auto& record : result.value().documents) {
|
|
|
+ pending.push_back({
|
|
|
+ {"executionId", record.value("_id", "")},
|
|
|
+ {"workflowId", record.value("workflowId", "")},
|
|
|
+ {"workflowName", record.value("workflowName", "")},
|
|
|
+ {"pausedNodeId", record.value("pausedNodeId", "")},
|
|
|
+ {"pauseExpiresAt", record.value("pauseExpiresAt", static_cast<int64_t>(0))},
|
|
|
+ {"startedAt", record.value("startedAt", static_cast<int64_t>(0))}
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ sendJson(res, {{"pending", pending}, {"total", pending.size()}});
|
|
|
+}
|
|
|
+
|
|
|
void ExecutionController::receiveExecutionEvent(const httplib::Request& req, httplib::Response& res) {
|
|
|
LOG_DEBUG("Received execution event request: {}", req.body);
|
|
|
|