|
@@ -147,6 +147,9 @@ void WebServerService::start() {
|
|
|
// Start workflow scheduler
|
|
// Start workflow scheduler
|
|
|
scheduler_->start();
|
|
scheduler_->start();
|
|
|
|
|
|
|
|
|
|
+ // Load active workflows from database and register with scheduler
|
|
|
|
|
+ loadScheduledWorkflows();
|
|
|
|
|
+
|
|
|
// Start NodeSync gRPC server for runners
|
|
// Start NodeSync gRPC server for runners
|
|
|
node_sync_server_->start();
|
|
node_sync_server_->start();
|
|
|
|
|
|
|
@@ -227,6 +230,64 @@ void WebServerService::setupRoutes() {
|
|
|
LOG_INFO("API routes registered");
|
|
LOG_INFO("API routes registered");
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+void WebServerService::loadScheduledWorkflows() {
|
|
|
|
|
+ LOG_INFO("Loading scheduled workflows from database...");
|
|
|
|
|
+
|
|
|
|
|
+ // Query all active workflows
|
|
|
|
|
+ storage::QueryOptions options;
|
|
|
|
|
+ options.filters.push_back({"active", true});
|
|
|
|
|
+ options.page_size = 1000; // Load up to 1000 workflows
|
|
|
|
|
+
|
|
|
|
|
+ auto result = storage_->query("workflows", options);
|
|
|
|
|
+ if (result.failed()) {
|
|
|
|
|
+ LOG_ERROR("Failed to load workflows: {}", result.error().message());
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ int registered_count = 0;
|
|
|
|
|
+ for (const auto& workflow : result.value().documents) {
|
|
|
|
|
+ std::string workflow_id = workflow.value("_id", "");
|
|
|
|
|
+ std::string workflow_name = workflow.value("name", "");
|
|
|
|
|
+ auto nodes = workflow.value("nodes", nlohmann::json::array());
|
|
|
|
|
+
|
|
|
|
|
+ // Find scheduled trigger nodes
|
|
|
|
|
+ for (const auto& node : nodes) {
|
|
|
|
|
+ std::string node_id = node.value("id", "");
|
|
|
|
|
+ std::string node_type = node.value("type", "");
|
|
|
|
|
+
|
|
|
|
|
+ // Get node definition to check if it's a scheduled trigger
|
|
|
|
|
+ auto node_result = node_store_->get(node_type);
|
|
|
|
|
+ if (node_result.failed()) {
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ const auto& node_def = node_result.value();
|
|
|
|
|
+ if (!node_def.is_trigger || !node_def.is_scheduled) {
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // Get interval from node config
|
|
|
|
|
+ auto config = node.value("config", nlohmann::json::object());
|
|
|
|
|
+ int interval = config.value("pollInterval", 0);
|
|
|
|
|
+
|
|
|
|
|
+ if (interval > 0) {
|
|
|
|
|
+ scheduler_->registerWorkflow(
|
|
|
|
|
+ workflow_id,
|
|
|
|
|
+ workflow_name,
|
|
|
|
|
+ node_id,
|
|
|
|
|
+ node_type,
|
|
|
|
|
+ interval
|
|
|
|
|
+ );
|
|
|
|
|
+ registered_count++;
|
|
|
|
|
+ LOG_DEBUG("Registered workflow '{}' ({}) for scheduled execution every {} minutes",
|
|
|
|
|
+ workflow_name, workflow_id, interval);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ LOG_INFO("Loaded {} scheduled workflows", registered_count);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
void WebServerService::executeScheduledWorkflow(const std::string& workflow_id,
|
|
void WebServerService::executeScheduledWorkflow(const std::string& workflow_id,
|
|
|
const std::string& trigger_node_id,
|
|
const std::string& trigger_node_id,
|
|
|
const std::string& trigger_type) {
|
|
const std::string& trigger_type) {
|