telega/storage

Unified key-value storage contract shared by sessions and flows.

KeyValueStorage is the single low-level contract that every backend implements (ETS in core; Postgres/SQLite/Redis as separate packages). Values are opaque Strings — callers serialize to/from JSON themselves.

The two bridges below derive the higher-level SessionSettings and FlowStorage contracts from a single KeyValueStorage, so a bot only needs to wire up one backend for both sessions and flows.

Types

KeyValueStorage

</>

Backend-agnostic key-value store.

get returns None for a missing key.
set stores a value with no expiration.
set_with_ttl stores a value that expires after ttl_ms milliseconds. Backends without native TTL emulate it with lazy expiration on access.
compare_and_set(key, expected, value, ttl_ms) writes value only if the key currently holds expected — None meaning absent or expired — and answers whether it did. The check and the write are one atomic step on the backend, which is what makes a read-modify-write safe across processes and nodes (store.update) and lets several nodes claim one job. ttl_ms is as for set_with_ttl; None never expires.
scan(prefix, cursor, limit) returns one page of live keys beginning with the prefix — about limit of them — and the cursor of the next page, None once there is no more. Start with None. Backends paging by key order can derive the cursor with next_cursor; scan_all walks every page for the callers that want the whole prefix.

telega/testing/storage.check is the contract as a test; run it from a backend’s suite.

pub type KeyValueStorage(error) {
  KeyValueStorage(
    get: fn(String) -> Result(option.Option(String), error),
    set: fn(String, String) -> Result(Nil, error),
    set_with_ttl: fn(String, String, Int) -> Result(Nil, error),
    compare_and_set: fn(
      String,
      option.Option(String),
      String,
      option.Option(Int),
    ) -> Result(Bool, error),
    delete: fn(String) -> Result(Nil, error),
    scan: fn(String, option.Option(String), Int) -> Result(
      #(List(String), option.Option(String)),
      error,
    ),
  )
}

Constructors

KeyValueStorage(
  get: fn(String) -> Result(option.Option(String), error),
  set: fn(String, String) -> Result(Nil, error),
  set_with_ttl: fn(String, String, Int) -> Result(Nil, error),
  compare_and_set: fn(
    String,
    option.Option(String),
    String,
    option.Option(Int),
  ) -> Result(Bool, error),
  delete: fn(String) -> Result(Nil, error),
  scan: fn(String, option.Option(String), Int) -> Result(
    #(List(String), option.Option(String)),
    error,
  ),
)

Values

dead_letters_from_storage

</>

pub fn dead_letters_from_storage(
  storage storage: KeyValueStorage(error),
  retention_ms retention_ms: option.Option(Int),
) -> dead_letter.DeadLetters

Derive a dead-letter queue from a KeyValueStorage.

Letters live under the dlq: prefix, alongside sessions (session:) and flows (flow:), so one backend covers all three. The backend’s error type is flattened to a String — a dead letter is written from the bot actor’s crash path, where nothing matches on it.

retention_ms bounds how long a letter is kept. None keeps letters until they are replayed or dropped, which is what you want while debugging and not what you want unattended: a bot crashing in a loop writes one entry per distinct update_id.

telega.with_dead_letters(
  builder,
  storage.dead_letters_from_storage(storage, retention_ms: Some(604_800_000)),
)

flow_storage_from_storage

</>

pub fn flow_storage_from_storage(
  storage storage: KeyValueStorage(error),
) -> types.FlowStorage(error)

Derive FlowStorage from a KeyValueStorage.

Flow instances are stored under the flow: key namespace as complete JSON (see instance.to_json), so subflows and parallel state survive restarts. list_by_user is served by scan over the namespace, replacing the secondary index used by the legacy ETS-only implementation.

flow_storage_from_storage_with_retention

</>

pub fn flow_storage_from_storage_with_retention(
  storage storage: KeyValueStorage(error),
  retention_ms retention_ms: Int,
) -> types.FlowStorage(error)

Derive FlowStorage that lets the backend reclaim abandoned instances.

Every save renews the entry, so retention_ms is “how long an instance may sit untouched before the backend drops it”. Redis expires it natively; the ETS and SQL backends expire lazily on access and skip expired keys in scan. Without this, a flow a user walked away from stays in storage forever — nothing sweeps it.

Pick a retention comfortably longer than the flow’s builder.with_ttl: once the entry is gone the instance simply looks absent, so on_timeout will not fire for it and the user’s next message starts a fresh flow.

map_error

</>

pub fn map_error(
  storage: KeyValueStorage(a),
  with f: fn(a) -> b,
) -> KeyValueStorage(b)

The same storage reporting its errors as another type — a sqlight.Error as the bot’s own error, or as a String for the subsystems that only log them (telega/jobs, the dead-letter queue).

next_cursor

</>

pub fn next_cursor(
  page: List(String),
  limit: Int,
) -> option.Option(String)

The cursor a backend paging by key order hands back for page: the last key when the page is full, nothing when it is the final page.

scan_all

</>

pub fn scan_all(
  storage: KeyValueStorage(error),
  prefix: String,
) -> Result(List(String), error)

Every live key under prefix, page by page through scan.

session_settings_from_storage

</>

pub fn session_settings_from_storage(
  storage storage: KeyValueStorage(error),
  encode encode: fn(session) -> json.Json,
  decode decoder: decode.Decoder(session),
  default default: fn() -> session,
) -> bot.SessionSettings(session, error)

Derive SessionSettings from a KeyValueStorage.

Sessions are stored under the session: key namespace as JSON produced by encode. A decode failure on load is treated as “no session” so the bot falls back to default instead of crashing on a corrupt or migrated value — but it is reported first (telega.storage.decode_error telemetry plus an error log), because the next handler will persist that default over the value that failed to decode.

session_settings_from_storage_versioned

</>

pub fn session_settings_from_storage_versioned(
  storage storage: KeyValueStorage(error),
  encode encode: fn(session) -> json.Json,
  decode decoder: decode.Decoder(session),
  default default: fn() -> session,
  version version: Int,
  migrate migrate: fn(Int, dynamic.Dynamic) -> Result(
    session,
    Nil,
  ),
) -> bot.SessionSettings(session, error)

Derive SessionSettings that carry a schema version.

The stored value is wrapped in {"v": <version>, "d": <encoded session>}, so a build that changes the shape of its session can still read what an older build wrote. On load:

the envelope’s version matches → decode with decode;
it does not → hand migrate the stored version and the raw payload;
there is no envelope (the value was written by session_settings_from_storage) → the version is 0 and migrate gets the whole value.

A migrate returning Error(Nil) is treated exactly like a value that will not decode: reported, then read as “no session”, so the caller falls back to default.

storage.session_settings_from_storage_versioned(
  storage:,
  encode: encode_session,
  decode: session_decoder(),
  default: fn() { Session(name: "", locale: "en") },
  version: 2,
  migrate: fn(from, raw) {
    case from {
      // v1 had no `locale`.
      1 -> decode.run(raw, v1_decoder()) |> result.replace_error(Nil)
      _ -> Error(Nil)
    }
  },
)