bull-queue.html

<script type="text/html" data-template-name="bullmq-queue-server">
  <div class="form-row">
    <label for="node-config-input-name"><i class="fa fa-tasks"></i> Queue</label>
    <input type="text" id="node-config-input-name" placeholder="email-jobs">
  </div>
  <div class="form-row">
    <label for="node-config-input-backend"><i class="fa fa-database"></i> Backend</label>
    <select id="node-config-input-backend" style="width: 70%;">
      <option value="redis">Redis</option>
      <option value="postgres">PostgreSQL</option>
    </select>
  </div>
  <div class="form-row bull-redis-row">
    <label for="node-config-input-deployment"><i class="fa fa-server"></i> Deployment</label>
    <select id="node-config-input-deployment" style="width: 70%;">
      <option value="single">Standalone Redis</option>
      <option value="cluster">Redis Cluster / AWS MemoryDB</option>
      <option value="sentinel">Redis Sentinel</option>
    </select>
  </div>
  <div class="form-row bull-single-row">
    <label for="node-config-input-address"><i class="fa fa-link"></i> Host</label>
    <input type="text" id="node-config-input-address" placeholder="localhost">
  </div>
  <div class="form-row bull-single-row">
    <label for="node-config-input-port"><i class="fa fa-plug"></i> Port</label>
    <input type="number" id="node-config-input-port" placeholder="6379">
  </div>
  <div class="form-row bull-cluster-row">
    <label for="node-config-input-clusterNodes"><i class="fa fa-sitemap"></i> Cluster Nodes</label>
    <textarea id="node-config-input-clusterNodes" rows="3" style="width: 70%;" placeholder="host-a:6379, host-b:6379"></textarea>
  </div>
  <div class="form-row bull-sentinel-row">
    <label for="node-config-input-sentinels"><i class="fa fa-sitemap"></i> Sentinels</label>
    <textarea id="node-config-input-sentinels" rows="3" style="width: 70%;" placeholder="sentinel-a:26379, sentinel-b:26379"></textarea>
  </div>
  <div class="form-row bull-sentinel-row">
    <label for="node-config-input-sentinelMasterName"><i class="fa fa-tag"></i> Master</label>
    <input type="text" id="node-config-input-sentinelMasterName" placeholder="mymaster">
  </div>
  <div class="form-row bull-db-row">
    <label for="node-config-input-db"><i class="fa fa-database"></i> Database</label>
    <input type="number" id="node-config-input-db" placeholder="0">
  </div>
  <div class="form-row bull-postgres-row">
    <label for="node-config-input-database"><i class="fa fa-database"></i> Database Name</label>
    <input type="text" id="node-config-input-database" placeholder="bullmq">
  </div>
  <div class="form-row bull-postgres-row">
    <label for="node-config-input-schema"><i class="fa fa-folder-o"></i> Schema</label>
    <input type="text" id="node-config-input-schema" placeholder="bullmq">
  </div>
  <div class="form-row bull-postgres-row">
    <label for="node-config-input-max"><i class="fa fa-database"></i> Pool Max</label>
    <input type="number" id="node-config-input-max" min="1" step="1" placeholder="2">
  </div>
  <div class="form-row bull-postgres-row">
    <label for="node-config-input-migrate"><i class="fa fa-wrench"></i> Migrations</label>
    <input type="checkbox" id="node-config-input-migrate" style="display:inline-block; width:auto; vertical-align:top;">
    <span>Create and update the BullMQ schema on connect</span>
  </div>
  <div class="form-row">
    <label for="node-config-input-username"><i class="fa fa-user"></i> Username</label>
    <input type="text" id="node-config-input-username" placeholder="user">
  </div>
  <div class="form-row">
    <label for="node-config-input-password"><i class="fa fa-lock"></i> Password</label>
    <input type="password" id="node-config-input-password">
  </div>
  <div class="form-row bull-sentinel-row">
    <label for="node-config-input-sentinelUsername"><i class="fa fa-user"></i> Sentinel User</label>
    <input type="text" id="node-config-input-sentinelUsername">
  </div>
  <div class="form-row bull-sentinel-row">
    <label for="node-config-input-sentinelPassword"><i class="fa fa-lock"></i> Sentinel Pass</label>
    <input type="password" id="node-config-input-sentinelPassword">
  </div>
  <div class="form-row">
    <label for="node-config-input-tls"><i class="fa fa-shield"></i> TLS</label>
    <input type="checkbox" id="node-config-input-tls" style="display:inline-block; width:auto; vertical-align:top;">
    <span>Use TLS for data connections</span>
  </div>
  <div class="form-row bull-sentinel-row">
    <label for="node-config-input-sentinelTls"><i class="fa fa-shield"></i> Sentinel TLS</label>
    <input type="checkbox" id="node-config-input-sentinelTls" style="display:inline-block; width:auto; vertical-align:top;">
    <span>Use TLS for Sentinel discovery</span>
  </div>
  <div class="form-row bull-tls-row">
    <label for="node-config-input-tlsRejectUnauthorized"><i class="fa fa-check"></i> Verify TLS</label>
    <input type="checkbox" id="node-config-input-tlsRejectUnauthorized" style="display:inline-block; width:auto; vertical-align:top;">
    <span>Reject unauthorized certificates</span>
  </div>
  <div class="form-row bull-tls-row">
    <label for="node-config-input-tlsServerName"><i class="fa fa-globe"></i> TLS Server Name</label>
    <input type="text" id="node-config-input-tlsServerName" placeholder="server.example.com">
  </div>
  <div class="form-row bull-tls-row">
    <label for="node-config-input-tlsCa"><i class="fa fa-certificate"></i> CA</label>
    <textarea id="node-config-input-tlsCa" rows="3" style="width: 70%;"></textarea>
  </div>
  <div class="form-row bull-tls-row">
    <label for="node-config-input-tlsCert"><i class="fa fa-certificate"></i> Client Cert</label>
    <textarea id="node-config-input-tlsCert" rows="3" style="width: 70%;"></textarea>
  </div>
  <div class="form-row bull-tls-row">
    <label for="node-config-input-tlsKey"><i class="fa fa-key"></i> Client Key</label>
    <textarea id="node-config-input-tlsKey" rows="3" style="width: 70%;"></textarea>
  </div>
  <div class="form-row bull-redis-row">
    <label for="node-config-input-prefix"><i class="fa fa-code"></i> Prefix</label>
    <input type="text" id="node-config-input-prefix" placeholder="{bull}">
  </div>
  <div class="form-row">
    <label for="node-config-input-removeOnComplete"><i class="fa fa-trash"></i> Completed</label>
    <input type="number" id="node-config-input-removeOnComplete" min="0" step="1" placeholder="1000">
  </div>
  <div class="form-row">
    <label for="node-config-input-removeOnFail"><i class="fa fa-trash-o"></i> Failed</label>
    <input type="number" id="node-config-input-removeOnFail" min="0" step="1" placeholder="5000">
  </div>
  <div class="form-row">
    <label for="node-config-input-telemetry"><i class="fa fa-line-chart"></i> Telemetry</label>
    <input type="checkbox" id="node-config-input-telemetry" style="display:inline-block; width:auto; vertical-align:top;">
    <span>Enable OpenTelemetry tracing via bullmq-otel</span>
  </div>
  <div class="form-row bull-telemetry-row">
    <label for="node-config-input-telemetryServiceName"><i class="fa fa-tag"></i> Service Name</label>
    <input type="text" id="node-config-input-telemetryServiceName" placeholder="queue name">
  </div>
  <div class="form-row bull-telemetry-row">
    <label for="node-config-input-telemetryMetrics"><i class="fa fa-bar-chart"></i> Metrics</label>
    <input type="checkbox" id="node-config-input-telemetryMetrics" style="display:inline-block; width:auto; vertical-align:top;">
    <span>Enable OpenTelemetry metrics</span>
  </div>
</script>

<script type="text/javascript">
  (function() {
    function validateKeepCount(value) {
      return value === "" || /^\d+$/.test(value);
    }

    // Blank is not zero: blank means "use the default pool size", while 0 is a
    // pool that can never hand out a connection, which the runtime rejects.
    // Undefined counts as blank so a saved flow with no max never shows an
    // error before the dialog has been opened once.
    function validatePoolMax(value) {
      if (value === "" || value === undefined || value === null) {
        return true;
      }
      return /^\d+$/.test(String(value)) && Number(value) >= 1;
    }

    function isPostgresBackend() {
      return $("#node-config-input-backend").val() === "postgres";
    }

    function updateBullQueueServerRows() {
      var postgres = isPostgresBackend();
      var deployment = $("#node-config-input-deployment").val() || "single";
      $(".bull-redis-row").toggle(!postgres);
      $(".bull-postgres-row").toggle(postgres);
      // Host and port are shared, so PostgreSQL always shows them. The cluster,
      // Sentinel, and Redis-database rows describe Redis topologies only.
      $(".bull-single-row").toggle(postgres || deployment === "single");
      $(".bull-cluster-row").toggle(!postgres && deployment === "cluster");
      $(".bull-sentinel-row").toggle(!postgres && deployment === "sentinel");
      $(".bull-db-row").toggle(!postgres && deployment !== "cluster");
      $(".bull-tls-row").toggle($("#node-config-input-tls").is(":checked"));
      $(".bull-telemetry-row").toggle($("#node-config-input-telemetry").is(":checked"));
    }

    RED.nodes.registerType("bullmq-queue-server", {
      category: "config",
      defaults: {
        name: { value: "", required: true },
        backend: { value: "redis", required: true },
        deployment: { value: "single", required: true },
        address: { value: "localhost" },
        port: { value: 6379, validate: RED.validators.number() },
        db: { value: "" },
        database: { value: "" },
        schema: { value: "" },
        max: {
          value: "",
          // Node-RED validates every default, visible or not, and calls this
          // with the node as `this`. A Redis config must not be flagged
          // invalid by the PostgreSQL pool size: that row is hidden on Redis,
          // so the dialog offers no way to correct it. lib/connections.js
          // draws the same line, validating only the selected backend's
          // fields. A flow saved before the selector existed has no backend
          // property and is Redis, so it is exempt too.
          validate: function(value) {
            return this.backend !== "postgres" || validatePoolMax(value);
          }
        },
        migrate: { value: true },
        clusterNodes: { value: "" },
        sentinels: { value: "" },
        sentinelMasterName: { value: "" },
        username: { value: "" },
        sentinelUsername: { value: "" },
        tls: { value: false },
        sentinelTls: { value: false },
        tlsRejectUnauthorized: { value: true },
        tlsServerName: { value: "" },
        prefix: { value: "" },
        removeOnComplete: { value: "1000", validate: validateKeepCount },
        removeOnFail: { value: "5000", validate: validateKeepCount },
        telemetry: { value: false },
        telemetryServiceName: { value: "" },
        telemetryMetrics: { value: false }
      },
      credentials: {
        password: { type: "password" },
        sentinelPassword: { type: "password" },
        tlsCa: { type: "password" },
        tlsCert: { type: "password" },
        tlsKey: { type: "password" }
      },
      label: function() {
        return this.name || this.address + ":" + this.port;
      },
      oneditprepare: function() {
        // A flow saved before the backend selector existed has no backend or
        // migrate property, and editor defaults never migrate saved JSON: the
        // inputs are populated from the saved node, so both would read as
        // "unset" here -- an empty select, and a migrations box unchecked when
        // the runtime default is on. Seed them to the runtime's own defaults
        // before anything reads them.
        if (!this.backend) {
          $("#node-config-input-backend").val("redis");
        }
        if (this.migrate === undefined) {
          $("#node-config-input-migrate").prop("checked", true);
        }
        // One saved port field serves both backends. Keep one value per mode
        // while the dialog is open so even a deliberate Redis port of 5432
        // survives a round-trip through PostgreSQL.
        var backendField = $("#node-config-input-backend");
        var portField = $("#node-config-input-port");
        var activeBackend = isPostgresBackend() ? "postgres" : "redis";
        var portValues = {
          redis: activeBackend === "redis" ? String(portField.val()) : "6379",
          postgres: activeBackend === "postgres" ? String(portField.val()) : "5432"
        };
        backendField.on("change", function() {
          var nextBackend = isPostgresBackend() ? "postgres" : "redis";
          if (nextBackend !== activeBackend) {
            portValues[activeBackend] = String(portField.val());
            portField.val(portValues[nextBackend]);
            activeBackend = nextBackend;
          }
          updateBullQueueServerRows();
        });
        $("#node-config-input-deployment").on("change", updateBullQueueServerRows);
        $("#node-config-input-tls").on("change", updateBullQueueServerRows);
        $("#node-config-input-telemetry").on("change", updateBullQueueServerRows);
        updateBullQueueServerRows();
      }
    });
  })();
</script>

<script type="text/html" data-help-name="bullmq-queue-server">
  <p>Configures the BullMQ queue name and the backend connection shared by the BullMQ nodes. BullMQ v6 can store queues in either Redis or PostgreSQL; pick one per config node with <b>Backend</b>.</p>
  <p>PostgreSQL requires the <code>pg</code> package, which is an optional peer dependency and must be installed separately (<code>npm install pg</code>). It is reported once per config node on first use, not at load.</p>
  <p>API reference: <a href="https://docs.bullmq.io/api/classes/v6.Queue.html" target="_blank" rel="noopener noreferrer">Queue</a> and <a href="https://docs.bullmq.io/api/interfaces/v6.QueueOptions.html" target="_blank" rel="noopener noreferrer">QueueOptions</a>.</p>

  <h3>Options</h3>
  <ul>
    <li><b>Queue</b>: BullMQ queue name. All producer, worker, event, and flow nodes that select this config use this queue name unless a BullMQ flow child specifies another <code>queueName</code>.</li>
    <li><b>Backend</b>: Where BullMQ stores this queue. <code>Redis</code> keeps the Redis topology, key prefix, and Redis database options below. <code>PostgreSQL</code> replaces them with the database options and needs <code>pg</code> installed.</li>
    <li><b>Deployment</b>: Redis topology. Use <code>Standalone Redis</code> for one Redis endpoint, <code>Redis Cluster / AWS MemoryDB</code> for cluster-compatible endpoints, or <code>Redis Sentinel</code> for Sentinel discovery.</li>
    <li><b>Host</b>: Standalone Redis host, or the PostgreSQL server host name or IP address.</li>
    <li><b>Port</b>: Standalone Redis port, or the PostgreSQL port. The defaults are <code>6379</code> for Redis and <code>5432</code> for PostgreSQL. While the dialog is open, switching <b>Backend</b> preserves a separate value for each backend so a custom port is never overwritten.</li>
    <li><b>Cluster Nodes</b>: Comma or newline separated cluster seed nodes such as <code>redis-a:6379, redis-b:6379</code>. AWS MemoryDB cluster configuration endpoints are entered here.</li>
    <li><b>Sentinels</b>: Comma or newline separated Sentinel endpoints such as <code>sentinel-a:26379, sentinel-b:26379</code>.</li>
    <li><b>Master</b>: Sentinel master name, for example <code>mymaster</code>.</li>
    <li><b>Database</b>: Redis database number for standalone or Sentinel data connections. Cluster and MemoryDB deployments use Redis database <code>0</code>.</li>
    <li><b>Database Name</b>: PostgreSQL database that holds the BullMQ schema. Distinct from the Redis <b>Database</b> number above, which is an index rather than a name. Left blank, node-postgres falls back to the <code>PGDATABASE</code> environment variable and then to the operating-system user name, as it does for a blank <b>Username</b> with <code>PGUSER</code>.</li>
    <li><b>Schema</b>: PostgreSQL schema for BullMQ's tables. Defaults to <code>bullmq</code>. Separate schemas keep independent queue sets in one database.</li>
    <li><b>Pool Max</b>: Maximum PostgreSQL connections in each pool. Blank uses the default of <code>2</code>; <code>0</code> is rejected, since a pool that can never hand out a connection cannot run a queue. BullMQ builds one pool per queue, worker, event listener, and flow producer; workers and event listeners also hold one extra <code>LISTEN</code> connection each. A config node feeding all four therefore costs up to 4 &times; <b>Pool Max</b> + 2 server connections &mdash; 10 at the default. Start with the default and raise it only when observed pool waits justify the extra server connections; worker <b>Concurrency</b> does not require a one-to-one pool size.</li>
    <li><b>Migrations</b>: Creates and updates BullMQ's PostgreSQL schema when connecting. On by default. Turn it off where the database user is not allowed to change the schema, and apply BullMQ's migrations separately -- a queue pointed at a database whose schema is missing or outdated reports that as an error rather than running.</li>
    <li><b>Username</b>: Redis ACL username, or the PostgreSQL user. Leave empty to use the client's backend default; this is also correct for Redis servers that only use a password.</li>
    <li><b>Password</b>: Password for data connections, for either backend. Stored as a Node-RED credential.</li>
    <li><b>Sentinel User</b>: Redis ACL username for Sentinel discovery connections when Sentinels require separate credentials.</li>
    <li><b>Sentinel Pass</b>: Password for Sentinel discovery connections. Stored as a Node-RED credential.</li>
    <li><b>TLS</b>: Enables TLS for data connections on either backend. Use this for AWS MemoryDB, any Redis deployment that requires encrypted data connections, and PostgreSQL servers that require SSL. <b>Verify TLS</b>, <b>TLS Server Name</b>, <b>CA</b>, <b>Client Cert</b>, and <b>Client Key</b> apply to both.</li>
    <li><b>Sentinel TLS</b>: Enables TLS for Sentinel discovery connections.</li>
    <li><b>Verify TLS</b>: Rejects certificates that cannot be verified. Keep enabled in production unless you intentionally use a private test certificate.</li>
    <li><b>TLS Server Name</b>: Optional server name used for certificate validation when it differs from the host value. For PostgreSQL, node-postgres uses the connection host when it is a hostname and honors this override only when <b>Host</b> is a literal IP address.</li>
    <li><b>CA</b>: PEM certificate authority text for validating private backend certificates. Stored as a Node-RED credential.</li>
    <li><b>Client Cert</b>: PEM client certificate for mutual TLS deployments. Stored as a Node-RED credential.</li>
    <li><b>Client Key</b>: PEM client private key for mutual TLS deployments. Stored as a Node-RED credential.</li>
    <li><b>Prefix</b>: BullMQ key prefix. Redis Cluster and MemoryDB prefixes must contain a hash tag such as <code>{bull}</code>. Independent queues may use different tags, but prefixes in one flow tree or bulk flow batch must contain the same hash tag for atomic operations. Each worker must use the exact prefix assigned to its queue in that flow.</li>
    <li><b>Keep Completed</b> / <b>Keep Failed</b>: Queue-level auto-removal, applied as BullMQ <code>defaultJobOptions</code>. Each keeps that many of the newest jobs in the state and removes older ones. Leaving a field blank keeps every job, which is BullMQ's own default and grows backend storage without bound, so blank is only appropriate when something else prunes the queue. <code>msg.jobopts</code> overrides these defaults for <code>bullmq cmd</code>. For <code>bullmq flow</code>, per-queue <code>msg.flowopts.queuesOptions</code> defaults and then each job's <code>opts</code> take precedence.</li>
    <li><b>Telemetry</b>: Enables OpenTelemetry tracing for this queue's <code>Queue</code>, <code>Worker</code>, and <code>FlowProducer</code> instances using <a href="https://www.npmjs.com/package/bullmq-otel" target="_blank" rel="noopener noreferrer">bullmq-otel</a>. Off by default. <code>bullmq-otel</code> is an optional dependency and must be installed separately (<code>npm install bullmq-otel</code>) before enabling this switch. This node does not start or configure an OpenTelemetry SDK or exporter; the host Node-RED process owns that (for example NODE_OPTIONS auto-instrumentation, or an SDK bootstrapped from Node-RED's <code>settings.js</code>). When no OpenTelemetry provider is registered, <code>bullmq-otel</code> simply no-ops rather than erroring.</li>
    <li><b>Service Name</b>: Tracer and meter name reported to OpenTelemetry. Leave blank to use the <b>Queue</b> name.</li>
    <li><b>Metrics</b>: Enables OpenTelemetry metrics in addition to tracing. A <code>MeterProvider</code> must be registered with the host process before this queue is first used, otherwise metrics are dropped. For metrics scraping without any OpenTelemetry stack, the <code>bullmq cmd</code> node's <code>msg.cmd</code> <code>exportPrometheusMetrics</code> is a zero-dependency alternative.</li>
  </ul>

  <h3>Example</h3>
  <p>For local Redis, set <b>Queue</b> to <code>email-jobs</code>, <b>Deployment</b> to <code>Standalone Redis</code>, <b>Host</b> to <code>localhost</code>, and <b>Port</b> to <code>6379</code>.</p>
  <p>For local PostgreSQL, set <b>Backend</b> to <code>PostgreSQL</code>, <b>Host</b> to <code>localhost</code>, <b>Port</b> to <code>5432</code>, <b>Database Name</b> to <code>bullmq</code>, and leave <b>Schema</b>, <b>Pool Max</b>, and <b>Migrations</b> at their defaults. <b>Deployment</b>, <b>Prefix</b>, and the Redis <b>Database</b> number do not apply.</p>
  <p>For AWS MemoryDB, set <b>Deployment</b> to <code>Redis Cluster / AWS MemoryDB</code>, enter the cluster endpoint in <b>Cluster Nodes</b>, enable <b>TLS</b>, keep <b>Verify TLS</b> enabled, and use <code>{bull}</code> as <b>Prefix</b>.</p>
  <p>Redis must use <code>maxmemory-policy=noeviction</code> and durable persistence; BullMQ recommends Append Only File (AOF) for self-managed Redis. Job data is stored in clear text, so avoid sensitive payloads or encrypt sensitive fields before adding a job. Bull v4 Redis queue data is not automatically migrated to BullMQ.</p>
</script>

<script type="text/html" data-template-name="bullmq cmd">
  <div class="form-row">
    <label for="node-input-name"><i class="fa fa-tag"></i> Name</label>
    <input type="text" id="node-input-name">
  </div>
  <div class="form-row">
    <label for="node-input-queue"><i class="fa fa-tasks"></i> Queue</label>
    <input type="text" id="node-input-queue">
  </div>
</script>

<script type="text/html" data-help-name="bullmq cmd">
  <p>Executes BullMQ producer and queue-administration commands from an input message.</p>
  <p>API reference: <a href="https://docs.bullmq.io/api/classes/v6.Queue.html" target="_blank" rel="noopener noreferrer">Queue</a>, <a href="https://docs.bullmq.io/api/classes/v6.Queue.html#add" target="_blank" rel="noopener noreferrer">Queue.add</a>, <a href="https://docs.bullmq.io/api/types/v6.JobsOptions.html" target="_blank" rel="noopener noreferrer">JobsOptions</a>, <a href="https://docs.bullmq.io/api/types/v6.DeduplicationOptions.html" target="_blank" rel="noopener noreferrer">DeduplicationOptions</a>, <a href="https://docs.bullmq.io/api/classes/v6.Queue.html#setglobalratelimit" target="_blank" rel="noopener noreferrer">Queue.setGlobalRateLimit</a>, and <a href="https://docs.bullmq.io/api/classes/v6.Queue.html#upsertjobscheduler" target="_blank" rel="noopener noreferrer">Queue.upsertJobScheduler</a>.</p>

  <h3>Options</h3>
  <ul>
    <li><b>Name</b>: Optional label shown in the Node-RED workspace.</li>
    <li><b>Queue</b>: Queue config node used to create the BullMQ <code>Queue</code>.</li>
  </ul>

  <h3>Input message</h3>
  <ul>
    <li><code>msg.cmd</code>: Command name. Defaults to <code>add</code>. Common commands include <code>add</code>, <code>addBulk</code>, <code>getJob</code>, <code>getJobs</code>, <code>removeJob</code>, <code>retryJob</code>, <code>getDelayed</code>, <code>promoteJob</code>, <code>getPrioritized</code>, <code>changePriority</code>, <code>setGlobalRateLimit</code>, <code>removeGlobalRateLimit</code>, <code>upsertJobScheduler</code>, <code>removeJobScheduler</code>, <code>isPaused</code>, <code>isMaxed</code>, and <code>getVersion</code>.</li>
    <li><code>msg.payload</code>: Job data for <code>add</code> when <code>msg.jobData</code> is not set. For some commands it carries command parameters, such as <code>{ "max": 2, "duration": 1000 }</code> for <code>setGlobalRateLimit</code>. <code>isPaused</code> and <code>isMaxed</code> return a boolean; <code>getVersion</code> returns BullMQ's queue metadata version string (for example <code>bullmq:6.3.1</code>), not the Redis server version.</li>
    <li><code>msg.jobName</code>: BullMQ job name for <code>add</code>. Defaults to <code>default</code>.</li>
    <li><code>msg.jobData</code>: Full job data object for <code>add</code>. When set, it takes precedence over <code>msg.payload</code>.</li>
    <li><code>msg.jobopts</code>: BullMQ job options for <code>add</code>, including <code>delay</code>, <code>priority</code>, <code>deduplication</code>, <code>attempts</code>, <code>backoff</code>, and <code>removeOnComplete</code>. Repeat options are rejected; use <code>upsertJobScheduler</code>.</li>
  </ul>

  <h3>Examples</h3>
  <p>Add a delayed, prioritized job:</p>
  <pre>msg.cmd = "add";
msg.jobName = "email";
msg.payload = { to: "user@example.com" };
msg.jobopts = {
  delay: 10000,
  priority: 1,
  removeOnComplete: true
};
return msg;</pre>
  <p>Add three one-off jobs that become eligible after 10, 20, and 30 seconds:</p>
  <pre>msg.cmd = "addBulk";
msg.payload = [
  { name: "delayed-1", data: { payload: "first" }, opts: { delay: 10000 } },
  { name: "delayed-2", data: { payload: "second" }, opts: { delay: 20000 } },
  { name: "delayed-3", data: { payload: "third" }, opts: { delay: 30000 } }
];
return msg;</pre>
  <p>Add one job per deduplication id:</p>
  <pre>msg.cmd = "add";
msg.jobName = "report";
msg.payload = { customerId: "customer-42" };
msg.jobopts = {
  deduplication: { id: "customer-42-report" }
};
return msg;</pre>
  <p>Set a global rate limit of two jobs per second:</p>
  <pre>msg.cmd = "setGlobalRateLimit";
msg.payload = { max: 2, duration: 1000 };
return msg;</pre>
  <p>Add or update a scheduler every minute:</p>
  <pre>msg.cmd = "upsertJobScheduler";
msg.schedulerId = "heartbeat-every-minute";
msg.repeat = { pattern: "*/1 * * * *" };
msg.template = {
  name: "heartbeat",
  data: { payload: "scheduled heartbeat" }
};
return msg;</pre>
  <p>Job Scheduler commands use the native BullMQ v6 fields: <code>msg.schedulerId</code>, <code>msg.repeat.pattern</code>, optional <code>msg.repeat.tz</code>, and <code>msg.template</code>.</p>
  <p>BullMQ v6 removed the <code>paused</code> job state, so <code>getJobCounts</code> no longer includes a <code>paused</code> key in its result.</p>
</script>

<script type="text/javascript">
  RED.nodes.registerType("bullmq cmd", {
    color: "#ffffff",
    category: "function",
    defaults: {
      name: { value: "" },
      queue: { type: "bullmq-queue-server", required: true }
    },
    inputs: 1,
    outputs: 1,
    align: "right",
    icon: "bull_icon.png",
    label: function() {
      return this.name || "bullmq cmd";
    }
  });
</script>

<script type="text/html" data-template-name="bullmq run">
  <div class="form-row">
    <label for="node-input-name"><i class="fa fa-tag"></i> Name</label>
    <input type="text" id="node-input-name">
  </div>
  <div class="form-row">
    <label for="node-input-queue"><i class="fa fa-tasks"></i> Queue</label>
    <input type="text" id="node-input-queue">
  </div>
  <div class="form-row">
    <label for="node-input-completionMode"><i class="fa fa-check-circle"></i> Completion</label>
    <select id="node-input-completionMode" style="width: 70%;">
      <option value="immediate">Immediate</option>
      <option value="manual">Manual acknowledgement</option>
    </select>
  </div>
  <div class="form-row bull-ack-timeout-row">
    <label for="node-input-ackTimeout"><i class="fa fa-clock-o"></i> Ack Timeout</label>
    <input type="number" id="node-input-ackTimeout" placeholder="300000">
  </div>
  <div class="form-row">
    <label for="node-input-concurrency"><i class="fa fa-random"></i> Concurrency</label>
    <input type="number" id="node-input-concurrency" min="1" step="1" placeholder="1">
  </div>
  <div class="form-row">
    <label for="node-input-maxStartedAttempts"><i class="fa fa-repeat"></i> Max Started Attempts</label>
    <input type="number" id="node-input-maxStartedAttempts" min="1" step="1" placeholder="100">
  </div>
  <div class="form-row">
    <label for="node-input-limiterMax"><i class="fa fa-tachometer"></i> Limiter Max</label>
    <input type="number" id="node-input-limiterMax">
  </div>
  <div class="form-row">
    <label for="node-input-limiterDuration"><i class="fa fa-hourglass-half"></i> Limiter Duration</label>
    <input type="number" id="node-input-limiterDuration">
  </div>
</script>

<script type="text/html" data-help-name="bullmq run">
  <p>Creates a BullMQ Worker for the configured queue and emits each job as a Node-RED message.</p>
  <p>API reference: <a href="https://docs.bullmq.io/api/classes/v6.Worker.html" target="_blank" rel="noopener noreferrer">Worker</a> and <a href="https://docs.bullmq.io/api/interfaces/v6.WorkerOptions.html" target="_blank" rel="noopener noreferrer">WorkerOptions</a>.</p>

  <h3>Options</h3>
  <ul>
    <li><b>Name</b>: Optional label shown in the Node-RED workspace.</li>
    <li><b>Queue</b>: Queue config node used by the worker.</li>
    <li><b>Completion</b>: <code>Immediate</code> completes the BullMQ job after the message is emitted. <code>Manual acknowledgement</code> keeps the job active until a downstream <code>bullmq job</code> node completes, fails, or updates it.</li>
    <li><b>Ack Timeout</b>: Maximum time in milliseconds to wait for manual acknowledgement before the node fails the active job. Set to <code>0</code> to wait indefinitely.</li>
    <li><b>Concurrency</b>: Number of jobs this worker can process at the same time.</li>
    <li><b>Max Started Attempts</b>: Maximum times one job may start processing. Defaults to <code>100</code>. BullMQ then fails the job unrecoverably, which bounds loops created by repeated <code>moveToWait</code> or <code>moveToDelayed</code> actions.</li>
    <li><b>Limiter Max</b>: Optional worker-side rate limit count.</li>
    <li><b>Limiter Duration</b>: Optional worker-side rate limit window in milliseconds. Set with <b>Limiter Max</b>.</li>
  </ul>

  <h3>Output message</h3>
  <ul>
    <li><code>msg.payload</code>: <code>job.data.payload</code> when present, otherwise the full BullMQ job data.</li>
    <li><code>msg.job</code>: Serialized job metadata, including id, name, queue name, state-related timestamps, progress, attempts, and options.</li>
    <li><code>msg.bull</code>: Queue and acknowledgement context used by downstream <code>bullmq job</code> nodes.</li>
  </ul>

  <p>BullMQ v6 removed the <code>paused</code> job state: <code>job.getState()</code> never returns <code>"paused"</code>, so a job on a paused queue reports as <code>"waiting"</code>.</p>
  <p>Only jobs from a <b>Completion</b> <code>Manual acknowledgement</code> worker can be cancelled through <code>bullmq job</code>'s <code>cancelJob</code> and <code>cancelAllJobs</code> actions, since cancellation is scoped to <code>msg.bull.ackId</code>.</p>

  <h3>Example</h3>
  <p>For fire-and-forget processing, use <b>Completion</b> <code>Immediate</code> and wire the output to your processing flow.</p>
  <p>For a flow that must decide success or failure, use <b>Completion</b> <code>Manual acknowledgement</code>, wire the output to processing nodes, then wire into <code>bullmq job</code> with <b>Action</b> <code>complete</code> or <code>fail</code>.</p>
</script>

<script type="text/javascript">
  (function() {
    function updateBullRunRows() {
      $(".bull-ack-timeout-row").toggle(
        $("#node-input-completionMode").val() === "manual"
      );
    }

    function positiveInteger(value) {
      var parsed = Number(value);
      return value !== "" && Number.isInteger(parsed) && parsed > 0;
    }

    function optionalPositivePair(value, otherProperty) {
      var input = $("#node-input-" + otherProperty);
      var other = input.length ? input.val() : this[otherProperty];
      return (
        ((value === "" || value == null) && (other === "" || other == null)) ||
        (positiveInteger(value) && positiveInteger(other))
      );
    }

    RED.nodes.registerType("bullmq run", {
      color: "#ffffff",
      category: "function",
      defaults: {
        name: { value: "" },
        queue: { type: "bullmq-queue-server", required: true },
        completionMode: { value: "immediate" },
        ackTimeout: { value: 300000, validate: RED.validators.number() },
        concurrency: { value: 1, validate: positiveInteger },
        maxStartedAttempts: { value: 100, validate: positiveInteger },
        limiterMax: {
          value: "",
          validate: function(value) {
            return optionalPositivePair.call(this, value, "limiterDuration");
          }
        },
        limiterDuration: {
          value: "",
          validate: function(value) {
            return optionalPositivePair.call(this, value, "limiterMax");
          }
        }
      },
      inputs: 0,
      outputs: 1,
      align: "left",
      icon: "bull_icon.png",
      label: function() {
        return this.name || "bullmq run";
      },
      oneditprepare: function() {
        $("#node-input-completionMode").on("change", updateBullRunRows);
        updateBullRunRows();
      }
    });
  })();
</script>

<script type="text/html" data-template-name="bullmq job">
  <div class="form-row">
    <label for="node-input-name"><i class="fa fa-tag"></i> Name</label>
    <input type="text" id="node-input-name">
  </div>
  <div class="form-row">
    <label for="node-input-action"><i class="fa fa-check"></i> Action</label>
    <select id="node-input-action" style="width: 70%;">
      <option value="complete">Complete</option>
      <option value="fail">Fail</option>
      <option value="failUnrecoverable">Fail unrecoverable</option>
      <option value="progress">Progress</option>
      <option value="rateLimit">Rate limit</option>
      <option value="removeDeduplicationKey">Remove deduplication key</option>
      <option value="getChildrenValues">Get children values</option>
      <option value="getFailedChildrenValues">Get failed children values</option>
      <option value="removeUnprocessedChildren">Remove unprocessed children</option>
      <option value="cancelJob">Cancel job</option>
      <option value="cancelAllJobs">Cancel all jobs</option>
      <option value="moveToWait">Requeue (moveToWait)</option>
      <option value="moveToDelayed">Delay and resume (moveToDelayed)</option>
      <option value="updateData">Update job data</option>
    </select>
  </div>
</script>

<script type="text/html" data-help-name="bullmq job">
  <p>Acts on an active job emitted by a manual-mode <code>bullmq run</code> node.</p>
  <p>API reference: <a href="https://docs.bullmq.io/api/classes/v6.Job.html" target="_blank" rel="noopener noreferrer">Job</a>, <a href="https://docs.bullmq.io/api/classes/v6.UnrecoverableError.html" target="_blank" rel="noopener noreferrer">UnrecoverableError</a>, and <a href="https://docs.bullmq.io/api/classes/v6.Worker.html#canceljob" target="_blank" rel="noopener noreferrer">Worker.cancelJob</a>.</p>

  <h3>Options</h3>
  <ul>
    <li><b>Name</b>: Optional label shown in the Node-RED workspace.</li>
    <li><b>Action</b>: Default job action. It can be overridden per message with <code>msg.cmd</code>.</li>
  </ul>

  <h3>Actions</h3>
  <ul>
    <li><code>complete</code>: Completes the active job. Use <code>msg.payload</code> as the return value.</li>
    <li><code>fail</code>: Fails the active job. Use <code>msg.error</code> or <code>msg.payload</code> to describe the error.</li>
    <li><code>failUnrecoverable</code>: Fails the active job with a BullMQ unrecoverable error so retry attempts are not used.</li>
    <li><code>progress</code>: Updates job progress from <code>msg.progress</code> or <code>msg.payload</code>.</li>
    <li><code>rateLimit</code>: Applies a worker rate limit delay. Use <code>msg.duration</code> as milliseconds.</li>
    <li><code>removeDeduplicationKey</code>: Removes the active job deduplication key.</li>
    <li><code>getChildrenValues</code>: Reads completed child return values for a flow parent job.</li>
    <li><code>getFailedChildrenValues</code>: Reads failed child values for a flow parent job.</li>
    <li><code>removeUnprocessedChildren</code>: Removes children that have not yet been processed.</li>
    <li><code>cancelJob</code>: Cancels the active job by aborting its processor signal, which fails the job and lets the queue's normal attempts/backoff retry policy apply. Acknowledgement-scoped: it needs <code>msg.bull.ackId</code>, so it only reaches manual-completion jobs that have not yet settled. <code>msg.reason</code> sets the cancellation reason, default <code>"BullMQ job cancelled"</code>. Outputs <code>true</code> on <code>msg.payload</code>; if BullMQ reports no cancellable processor for the job, the node raises an error instead.</li>
    <li><code>cancelAllJobs</code>: Cancels every active manual job on the same <code>bullmq run</code> node the same way as <code>cancelJob</code>. <code>msg.reason</code> sets the cancellation reason, default <code>"BullMQ job cancelled"</code>. Outputs <code>true</code> on <code>msg.payload</code>.</li>
    <li><code>moveToWait</code>: Puts the active job back in the waiting state to be picked up again, without counting a failed attempt. Terminal for this acknowledgement.</li>
    <li><code>moveToDelayed</code>: Delays the active job by <code>msg.delay</code> milliseconds and resumes it later, without counting a failed attempt. Terminal for this acknowledgement. Combine with <code>updateData</code> to resume from a recorded step.</li>
    <li><code>updateData</code>: Replaces the job's stored data with <code>msg.jobData</code>, or <code>msg.payload</code> when that is absent, and survives a retry. Non-terminal, so the flow still has to settle the acknowledgement. Use it to record which step a job reached before <code>moveToDelayed</code>.</li>
  </ul>

  <h3>Example</h3>
  <p>Update progress, then complete a manual job:</p>
  <pre>msg.cmd = "progress";
msg.progress = 50;
return msg;</pre>
  <p>To use another <code>bullmq job</code> node with configured <b>Action</b> <code>complete</code>, pass the progress output through a Function node that removes the per-message override:</p>
  <pre>delete msg.cmd;
msg.payload = { ok: true };
return msg;</pre>
  <p>Alternatively, override the downstream action explicitly:</p>
  <pre>msg.cmd = "complete";
msg.payload = { ok: true };
return msg;</pre>
  <p>The node uses the opaque <code>msg.bull.ackId</code> from <code>bullmq run</code> and does not expose BullMQ lock tokens.</p>
</script>

<script type="text/javascript">
  RED.nodes.registerType("bullmq job", {
    color: "#ffffff",
    category: "function",
    defaults: {
      name: { value: "" },
      action: { value: "complete" }
    },
    inputs: 1,
    outputs: 1,
    align: "right",
    icon: "bull_icon.png",
    label: function() {
      return this.name || "bullmq job";
    }
  });
</script>

<script type="text/html" data-template-name="bullmq events">
  <div class="form-row">
    <label for="node-input-name"><i class="fa fa-tag"></i> Name</label>
    <input type="text" id="node-input-name">
  </div>
  <div class="form-row">
    <label for="node-input-queue"><i class="fa fa-tasks"></i> Queue</label>
    <input type="text" id="node-input-queue">
  </div>
  <div class="form-row">
    <label for="node-input-events"><i class="fa fa-bell"></i> Events</label>
    <textarea id="node-input-events" rows="4" style="width: 70%;" placeholder="completed, failed, progress"></textarea>
  </div>
</script>

<script type="text/html" data-help-name="bullmq events">
  <p>Subscribes to BullMQ QueueEvents and emits global queue events as Node-RED messages.</p>
  <p>API reference: <a href="https://docs.bullmq.io/api/classes/v6.QueueEvents.html" target="_blank" rel="noopener noreferrer">QueueEvents</a>.</p>

  <h3>Options</h3>
  <ul>
    <li><b>Name</b>: Optional label shown in the Node-RED workspace.</li>
    <li><b>Queue</b>: Queue config node whose global events will be observed.</li>
    <li><b>Events</b>: Comma or newline separated event names. Leave empty to subscribe to the default set. Useful events include <code>completed</code>, <code>failed</code>, <code>progress</code>, <code>delayed</code>, <code>deduplicated</code>, <code>duplicated</code>, <code>waiting</code>, <code>active</code>, and <code>retries-exhausted</code>.</li>
  </ul>
  <p><code>retries-exhausted</code> is part of the default event set. A flow that leaves <b>Events</b> empty now receives this event in addition to the previous defaults.</p>

  <h3>Output message</h3>
  <ul>
    <li><code>msg.topic</code>: Event name.</li>
    <li><code>msg.payload</code>: BullMQ event payload.</li>
    <li><code>msg.bull</code>: Queue, event name, and event id metadata.</li>
  </ul>

  <h3>Example</h3>
  <p>Set <b>Events</b> to <code>completed, failed, delayed, deduplicated, progress</code> and wire the node to a debug node to inspect queue activity.</p>
</script>

<script type="text/javascript">
  RED.nodes.registerType("bullmq events", {
    color: "#ffffff",
    category: "function",
    defaults: {
      name: { value: "" },
      queue: { type: "bullmq-queue-server", required: true },
      events: { value: "" }
    },
    inputs: 0,
    outputs: 1,
    align: "left",
    icon: "bull_icon.png",
    label: function() {
      return this.name || "bullmq events";
    }
  });
</script>

<script type="text/html" data-template-name="bullmq flow">
  <div class="form-row">
    <label for="node-input-name"><i class="fa fa-tag"></i> Name</label>
    <input type="text" id="node-input-name">
  </div>
  <div class="form-row">
    <label for="node-input-queue"><i class="fa fa-tasks"></i> Queue</label>
    <input type="text" id="node-input-queue">
  </div>
</script>

<script type="text/html" data-help-name="bullmq flow">
  <p>Adds atomic BullMQ parent and child job trees with <code>FlowProducer</code>.</p>
  <p>API reference: <a href="https://docs.bullmq.io/api/classes/v6.FlowProducer.html" target="_blank" rel="noopener noreferrer">FlowProducer</a> and <a href="https://docs.bullmq.io/api/types/v6.FlowJob.html" target="_blank" rel="noopener noreferrer">FlowJob</a>.</p>

  <h3>Options</h3>
  <ul>
    <li><b>Name</b>: Optional label shown in the Node-RED workspace.</li>
    <li><b>Queue</b>: Queue config node used by the FlowProducer connection. Each flow job still supplies its own <code>queueName</code>.</li>
  </ul>

  <h3>Input message</h3>
  <ul>
    <li><code>msg.payload</code>: One BullMQ flow tree, or an array of independent root trees added atomically with <code>FlowProducer.addBulk</code>. A root object contains <code>name</code>, <code>queueName</code>, <code>data</code>, optional <code>opts</code>, and optional <code>children</code>.</li>
    <li><code>msg.flowopts</code>: Optional FlowProducer options applied when adding one tree. It does not apply to an array.</li>
  </ul>

  <p>A child job without an explicit <code>opts.jobId</code> gets a UUID from BullMQ v6, not the incremental numeric id used before.</p>
  <p>Use one tree for related parent/child jobs, including dependencies that span queues. Use an array only for independent root trees that must all be added or none added. On Redis Cluster, omit each flow job's <code>prefix</code> to use this node's prefix throughout. If prefixes are set per job, they must contain the same hash tag, and each worker must use its queue's exact prefix.</p>

  <h3>Example</h3>
  <pre>msg.payload = {
  name: "parent-report",
  queueName: "reports",
  data: { payload: "parent" },
  children: [
    {
      name: "child-report",
      queueName: "reports",
      data: { payload: "child" }
    }
  ]
};
return msg;</pre>
  <p>The child jobs are processed before the parent job. A parent can read child results with a manual-mode <code>bullmq run</code> and a <code>bullmq job</code> action such as <code>getChildrenValues</code>.</p>
</script>

<script type="text/javascript">
  RED.nodes.registerType("bullmq flow", {
    color: "#ffffff",
    category: "function",
    defaults: {
      name: { value: "" },
      queue: { type: "bullmq-queue-server", required: true }
    },
    inputs: 1,
    outputs: 1,
    align: "right",
    icon: "bull_icon.png",
    label: function() {
      return this.name || "bullmq flow";
    }
  });
</script>