feat: the database-change trigger is told about changes instead of polling for them

It used to be a pseudo-trigger: something you wired after a schedule node,
which every tick asked the database for documents newer than a stored
cursor. That only noticed as often as it was asked, needed a cursor
document per trigger to remember where it got to, and never saw a change
made and undone between two ticks.

The database can say when a collection changes, so the webserver listens.
A workflow with this trigger gets a subscription while it is active, and a
change starts it with the document already attached - no interval, no
cursor, nothing missed in between.

The subscription lives in the webserver because that is where triggers
are already registered and where runs are dispatched from. It is opened
on activate and at startup, and closed on deactivate and on delete.

Things this turned up:

- subscribe() is not project-scoped, unlike the document calls: it wants
  the fully qualified collection name and reports events under it too.
  Subscribing to a bare name silently delivers nothing, which looks
  exactly like a collection nobody is writing to. The adapter now
  qualifies going in and strips coming out.
- context.triggerData does not exist. The JS context carries only
  executionId, nodeId and workflowId; a trigger node is handed its
  trigger data as its input. Reading context.triggerData gets undefined
  every time, so the node quietly took the polling path on every event.
  Worth knowing: nodes/triggers/get-trigger.js reads it the same way.
- Deactivating a workflow reached straight past the function that owns
  trigger registration and unregistered the scheduler itself, so a
  workflow switched off kept its database watch and went on running.
  Deleting one had the same gap. Both now go through the one function
  that knows about every kind of trigger.

The node declares @trigger now, which means the engine will only run it
at the start of a flow - a trigger node found anywhere else is skipped.
No stored workflow used it, so nothing in production changes. A manual
run has no event to work from and still falls back to asking what changed
since last time, so testing a workflow by hand keeps working.

Verified live against a watched collection: insert, update and delete each
started the workflow on their own with the right event type, delete
carrying no document, the run stopping once the workflow was deactivated,
and the watch coming back by itself after a restart.

CMakeLists.txt

@@ -161,6 +161,7 @@ add_executable(smartbotic-webserver
     src/webserver/runners/load_balancer.cpp
     src/webserver/nodes/node_store.cpp
     src/webserver/scheduler/workflow_scheduler.cpp
+    src/webserver/scheduler/database_watcher.cpp
     src/webserver/grpc/node_sync_service.cpp
     src/webserver/grpc/credential_service.cpp
     src/webserver/api/auth_controller.cpp

lib/storage/storage_client.cpp

@@ -237,6 +237,39 @@ Result<void> StorageClient::createCollection(const std::string& name,
     return {};
 }
 
+std::shared_ptr<void> StorageClient::subscribe(const std::vector<std::string>& collections,
+                                               ChangeCallback callback) {
+    // subscribe() is not project-scoped, unlike the document calls: it wants the
+    // fully qualified name and reports events under it too. Verified against
+    // 2.4.5 - subscribing to a bare name silently delivers nothing at all,
+    // which looks exactly like a collection nobody is writing to. So the names
+    // are qualified going in and stripped coming out, and callers above this
+    // layer never see the difference.
+    const bool is_default = project_isDefault(impl_->project_);
+    const std::string prefix = is_default ? std::string() : impl_->project_ + ":";
+
+    std::vector<std::string> qualified;
+    qualified.reserve(collections.size());
+    for (const auto& name : collections) {
+        qualified.push_back(name.find(':') == std::string::npos ? prefix + name : name);
+    }
+
+    return impl_->client_->subscribe(
+        qualified,
+        [callback = std::move(callback), prefix](const std::string& collection,
+                                                 const std::string& id,
+                                                 const std::string& event_type,
+                                                 const std::optional<nlohmann::json>& data) {
+            std::string name = collection;
+            if (!prefix.empty() && name.rfind(prefix, 0) == 0) {
+                name = name.substr(prefix.size());
+            }
+            // A delete carries no document. Callers get an empty object rather
+            // than having to reason about an absent one.
+            callback(name, id, event_type, data.value_or(nlohmann::json::object()));
+        });
+}
+
 std::vector<CollectionInfo> StorageClient::listCollectionInfos() {
     std::vector<CollectionInfo> out;
     for (const auto& name : listCollections()) {

lib/storage/storage_client.hpp

@@ -2,6 +2,7 @@
 
 #include <map>
 #include <memory>
+#include <functional>
 #include <optional>
 #include <string>
 #include <vector>
@@ -192,6 +193,19 @@ public:
     // than a hot path.
     std::vector<CollectionInfo> listCollectionInfos();
 
+    // Server-side change events, so a watcher does not have to poll and keep a
+    // timestamp cursor. `document` is empty for a delete.
+    //
+    // The returned handle owns the subscription: let it go and the stream is
+    // closed. The callback runs on the client's own thread, so it must not
+    // block - hand the work somewhere else and return.
+    using ChangeCallback = std::function<void(const std::string& collection,
+                                              const std::string& id,
+                                              const std::string& event_type,
+                                              const nlohmann::json& document)>;
+    std::shared_ptr<void> subscribe(const std::vector<std::string>& collections,
+                                    ChangeCallback callback);
+
     common::Result<nlohmann::json> getVersion(const std::string& collection,
                                               const std::string& id,
                                               int64_t version);

nodes/triggers/database-change.js

@@ -3,8 +3,9 @@
  * @name Database Change
  * @category triggers
  * @version 1.0.0
- * @description Report documents added or changed in a collection since the last run, paired with a schedule trigger for its cadence
+ * @description Start the workflow when a collection changes. The database reports the change, so nothing polls and nothing is missed between ticks
  * @icon database
+ * @trigger
  */
 
 const configSchema = {
@@ -45,6 +46,13 @@ const configSchema = {
             description: 'Off records the current high-water mark and reports nothing, so adding this to a live workflow does not fire for every document already there',
             default: false
         },
+        eventTypes: {
+            type: 'array',
+            title: 'Event Types',
+            description: 'Which changes should start the workflow. Empty means all of them',
+            items: { type: 'string', enum: ['insert', 'update', 'delete'] },
+            default: []
+        },
         cursorCollection: {
             dynamicOptions: {
                 source: 'storage.collections',
@@ -54,6 +62,7 @@ const configSchema = {
             },
             type: 'string',
             title: 'Cursor Collection',
+            description: 'Only used by a manual run, which has no event to work from and asks what changed since last time instead',
             default: 'watch_cursors'
         }
     },
@@ -73,7 +82,10 @@ const outputSchema = {
         documents: { type: 'array', description: 'Documents new or changed since the last run' },
         count: { type: 'number' },
         isFirstRun: { type: 'boolean' },
-        cursor: { type: 'number', description: 'High-water timestamp stored for the next run' }
+        cursor: { type: 'number', description: 'High-water timestamp stored for the next run. Zero for an event, which needs no cursor' },
+        eventType: { type: 'string', description: 'insert, update or delete - only present when started by a change' },
+        documentId: { type: 'string', description: 'The document that changed - only present when started by a change' },
+        collection: { type: 'string' }
     }
 };
 
@@ -83,6 +95,35 @@ async function execute(config, input, context) {
         throw new Error('Database Change: a collection is required');
     }
 
+    // The usual path: the database said what changed and the workflow was
+    // started because of it. Nothing to look up - the document is right here.
+    //
+    // A trigger node is handed the trigger data as its input; context carries
+    // only the execution, node and workflow ids. Reading context.triggerData
+    // gets undefined every time, which looks like "no event" and quietly sends
+    // every run down the polling path.
+    const triggerData = (input && typeof input === 'object' ? input : {});
+    if (triggerData.eventType && triggerData.documentId) {
+        const changed = triggerData.document && Object.keys(triggerData.document).length
+            ? [triggerData.document]
+            : [];
+        smartbotic.log.info('Database Change: ' + triggerData.eventType + ' on ' +
+                            (triggerData.collection || collection) + '/' + triggerData.documentId);
+        return {
+            documents: changed,
+            count: changed.length,
+            isFirstRun: false,
+            cursor: 0,
+            eventType: triggerData.eventType,
+            documentId: triggerData.documentId,
+            collection: triggerData.collection || collection
+        };
+    }
+
+    // No event, so this is somebody running the workflow by hand to see what it
+    // does. Falling back to the old query means a manual run still produces
+    // something to work with instead of an empty result that looks broken.
+
     const timestampField = config.timestampField || '_updated_at';
     const cursorCollection = config.cursorCollection || 'watch_cursors';
     const maxDocuments = Number(config.maxDocuments) || 100;

src/webserver/api/workflow_controller.cpp

@@ -16,6 +16,7 @@ WorkflowController::WorkflowController(storage::StorageClient& storage,
                                        runners::LoadBalancer& load_balancer,
                                        WebSocketServer& ws_server,
                                        WorkflowScheduler& scheduler,
+                                       DatabaseWatcher& db_watcher,
                                        nodes::NodeStore& node_store)
     : storage_(storage)
     , middleware_(middleware)
@@ -23,6 +24,7 @@ WorkflowController::WorkflowController(storage::StorageClient& storage,
     , load_balancer_(load_balancer)
     , ws_server_(ws_server)
     , scheduler_(scheduler)
+    , db_watcher_(db_watcher)
     , node_store_(node_store) {}
 
 void WorkflowController::materializeNodeConfigDefaults(nlohmann::json& body) {
@@ -406,8 +408,10 @@ void WorkflowController::deleteWorkflow(const httplib::Request& req, httplib::Re
                                          const auth::AuthContext& ctx) {
     std::string id = req.matches[1];
 
-    // Unregister from scheduler before deleting
-    scheduler_.unregisterWorkflow(id);
+    // Every trigger it had, not just the scheduled ones - a deleted workflow
+    // that kept its database watch would go on being started by changes, with
+    // nothing left to run.
+    updateScheduledTriggers(id, false);
 
     auto result = storage_.remove("workflows", id);
     if (result.failed()) {
@@ -677,8 +681,10 @@ void WorkflowController::deactivateWorkflow(const httplib::Request& req, httplib
         return;
     }
 
-    // Unregister from scheduler
-    scheduler_.unregisterWorkflow(id);
+    // Through the one function that owns trigger registration, not straight at
+    // the scheduler. Reaching past it meant a workflow switched off kept its
+    // database watch and went on running on every change.
+    updateScheduledTriggers(id, false);
 
     ws_server_.broadcast("workflows.deactivated", {{"id", id}});
     sendJson(res, {{"success", true}, {"active", false}});
@@ -687,9 +693,15 @@ void WorkflowController::deactivateWorkflow(const httplib::Request& req, httplib
 void WorkflowController::updateScheduledTriggers(const std::string& workflow_id, bool activate) {
     if (!activate) {
         scheduler_.unregisterWorkflow(workflow_id);
+        db_watcher_.unwatch(workflow_id);
         return;
     }
 
+    // Re-registering from scratch. A workflow whose trigger was edited - a
+    // different collection, or the trigger removed - would otherwise keep
+    // listening to the old one.
+    db_watcher_.unwatch(workflow_id);
+
     // Get workflow
     auto workflow_result = storage_.get("workflows", workflow_id);
     if (workflow_result.failed()) {
@@ -713,6 +725,27 @@ void WorkflowController::updateScheduledTriggers(const std::string& workflow_id,
 
         auto& node_def = node_result.value();
 
+        // A database-change trigger is told rather than asked: the database
+        // reports the change, so there is no interval to register.
+        if (node_type == "database-change") {
+            auto config = common::applyConfigDefaults(
+                node.value("config", nlohmann::json::object()), node_def.config_schema);
+            const std::string collection = config.value("collection", "");
+            if (collection.empty()) {
+                LOG_WARN("Workflow {} has a database-change trigger with no collection set",
+                         workflow_id);
+                continue;
+            }
+            std::vector<std::string> event_types;
+            if (config.contains("eventTypes") && config["eventTypes"].is_array()) {
+                for (const auto& t : config["eventTypes"]) {
+                    if (t.is_string()) event_types.push_back(t.get<std::string>());
+                }
+            }
+            db_watcher_.watch(workflow_id, workflow_name, node_id, collection, event_types);
+            continue;
+        }
+
         // Check if node has scheduling capability (is_scheduled flag or pollInterval config)
         bool is_scheduled = node_def.is_scheduled;
         if (!is_scheduled) {

src/webserver/api/workflow_controller.hpp

@@ -9,6 +9,7 @@
 #include "../websocket_server.hpp"
 #include "../scheduler/workflow_scheduler.hpp"
 #include "../nodes/node_store.hpp"
+#include "../scheduler/database_watcher.hpp"
 #include "proto/runner.grpc.pb.h"
 
 namespace smartbotic::webserver::api {
@@ -21,6 +22,7 @@ public:
                       runners::LoadBalancer& load_balancer,
                       WebSocketServer& ws_server,
                       WorkflowScheduler& scheduler,
+                      DatabaseWatcher& db_watcher,
                       nodes::NodeStore& node_store);
 
     void registerRoutes(httplib::Server& server);
@@ -62,6 +64,7 @@ private:
     runners::LoadBalancer& load_balancer_;
     WebSocketServer& ws_server_;
     WorkflowScheduler& scheduler_;
+    DatabaseWatcher& db_watcher_;
     nodes::NodeStore& node_store_;
 
     // Helper to check for scheduled triggers and register/unregister with scheduler

src/webserver/scheduler/database_watcher.cpp

@@ -0,0 +1,132 @@
+#include "database_watcher.hpp"
+
+#include <algorithm>
+
+#include "logging/logger.hpp"
+
+namespace smartbotic::webserver {
+
+DatabaseWatcher::DatabaseWatcher(storage::StorageClient& storage) : storage_(storage) {}
+
+DatabaseWatcher::~DatabaseWatcher() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    // Dropping the handles closes the streams.
+    watches_.clear();
+}
+
+void DatabaseWatcher::setExecuteCallback(ExecuteCallback callback) {
+    execute_callback_ = std::move(callback);
+}
+
+void DatabaseWatcher::watch(const std::string& workflow_id,
+                            const std::string& workflow_name,
+                            const std::string& trigger_node_id,
+                            const std::string& collection,
+                            const std::vector<std::string>& event_types) {
+    if (workflow_id.empty() || collection.empty()) {
+        return;
+    }
+
+    Watch entry;
+    entry.workflow_id = workflow_id;
+    entry.workflow_name = workflow_name;
+    entry.trigger_node_id = trigger_node_id;
+    entry.collection = collection;
+    entry.event_types = event_types;
+
+    // Subscribing happens outside the lock: the callback can fire on the
+    // client's thread the moment the stream opens, and it takes the same lock.
+    auto subscription = storage_.subscribe(
+        {collection},
+        [this, workflow_id](const std::string& changed_collection,
+                            const std::string& id,
+                            const std::string& event_type,
+                            const nlohmann::json& document) {
+            onEvent(workflow_id, changed_collection, id, event_type, document);
+        });
+
+    if (!subscription) {
+        LOG_WARN("Database watch for workflow {} on {} could not be opened",
+                 workflow_id, collection);
+        return;
+    }
+    entry.subscription = std::move(subscription);
+
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        // Assigning replaces whatever was there, and the old handle goes with
+        // it - which is how a workflow that changed collections stops listening
+        // to the one it used to watch.
+        watches_[workflow_id] = std::move(entry);
+    }
+
+    LOG_INFO("Watching {} for workflow {} ({})", collection, workflow_name, workflow_id);
+}
+
+void DatabaseWatcher::unwatch(const std::string& workflow_id) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (watches_.erase(workflow_id) > 0) {
+        LOG_INFO("Stopped watching for workflow {}", workflow_id);
+    }
+}
+
+size_t DatabaseWatcher::watchCount() const {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return watches_.size();
+}
+
+nlohmann::json DatabaseWatcher::describe() const {
+    std::lock_guard<std::mutex> lock(mutex_);
+    nlohmann::json out = nlohmann::json::array();
+    for (const auto& [id, w] : watches_) {
+        out.push_back({
+            {"workflowId", id},
+            {"workflowName", w.workflow_name},
+            {"collection", w.collection},
+            {"triggerNodeId", w.trigger_node_id},
+            {"eventTypes", w.event_types},
+        });
+    }
+    return out;
+}
+
+void DatabaseWatcher::onEvent(const std::string& workflow_id,
+                              const std::string& collection,
+                              const std::string& id,
+                              const std::string& event_type,
+                              const nlohmann::json& document) {
+    std::string trigger_node_id;
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        auto it = watches_.find(workflow_id);
+        // The workflow may have been switched off between the event being sent
+        // and this running.
+        if (it == watches_.end()) return;
+
+        const auto& wanted = it->second.event_types;
+        if (!wanted.empty() &&
+            std::find(wanted.begin(), wanted.end(), event_type) == wanted.end()) {
+            return;
+        }
+        trigger_node_id = it->second.trigger_node_id;
+    }
+
+    if (!execute_callback_) return;
+
+    nlohmann::json event = {
+        {"collection", collection},
+        {"documentId", id},
+        {"eventType", event_type},
+        {"document", document},
+    };
+
+    LOG_DEBUG("Database event {} on {}/{} starting workflow {}",
+              event_type, collection, id, workflow_id);
+
+    // Straight into the callback, which hands the run to a runner over gRPC.
+    // This is the database client's own thread, so nothing here may block for
+    // long or events queue up behind it.
+    execute_callback_(workflow_id, trigger_node_id, "database-change", event);
+}
+
+} // namespace smartbotic::webserver

src/webserver/scheduler/database_watcher.hpp

@@ -0,0 +1,81 @@
+#pragma once
+
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include <nlohmann/json.hpp>
+
+#include "storage/storage_client.hpp"
+
+namespace smartbotic::webserver {
+
+/**
+ * Starts a workflow when a collection it watches changes.
+ *
+ * The database-change trigger used to poll: every schedule tick it asked for
+ * documents whose timestamp was newer than a stored cursor. That works, but it
+ * only notices as often as it is asked, it needs a cursor document per trigger
+ * to remember where it got to, and a change made and undone between two ticks
+ * is never seen at all.
+ *
+ * The database can say when something changed, so this listens instead. There
+ * is no cursor to keep and nothing to miss between ticks.
+ */
+class DatabaseWatcher {
+public:
+    // Called when a watched collection changes. Mirrors the scheduler's
+    // callback, with the event itself so the workflow can be told what happened.
+    using ExecuteCallback = std::function<void(const std::string& workflow_id,
+                                               const std::string& trigger_node_id,
+                                               const std::string& trigger_type,
+                                               const nlohmann::json& event)>;
+
+    explicit DatabaseWatcher(storage::StorageClient& storage);
+    ~DatabaseWatcher();
+
+    DatabaseWatcher(const DatabaseWatcher&) = delete;
+    DatabaseWatcher& operator=(const DatabaseWatcher&) = delete;
+
+    void setExecuteCallback(ExecuteCallback callback);
+
+    // Watch `collection` for `workflow_id`. Replaces any watch that workflow
+    // already had, so re-registering after an edit is safe.
+    void watch(const std::string& workflow_id,
+               const std::string& workflow_name,
+               const std::string& trigger_node_id,
+               const std::string& collection,
+               const std::vector<std::string>& event_types);
+
+    void unwatch(const std::string& workflow_id);
+
+    size_t watchCount() const;
+    nlohmann::json describe() const;
+
+private:
+    struct Watch {
+        std::string workflow_id;
+        std::string workflow_name;
+        std::string trigger_node_id;
+        std::string collection;
+        std::vector<std::string> event_types;  // empty means every kind
+        std::shared_ptr<void> subscription;
+    };
+
+    void onEvent(const std::string& workflow_id,
+                 const std::string& collection,
+                 const std::string& id,
+                 const std::string& event_type,
+                 const nlohmann::json& document);
+
+    storage::StorageClient& storage_;
+    ExecuteCallback execute_callback_;
+
+    mutable std::mutex mutex_;
+    std::unordered_map<std::string, Watch> watches_;  // by workflow id
+};
+
+} // namespace smartbotic::webserver

src/webserver/webserver_service.cpp

@@ -76,6 +76,16 @@ WebServerService::WebServerService(const WebServerServiceConfig& config)
         executeScheduledWorkflow(workflow_id, trigger_node_id, trigger_type);
     });
 
+    // Triggers that are told rather than asking. The database says when a
+    // collection changed, so these workflows do not poll for it.
+    db_watcher_ = std::make_unique<DatabaseWatcher>(*storage_);
+    db_watcher_->setExecuteCallback([this](const std::string& workflow_id,
+                                           const std::string& trigger_node_id,
+                                           const std::string& trigger_type,
+                                           const nlohmann::json& event) {
+        executeScheduledWorkflow(workflow_id, trigger_node_id, trigger_type, event);
+    });
+
     // Initialize HTTP server
     HttpServerConfig http_config;
     http_config.port = config_.http_port;
@@ -229,7 +239,7 @@ void WebServerService::setupRoutes() {
 
     workflow_ctrl_ = std::make_unique<api::WorkflowController>(
         *storage_, *auth_middleware_, *runner_registry_, *load_balancer_, *ws_server_,
-        *scheduler_, *node_store_);
+        *scheduler_, *db_watcher_, *node_store_);
     workflow_ctrl_->registerRoutes(server);
 
     workflow_group_ctrl_ = std::make_unique<api::WorkflowGroupController>(
@@ -340,7 +350,32 @@ void WebServerService::loadScheduledWorkflows() {
             }
 
             const auto& node_def = node_result.value();
-            if (!node_def.is_trigger || !node_def.is_scheduled) {
+            if (!node_def.is_trigger) {
+                continue;
+            }
+
+            // Told, not asked: the database reports the change, so there is no
+            // interval. Registered here as well as on activate, or a restart
+            // would leave every event-driven workflow deaf until somebody
+            // toggled it.
+            if (node_type == "database-change") {
+                auto config = smartbotic::common::applyConfigDefaults(
+                    node.value("config", nlohmann::json::object()), node_def.config_schema);
+                const std::string collection = config.value("collection", "");
+                if (collection.empty()) continue;
+
+                std::vector<std::string> event_types;
+                if (config.contains("eventTypes") && config["eventTypes"].is_array()) {
+                    for (const auto& t : config["eventTypes"]) {
+                        if (t.is_string()) event_types.push_back(t.get<std::string>());
+                    }
+                }
+                db_watcher_->watch(workflow_id, workflow_name, node_id, collection, event_types);
+                registered_count++;
+                continue;
+            }
+
+            if (!node_def.is_scheduled) {
                 continue;
             }
 
@@ -588,7 +623,8 @@ void WebServerService::runErrorWorkflow(const std::string& failed_workflow_id,
 
 void WebServerService::executeScheduledWorkflow(const std::string& workflow_id,
                                                  const std::string& trigger_node_id,
-                                                 const std::string& trigger_type) {
+                                                 const std::string& trigger_type,
+                                                 const nlohmann::json& extra_trigger_data) {
     // Select a runner
     auto runner = load_balancer_->selectRunner();
     if (!runner) {
@@ -612,6 +648,13 @@ void WebServerService::executeScheduledWorkflow(const std::string& workflow_id,
     nlohmann::json trigger_data;
     trigger_data["triggerNodeId"] = trigger_node_id;
     trigger_data["scheduledExecution"] = true;
+    // What actually happened, for the triggers that are told rather than the
+    // ones that ask - a database change carries the document with it.
+    if (extra_trigger_data.is_object()) {
+        for (const auto& [key, value] : extra_trigger_data.items()) {
+            trigger_data[key] = value;
+        }
+    }
     request.set_trigger_data(trigger_data.dump());
     request.set_wait_for_completion(false);
 

src/webserver/webserver_service.hpp

@@ -13,6 +13,7 @@
 #include "storage/storage_client.hpp"
 #include "config/config_loader.hpp"
 #include "scheduler/workflow_scheduler.hpp"
+#include "scheduler/database_watcher.hpp"
 
 namespace smartbotic::webserver::nodes {
     class NodeStore;
@@ -89,7 +90,8 @@ private:
     void ensureExecutionsSummaryView();
     void executeScheduledWorkflow(const std::string& workflow_id,
                                   const std::string& trigger_node_id,
-                                  const std::string& trigger_type);
+                                  const std::string& trigger_type,
+                                  const nlohmann::json& extra_trigger_data = nlohmann::json::object());
 
     // Runs the workflow a failing workflow nominates as its error handler, if
     // it names one. Called when an execution reports failure.
@@ -133,6 +135,7 @@ private:
 
     // Scheduler
     std::unique_ptr<WorkflowScheduler> scheduler_;
+    std::unique_ptr<DatabaseWatcher> db_watcher_;
 
     // Controllers (must outlive httplib::Server callbacks)
     std::unique_ptr<api::AuthController> auth_ctrl_;

tests/nodes/database-change.json

@@ -5,28 +5,18 @@
       "collections": { "watch_cursors": "read-write", "dbchange_test": "read-write" }
     }
   },
+  "_comment": "Database Change is a real trigger now - the database reports the change and the webserver starts the workflow - so it can only sit at the start of a flow, and the engine skips a trigger node found anywhere else. That is what this fixture covers: a run started by hand, which has no event to work from and falls back to asking what changed since last time. The event path itself needs a live subscription and a running webserver, so it is not reachable from here; it was verified by hand against insert, update and delete.",
   "nodes": [
-    {"id": "n1", "name": "Trigger", "type": "click-trigger", "position": {"x": 0, "y": 0}, "config": {}},
-    {"id": "setup", "name": "Setup", "type": "code", "position": {"x": 0, "y": 100},
-     "config": {"code": "smartbotic.storage.delete('watch_cursors', context.workflowId + ':first');\nsmartbotic.storage.delete('watch_cursors', context.workflowId + ':second');\nsmartbotic.storage.delete('dbchange_test', 'before-1');\nsmartbotic.storage.delete('dbchange_test', 'after-1');\nsmartbotic.storage.insert('dbchange_test', { name: 'before', at: Date.now() }, 'before-1');\nconst beforeDoc = smartbotic.storage.get('dbchange_test', 'before-1');\nconst raw = Number(beforeDoc.document._updated_at);\nconst stamp = raw > 1e15 ? Math.floor(raw / 1000000) : raw;\nsmartbotic.storage.insert('watch_cursors', { since: stamp, updatedAt: Date.now(), collection: 'dbchange_test' }, context.workflowId + ':second');\nreturn { ready: true };"}},
-    {"id": "first", "name": "First Run", "type": "database-change", "position": {"x": 0, "y": 200},
+    {"id": "first", "name": "Manual Run", "type": "database-change", "position": {"x": 0, "y": 0},
      "config": {"collection": "dbchange_test", "emitOnFirstRun": false}},
-    {"id": "add", "name": "Insert One", "type": "code", "position": {"x": 0, "y": 300},
-     "config": {"code": "smartbotic.utils.sleep(1100);\nsmartbotic.storage.insert('dbchange_test', { name: 'after', at: Date.now() }, 'after-1');\nreturn { inserted: 'after-1' };"}},
-    {"id": "second", "name": "Second Run", "type": "database-change", "position": {"x": 0, "y": 400},
-     "config": {"collection": "dbchange_test", "emitOnFirstRun": false}},
-    {"id": "cleanup", "name": "Cleanup", "type": "code", "position": {"x": 0, "y": 500},
-     "config": {"code": "smartbotic.storage.delete('dbchange_test', 'before-1');\nsmartbotic.storage.delete('dbchange_test', 'after-1');\nreturn { cleaned: true };"}}
+    {"id": "cleanup", "name": "Cleanup", "type": "code", "position": {"x": 0, "y": 100},
+     "config": {"code": "// The cursor is keyed by workflow and node, and the harness builds a fresh\n// workflow each run, so this run was always a first run. Clearing it anyway\n// keeps the collection from filling up with one cursor per test run.\nsmartbotic.storage.delete('watch_cursors', context.workflowId + ':first');\nreturn { cleaned: true };"}}
   ],
   "connections": [
-    {"sourceNodeId": "n1", "sourceOutput": "main", "targetNodeId": "setup", "targetInput": "data"},
-    {"sourceNodeId": "setup", "sourceOutput": "main", "targetNodeId": "first", "targetInput": "data"},
-    {"sourceNodeId": "first", "sourceOutput": "main", "targetNodeId": "add", "targetInput": "data"},
-    {"sourceNodeId": "add", "sourceOutput": "main", "targetNodeId": "second", "targetInput": "data"},
-    {"sourceNodeId": "second", "sourceOutput": "main", "targetNodeId": "cleanup", "targetInput": "data"}
+    {"sourceNodeId": "first", "sourceOutput": "main", "targetNodeId": "cleanup", "targetInput": "data"}
   ],
   "expect": {
     "first": {"status": "completed", "output": {"count": 0, "isFirstRun": true}},
-    "second": {"status": "completed", "output": {"count": 1, "isFirstRun": false, "documents": [{"name": "after"}]}}
+    "cleanup": {"status": "completed"}
   }
 }