telega/jobs

Work the bot does later: a reminder in an hour, a nightly digest, a retry after the rate limit clears.

Two kinds of job, and the difference is what happens when the bot restarts:

In-memory (run_after, run_every) — a closure and a BEAM timer. Cheap, cancellable, and gone the moment the VM stops. Good for the periodic chores of a running bot: refreshing a cache, sweeping a table.
Persisted (persisted, persisted_every) — a named handler, a chat, a payload and a due time written to a KeyValueStorage. The scheduler re-reads them at start, so a reminder still fires after a deploy. Anything a user is waiting for belongs here.

import telega/jobs

let assert Ok(scheduler) =
  jobs.new(bot)
  |> jobs.with_storage(storage)
  |> jobs.with_handler("reminder", fn(ctx, payload) {
    let text = decode.run(payload, decode.at(["text"], decode.string))
    let _ = reply.with_text(ctx, result.unwrap(text, "⏰"))
    Nil
  })
  |> jobs.start()

// ...from a handler, an hour from now:
jobs.persisted(
  scheduler,
  id: "reminder:" <> int.to_string(ctx.update.chat_id),
  handler: "reminder",
  chat_id: ctx.update.chat_id,
  user_id: ctx.update.from_id,
  at: timestamp.add(timestamp.system_time(), duration.hours(1)),
  payload: json.object([#("text", json.string("stand up"))]),
)

A persisted job runs against a telega.background_context: the session is loaded, dependencies are injected, and everything a handler can do — reply, edit, dialog.refresh — works. What does not is wait_*: there is no chat instance to suspend.

Ids, and what they cost

A persisted job’s id is its identity in storage. Scheduling twice with the same id replaces the first — which is what you want for “remind me at 9”, and not what you want for a queue of independent reminders. Include whatever makes them distinct ("digest:" <> chat, "reminder:" <> chat <> ":" <> nonce). cancel takes the same id.

Guarantees

At-most-once, not exactly-once. A job whose handler crashes mid-run is not retried — it has already been taken out of storage. A job whose context cannot be built (an unreadable session) is retried a few times and then dropped with an error log, since retrying forever would be a louder failure than losing the job. A job whose handler name is not registered stays in storage untouched, so the deploy that adds the handler picks it up at start.

Every run emits ["telega", "job", "run"]; a failure to run one emits ["telega", "job", "error"].

Types

Builder

opaque </>

Scheduler configuration, built by new and started by start.

pub opaque type Builder(session, error, dependencies)

JobHandler

</>

What a persisted job runs: the update-less context for its chat and the payload it was scheduled with, still undecoded.

pub type JobHandler(session, error, dependencies) =
  fn(bot.Context(session, error, dependencies), dynamic.Dynamic) -> Nil

Message

opaque </>

The scheduler actor’s protocol. Opaque — every message has a function that sends it; the type is public only so it can be named.

pub opaque type Message(session, error, dependencies)

Scheduler

opaque </>

A running scheduler. Hand it to handlers through dependencies, or name it with with_name and rebuild the handle with from_name.

pub opaque type Scheduler(session, error, dependencies)

Values

cancel

</>

pub fn cancel(
  scheduler scheduler: Scheduler(session, error, dependencies),
  id id: String,
) -> Nil

Forget a job, in memory or in storage. Unknown ids are ignored.

from_name

</>

pub fn from_name(
  name: process.Name(Message(session, error, dependencies)),
) -> Scheduler(session, error, dependencies)

A handle for a scheduler started under name.

Nothing checks that it is running — a send to a dead name is dropped, the same as any named subject.

job_retry_delay_ms

</>

pub const job_retry_delay_ms: Int

How long a job waits before its context is tried again.

max_job_attempts

</>

pub const max_job_attempts: Int

How many times a job whose context cannot be built is put back before it is given up on.

new

</>

pub fn new(
  telega telega: telega.Telega(session, error, dependencies),
) -> Builder(session, error, dependencies)

Start configuring a scheduler for a started bot.

pending

</>

pub fn pending(
  scheduler scheduler: Scheduler(session, error, dependencies),
  timeout timeout: Int,
) -> Result(List(String), Nil)

The ids the scheduler is currently holding, in no particular order.

Mostly for tests and health endpoints. A scheduler that is gone, or does not answer within timeout milliseconds, reports Error(Nil) rather than taking the caller down with it.

persisted

</>

pub fn persisted(
  scheduler scheduler: Scheduler(session, error, dependencies),
  id id: String,
  handler handler: String,
  chat_id chat_id: Int,
  user_id user_id: Int,
  at at: timestamp.Timestamp,
  payload payload: json.Json,
) -> Nil

Schedule a job that survives a restart: run handler for this chat at at, with payload.

Scheduling an id that already exists replaces it. A time already in the past runs as soon as the scheduler sees it.

persisted_every

</>

pub fn persisted_every(
  scheduler scheduler: Scheduler(session, error, dependencies),
  id id: String,
  handler handler: String,
  chat_id chat_id: Int,
  user_id user_id: Int,
  interval_ms interval_ms: Int,
  payload payload: json.Json,
) -> Nil

Schedule a repeating job that survives a restart: run handler for this chat every interval_ms, starting one interval from now.

Each run schedules the next one and writes it before it dispatches, so a restart mid-run resumes the series rather than ending it.

run_after

</>

pub fn run_after(
  scheduler scheduler: Scheduler(session, error, dependencies),
  delay_ms delay_ms: Int,
  job job: fn(client.TelegramClient) -> Nil,
) -> String

Run job once, delay_ms from now, and forget it if the bot restarts first. Returns the id cancel takes.

run_every

</>

pub fn run_every(
  scheduler scheduler: Scheduler(session, error, dependencies),
  interval_ms interval_ms: Int,
  job job: fn(client.TelegramClient) -> Nil,
) -> String

Run job every interval_ms, starting one interval from now, until the bot stops or cancel is called with the returned id.

The next run is armed after the previous one is dispatched, so a job that takes longer than its interval does not pile up.

start

</>

pub fn start(
  builder builder: Builder(session, error, dependencies),
) -> Result(
  Scheduler(session, error, dependencies),
  error.TelegaError,
)

Start the scheduler.

Persisted jobs are read back from storage here: everything already due runs straight away, the rest is armed for its due time.

with_handler

</>

pub fn with_handler(
  builder builder: Builder(session, error, dependencies),
  name name: String,
  run run: fn(
    bot.Context(session, error, dependencies),
    dynamic.Dynamic,
  ) -> Nil,
) -> Builder(session, error, dependencies)

The name is what is written to storage in place of the closure, so it has to mean the same thing across deploys. Registering the same name twice keeps the last one.

with_name

</>

pub fn with_name(
  builder builder: Builder(session, error, dependencies),
  name name: process.Name(Message(session, error, dependencies)),
) -> Builder(session, error, dependencies)

Register the scheduler actor under a process name, so handlers can reach it with from_name instead of being handed the value.

with_storage

</>

pub fn with_storage(
  builder builder: Builder(session, error, dependencies),
  storage storage: storage.KeyValueStorage(storage_error),
) -> Builder(session, error, dependencies)

Give the scheduler somewhere to keep persisted jobs.

Without it persisted has nowhere to write and logs an error instead; the in-memory jobs work either way. Use the same backend the sessions and flows use — jobs live under their own job: namespace.