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.
scan returns every key beginning with the given prefix (live keys only).

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),
    delete: fn(String) -> Result(Nil, error),
    scan: fn(String) -> Result(List(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),
  delete: fn(String) -> Result(Nil, error),
  scan: fn(String) -> Result(List(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.

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)
    }
  },
)