Automate pipeline actions with schedules, chained events, and webhooks

Connect to the target database;

Run the following statements:

CREATE SCHEMA IF NOT EXISTS shop_chronos;

CREATE TABLE shop_chronos.customers (
  id integer PRIMARY KEY,
  full_name varchar(100) NOT NULL,
  email varchar(150),
  country varchar(2) NOT NULL,
  tier varchar(10) NOT NULL,
  status varchar(10) NOT NULL,
  credit_limit numeric(10,2),
  created_at timestamp NOT NULL
);

CREATE TABLE shop_chronos.orders (
  id integer PRIMARY KEY,
  customer_id integer NOT NULL,
  amount numeric(10,2) NOT NULL,
  currency varchar(3) NOT NULL,
  status varchar(12) NOT NULL,
  order_date date NOT NULL
);

Check that both tables are empty:

SELECT (SELECT count(*) FROM shop_chronos.customers) AS customers,
       (SELECT count(*) FROM shop_chronos.orders)    AS orders;

In the left rail, select Schedules & events;

Select Schedules;

Select the gear icon (Scheduler settings) above the table;

Read the value of Timezone;

Select Cancel to leave the setting unchanged, or select another time zone and then Save.

On the Schedules page, select New;

In Name this schedule, enter Customers snapshot every minute;

In Select the action that you want to be applied, select Snapshot without truncate;

Select the CRON tab;

Leave all five fields (Minutes, Hour, Day of the month, Month, Day of the week) set to *;

In Pipeline, select Chronos demo;

In Entity, select customers;

Select Save.

Wait until the next minute starts, then reload the page;

Read the Last run column of the schedule;

In the target database, count the loaded rows and compare them with the source:

SELECT count(*) AS target_customers FROM shop_chronos.customers;   -- run on the target
SELECT count(*) AS source_customers FROM shop.customers;           -- run on the source

To prove that each run reloads the data, delete a few target rows and count again after the next minute starts:

DELETE FROM shop_chronos.customers WHERE id <= 5;
SELECT count(*) FROM shop_chronos.customers;   -- 5 fewer than the source, for now

After the next minute starts, run the count again.

In the schedule row, select Edit schedule;

In Name this schedule, enter Customers nightly snapshot;

In Minutes, enter 0;

In Hour, enter 2;

Leave Day of the month, Month, and Day of the week set to *;

Select Save.

On the Schedules page, select New;

In Name this schedule, enter Customers then orders;

In Select the action that you want to be applied, select Snapshot without truncate;

Select the CRON tab and leave the five fields set to *;

In Pipeline, select Chronos demo, and in Entity select customers;

Under Chained Events, select Add event;

In Execution mode, keep Async (fire & forget);

In Action, select Snapshot without truncate;

In the Pipeline list of the chained event, select Chronos demo, and in its Entity list select orders;

Select Save.

Before the next minute starts, delete rows from both target tables:

DELETE FROM shop_chronos.customers WHERE id <= 20;
DELETE FROM shop_chronos.orders WHERE id <= 100;

After the next run, count the rows again;

In the schedule row, select Disable schedule so that the chain stops firing every minute.

Optional, to see the truncate at work: add a row to the target that does not exist in the source:

INSERT INTO shop_chronos.customers (id, full_name, country, tier, status, created_at)
VALUES (99999, 'Stale row, not in the source', 'IT', 'gold', 'active', now());   -- run on the target

On the Schedules page, select New;

In Name this schedule, enter Customers truncate and CDC every minute;

In Select the action that you want to be applied, select Snapshot with truncate + CDC;

Select the CRON tab and leave all five fields set to *;

In Pipeline, select Chronos demo, and in Entity select customers;

Leave Chained Events empty and select Save.

Wait until the next minute starts, reload the page, and read the Action and Last run columns of the schedule:

In the target database, check that the extra row is gone and that the counts match:

SELECT count(*) AS target_customers,
       count(*) FILTER (WHERE id = 99999) AS stale_rows
FROM shop_chronos.customers;                                   -- run on the target
SELECT count(*) AS source_customers FROM shop.customers;       -- run on the source

Open the pipeline Chronos demo and look at the customers entity:

Prove that live changes flow: insert a row in the source, wait a few seconds, and look for it in the target:

INSERT INTO shop.customers (id, full_name, email, country, tier, status, credit_limit, created_at)
VALUES (9601, 'Added after the schedule', 'after@example.com', 'DE', 'silver', 'active', 500, now());   -- run on the source

SELECT id, full_name FROM shop_chronos.customers WHERE id = 9601;                                         -- run on the target

In the schedule row, select Disable schedule so that the table is not emptied and reloaded every minute.

Open the receiver and copy its endpoint URL. In WebHook Tester, create a session and copy the address that Core Hub must call, for example http://webhook-tester:8080/<session-uuid>;;

In the Control Plane, select the gear icon (Settings) in the top bar;

Select Notifications, then the Webhooks tab;

Select Add webhook;

In Name, enter Chronos demo to webhook tester;

In Endpoint URL, enter the receiver address;

In Description (optional), enter a short note.

Scroll down to Objects filters, and in Pipeline select Chronos demo;

In Enabled events, select Entity Snapshot Started, Entity Snapshot Completed, and Entity Snapshot Failed;

Next to Severity filter, select Select all;

Select Send test;

Select Create webhook.

Open Schedules & events > Schedules and, in the Customers then orders row, select Run now;

Open the receiver and select the newest request.

Select Schedules & events, then Events;

Select New;

In Name, enter Refresh customers on demand;

Select the Webhook trigger tab;

In Description (optional), enter a short note;

Under Chained Events, select Add event;

In Action, select Snapshot without truncate;

In Pipeline, select Chronos demo, and in Entity select customers;

Select Save.

In the Trigger flow created — save your token dialog, select the copy icon next to Secret token and store the token in a password manager or secret store;

Copy the Trigger URL;

Select Done.

Delete some rows from the target table, so that the effect is visible:

DELETE FROM shop_chronos.customers WHERE id <= 20;

From a shell, call the trigger URL with the token in the X-Trigger-Token header. Use -k only if the Control Plane has a self-signed certificate:

curl -k -i -X POST "https://<control-plane-address>/chronos/api/triggers/1/fire" \
  -H "X-Trigger-Token: <secret-token>"

Count the target rows again after a few seconds, and select the expand arrow of the event in the Events table.