Schedules and change detection

A schedule is a stored job request plus a cron: every time it fires, Kaabist creates an ordinary job from it, with the same envelope, the same credits and the same webhooks. With change detection on, Kaabist also compares the result with the previous run's and tells you exactly what moved, so you hear about a competitor's new price instead of reading the same page every hour.

#When to use it

Use a schedule when the same program should run on a clock: a price watch, a daily report export, a stock check. Turn on change detection when you care about differences rather than every result.

Use something else when:

• the run depends on an event in your own system (an order, a sign-up): create the job from that event with POST /v1/jobs;
• you need many inputs at once: create a batch (POST /v1/jobs/batch), from your own scheduler if it must recur;
• you want to know that the program broke, not that the data changed: that is drift detection and canaries, which Kaabist runs for you (see drift and canaries).

#Creating a schedule

{
  "program": "competitor-price@production",
  "inputs": { "product_id": "1" },
  "cron": "0 * * * *",
  "timezone": "Europe/Istanbul",
  "change_detection": { "enabled": true, "fields": ["price"] },
  "webhook_url": "https://hooks.example.com/kaab",
  "options": { "max_tier": "http" },
  "overlap_policy": "skip",
  "catch_up": false
}

• program is any program reference (name, name@3, name@alias, prg_…, pv_…). It must resolve now, and it is resolved again at every run, so a schedule on @production follows the alias when you move it.
• inputs are checked against the program's inputs as a job's would be and stored encrypted. secret inputs are never returned.
• cron has five fields and is read in timezone (an IANA zone, default UTC).
• change_detection.fields are field paths of the program's extract (price, products[].price). Without them the whole of data is compared. An unknown path is refused with 422 validation_failed.
• options are laid over the program's options for every run.
• webhook_url receives each run's job events, as on POST /v1/jobs. Schedule events (schedule.changed and the rest) go to your registered webhook endpoints.
• A schedule created with an sk_test_ key runs test jobs.

const schedule = await kaab.schedules.create({
  program: `${name}@latest`, // follows the alias: publishing a fix changes what runs
  inputs,
  cron: "0 * * * *",
  timezone: "Europe/Istanbul",
  change_detection: { enabled: true, fields: ["price"] },
  overlap_policy: "skip",
});
console.log(
  `${schedule.id}: "${schedule.cron}" ${schedule.timezone}, next run ${schedule.next_run_at}`,
);

schedule = kaab.schedules.create(
    program=f"{name}@latest",  # follows the alias: publishing a fix changes what runs
    inputs=inputs,
    cron="0 * * * *",
    timezone="Europe/Istanbul",
    change_detection={"enabled": True, "fields": ["price"]},
    overlap_policy="skip",
)
print(f'{schedule.id}: "{schedule.cron}" {schedule.timezone}, next run {schedule.next_run_at}')

body, err := kaab.Body[client.CreateScheduleJSONBody](map[string]any{
    "program":          name + "@latest", // follows the alias: publishing a fix changes what runs
    "inputs":           inputs,
    "cron":             "0 * * * *",
    "timezone":         "Europe/Istanbul",
    "change_detection": map[string]any{"enabled": true, "fields": []string{"price"}},
    "overlap_policy":   "skip",
})
if err != nil {
    log.Fatal(err)
}
schedule, err := c.Schedules.Create(ctx, body)
if err != nil {
    log.Fatal(err)
}
next := "-"
if schedule.NextRunAt != nil {
    next = schedule.NextRunAt.Format(time.RFC3339)
}
fmt.Printf("%s: %q %s, next run %s\n", schedule.ID, schedule.Cron, schedule.Timezone, next)

#How often a schedule may run

Your plan sets the shortest interval between two fires: every hour on Free, every five minutes on Developer, every minute on Team and above (see plans). A cron that fires more often is refused with 422 schedule_interval_too_small; its details carry min_interval_seconds, requested_seconds, plan and an upgrade_url. The interval is measured as the shortest gap between the next ten fires in your timezone, so */5 9-17 * * * counts as five minutes.

#When it runs

Each schedule fires a few seconds after its cron time: the same offset every time, between 0 and 29 seconds, so that thousands of schedules written 0 * * * * do not all start in the same second. Daylight saving follows the zone: a time that does not exist on the spring-forward day runs as far after the gap as it was into it (30 2 * * * in Europe/Berlin runs at 03:30 that day), and a time that happens twice on the fall-back day runs once, at its first occurrence.

Each fire creates a job tagged kaab:schedule:<id> (GET /v1/jobs?schedule=sch_… lists them), or records a skipped run:

A schedule you pause yourself (POST /v1/schedules/{id}/pause) stays paused until you resume it; only a pause for credits is lifted automatically. Resuming while your organisation is still out of credits is refused with 402 insufficient_credits.

#Change detection

After a scheduled job succeeds, Kaabist takes its data restricted to the watched fields and compares it with the previous run's (key order never counts as a change):

• changed: schedule.changed, with up to 50 changed_paths (JSON Pointers such as /price) and a diff_url;
• the same, or the first run (which has nothing to compare with): schedule.run with changed: false.

GET /v1/schedules/{id}/runs/{job_id}/diff (the diff_url) answers 302 with a link, valid for 15 minutes, to an RFC 6902 JSON Patch from the previous values to the new ones:

[{ "op": "replace", "path": "/price", "value": 1338.4 }]

A run with no change has no diff (404). Changing change_detection with PATCH starts comparing afresh.

#When runs fail

A run that does not succeed sends schedule.run with its status. Five in a row send schedule.failing once, with failure_streak. A failing schedule keeps running: the site may come back, and a fix you publish to the alias is picked up at the next fire.

#Run history

GET /v1/schedules/{id}/runs lists every fire, newest first, skipped ones included: job_id, scheduled_for, started_at, finished_at, status, skipped, skip_reason, changed and diff_url.

// Each fire is a run: its job, whether it changed, and where the JSON Patch of the change is.
const runs = await kaab.schedules.runs(schedule.id);
for (const run of runs.data) {
  console.log(`${run.scheduled_for} ${run.status} changed=${run.changed}`);
  if (run.changed && run.job_id !== null) {
    console.log(`  diff: ${await kaab.schedules.runDiff(schedule.id, run.job_id)}`);
  }
}
console.log(`${runs.data.length} run(s) so far`);

# Each fire is a run: its job, whether it changed, and where the JSON Patch of the change is.
runs = kaab.schedules.runs(schedule.id)
for run in runs.data:
    print(f"{run.scheduled_for} {run.status} changed={run.changed}")
    if run.changed and run.job_id:
        print(f"  diff: {kaab.schedules.run_diff(schedule.id, run.job_id)}")
print(f"{len(runs.data)} run(s) so far")

// Each fire is a run: its job, whether it changed, and where the JSON Patch of the change is.
runs, err := c.Schedules.Runs(ctx, schedule.ID, nil)
if err != nil {
    log.Fatal(err)
}
for _, run := range runs.Data {
    changed := run.Changed != nil && *run.Changed
    fmt.Printf("%v skipped=%v changed=%v\n", run.StartedAt, run.Skipped, changed)
    if changed && run.JobID != nil {
        diff, err := c.Schedules.RunDiff(ctx, schedule.ID, *run.JobID)
        if err != nil {
            log.Fatal(err)
        }
        fmt.Printf("  diff: %s\n", diff)
    }
}
fmt.Printf("%d run(s) so far\n", len(runs.Data))

#Worked example

Watch a competitor's price hourly and hear about it only when it changes. The competitor here is a shop at https://shop.example.com; its product page shows a title, a price in Turkish lira and a SKU.

1. Publish the program. It reads price with "type": "money" from /p/1, at the http tier.
2. Register a webhook endpoint for the schedule events, and keep its secret:
   Copy

   { "url": "https://hooks.example.com/kaab", "events": ["schedule.*"] }
   

3. Create the schedule with "cron": "0 * * * *", your timezone and "change_detection": { "enabled": true, "fields": ["price"] }, as in the snippet above. Its first run records the price and sends schedule.run with changed: false.
4. The price moves. The competitor raises the price by one lira. The next run sends schedule.changed:
   Copy

   {
     "type": "schedule.changed",
     "data": {
       "schedule": "sch_…", "job": "job_…", "changed_paths": ["/price"],
       "url": "https://api.kaab.ist/v1/schedules/sch_…",
       "diff_url": "https://api.kaab.ist/v1/schedules/sch_…/runs/job_…/diff"
     }
   }
   

5. Receive it. Your receiver verifies Kaab-Signature, answers 2xx at once, skips an event id it has already handled, and then reads the new value: the patch at diff_url, or the job itself.

const server = createServer((request, response) => {
  const chunks: Buffer[] = [];
  request.on("data", (chunk: Buffer) => chunks.push(chunk));
  request.on("end", () => {
    try {
      // The raw bytes, before any JSON parser: the signature is over them.
      const event = kaab.webhooks.constructEvent(
        Buffer.concat(chunks),
        request.headers["kaab-signature"]?.toString(),
        secret,
      );
      response.writeHead(204).end(); // answer fast; work afterwards
      if (seen.has(event.id)) return;
      seen.add(event.id);
      console.log(`verified ${event.type} ${event.id}`);
      resolveFirst(event.type);
    } catch (error) {
      response.writeHead(error instanceof KaabSignatureError ? 401 : 400).end();
      console.log(`refused a delivery: ${(error as Error).message}`);
    }
  });
});

class Receiver(BaseHTTPRequestHandler):
    def do_POST(self) -> None:  # noqa: N802
        raw = self.rfile.read(int(self.headers.get("content-length", "0")))  # the bytes, unparsed
        try:
            event = webhooks.construct_event(raw, self.headers.get("kaab-signature"), secret)
        except KaabSignatureError as error:
            self.send_response(401)
            self.end_headers()
            print(f"refused a delivery: {error}")
            return
        self.send_response(204)  # answer fast; work afterwards
        self.end_headers()
        if event["id"] not in seen:
            seen.add(event["id"])
            print(f"verified {event['type']} {event['id']}")
            received.put(event["type"])

    def log_message(self, *_: object) -> None:
        pass

handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    raw, err := io.ReadAll(r.Body) // the bytes, before any JSON decoding
    if err != nil {
        w.WriteHeader(http.StatusBadRequest)
        return
    }
    mu.Lock()
    key := secret
    mu.Unlock()
    event, err := webhooks.ConstructEvent(raw, r.Header.Get(webhooks.SignatureHeader), key)
    if err != nil {
        w.WriteHeader(http.StatusUnauthorized)
        log.Printf("refused a delivery: %v", err)
        return
    }
    w.WriteHeader(http.StatusNoContent) // answer fast; work afterwards
    mu.Lock()
    defer mu.Unlock()
    if !seen[event.ID] {
        seen[event.ID] = true
        fmt.Printf("verified %s %s\n", event.Type, event.ID)
        received <- event.Type
    }
})

A webhook never carries the job's data, only where to read it, so the price itself comes from the API with your key.

#Pitfalls

• Watch the fields that matter. Without fields, every value in data counts, including ones that change on every load (a timestamp, a "viewed by" counter), and every run reports a change.
• A failed run is not a change. Change detection only compares successful runs; a failure sends schedule.run with its status, and five in a row schedule.failing. Subscribe to both if a silent schedule would hurt.
• The diff link expires. diff_url itself is permanent, but the 302 it answers points at a link valid for 15 minutes. Follow it when you need it rather than storing the redirect.
• queue can pile up. overlap_policy: "queue" starts a new job even when the previous one is still running; on a slow site and a tight cron, jobs accumulate.
• Delete schedules before programs. A program cannot be deleted while a schedule points at it (409 program_has_schedules).
• Out of credits pauses everything. Every active schedule pauses with schedule.paused and resumes when credits arrive. A schedule you paused by hand is not resumed with them.

#Reference

• Schedule schema, events (schedule.run, schedule.changed, schedule.paused, schedule.failing)
• CLI: kaab schedules create, kaab schedules runs
• Plans for each plan's minimum interval