Schedule

Schedule is a pipeline for processing deferred events. Events are written to the database using the Scheduler pipe with a specified execution time. The Schedule pipeline periodically scans the database and launches accumulated events into its pipe chain when the scheduled time arrives.

Features

• Event scanning occurs every SCHEDULER_PIPE_ROUTER_PERIOD seconds (default — 10).
• The event passed to Schedule replaces the data in $origin with the value of the Scheduler pipe's params.template.
• In $origin of the Schedule pipeline, the system field _schedule_try_count is available — a counter of event execution attempts (starting from 1). Useful for implementing retry logic with increasing delays.

caution

In Schedule, $origin is replaced with the Scheduler pipe's params.template. The original data is not available — pass everything needed (including send_message_id) through the template.

Usage example in Message

In the Message pipeline, the Scheduler pipe passes an event to Schedule with a delay:

{
  "id": 801,
  "type": "scheduled",
  "params": {
    "pipe_id": 1,
    "delay": 600,
    "template": {
      "camp_id": "$origin.c_id",
      "send_message_id": "$origin.send_message_id",
      "external_id": "$#in.data.externalId"
    }
  },
  "outs": {
    "success": 802,
    "error": 803
  }
}

After 600 seconds, the event will enter pipe 1 of the Schedule pipeline.

Schedule pipeline example

Typical scenario: checking delivery status 10 minutes after sending.

[
  {
    "id": 1,
    "type": "log",
    "params": {
      "message": "[Schedule] try_count: $origin._schedule_try_count, external_id: $origin.external_id",
      "level": "debug"
    },
    "outs": {
      "success": 2
    }
  },
  {
    "id": 2,
    "type": "http_request",
    "params": {
      "method": "GET",
      "timeout": 10,
      "content_type": "json",
      "url": "https://api.example.com/status?id=$origin.external_id",
      "success_codes": [200]
    },
    "outs": {
      "success": 3,
      "error": 5
    }
  },
  {
    "id": 3,
    "type": "unpack",
    "outs": {
      "success": 4,
      "error": 5
    }
  },
  {
    "id": 4,
    "type": "event",
    "params": {
      "channel_id": 10001,
      "action_type": 100002,
      "event": {
        "send_message_id": "$origin.send_message_id"
      }
    }
  },
  {
    "id": 5,
    "type": "log",
    "params": {
      "message": "[Schedule Error] code: $error.code, message: $error.message",
      "level": "error"
    }
  }
]

Here:

1. Log (id: 1) — logs the attempt number and external_id
2. HTTP Request (id: 2) — checks delivery status by external_id
3. Unpack (id: 3) — unpacks the JSON response
4. Eventer (id: 4) — registers a delivery event (deliv, action_type 100002)
5. Log (id: 5) — logs an error if the request fails