telega/roles

Role-based access control: gate handlers on a user’s chat role.

Telegram exposes a user’s role in a chat through getChatMember. This module wraps that call with a small TTL cache (one API round-trip is too slow to repeat on every message) and exposes it three ways:

Booleans — is_admin / is_owner for ad-hoc checks inside a handler.
use guards — ensure_admin / ensure_owner wrap a handler body and run an on_denied branch otherwise.
Router middleware — require_admin / require_owner gate every route of a (sub-)router.

“Admin” means administrator or owner; “owner” means the chat creator only.

Caching

new_cache returns a cache backed by an ETS table owned by the calling process — create it once at bot setup, not inside a handler. Entries expire after ttl_ms; pass ttl_ms: 0 to disable caching and always hit the API. The cache is keyed by {chat_id}:{user_id}, so a role change (promote / demote) is picked up after at most ttl_ms.

import telega/roles
import telega/router

// Cache roles for 60s.
let cache = roles.new_cache(ttl_ms: 60_000)

// Admin-only /ban command:
router.new("admin")
|> router.on_command("ban", fn(ctx, _cmd) {
  use ctx <- roles.ensure_admin(ctx, cache, on_denied: fn(ctx) {
    reply.with_text(ctx, "Admins only.")
  })
  // ... ban logic, only reached for admins ...
  Ok(ctx)
})

On an API error the check fails closed (access denied) and the result is not cached, so the next update retries.

Types

RoleCache

opaque </>

A TTL cache of resolved chat roles. Create once with new_cache.

pub opaque type RoleCache

Values

ensure_admin

</>

pub fn ensure_admin(
  ctx ctx: bot.Context(session, error, dependencies),
  cache cache: RoleCache,
  on_denied on_denied: fn(
    bot.Context(session, error, dependencies),
  ) -> Result(bot.Context(session, error, dependencies), error),
  next next: fn(bot.Context(session, error, dependencies)) -> Result(
    bot.Context(session, error, dependencies),
    error,
  ),
) -> Result(bot.Context(session, error, dependencies), error)

use-friendly guard: run next only if the user is an admin/owner of the current chat, otherwise run on_denied.

use ctx <- roles.ensure_admin(ctx, cache, on_denied: deny)
// admin-only body

ensure_owner

</>

pub fn ensure_owner(
  ctx ctx: bot.Context(session, error, dependencies),
  cache cache: RoleCache,
  on_denied on_denied: fn(
    bot.Context(session, error, dependencies),
  ) -> Result(bot.Context(session, error, dependencies), error),
  next next: fn(bot.Context(session, error, dependencies)) -> Result(
    bot.Context(session, error, dependencies),
    error,
  ),
) -> Result(bot.Context(session, error, dependencies), error)

use-friendly guard: run next only if the user owns the current chat, otherwise run on_denied.

is_admin

</>

pub fn is_admin(
  cache cache: RoleCache,
  client client: client.TelegramClient,
  chat_id chat_id: Int,
  user_id user_id: Int,
) -> Bool

True if the user is an administrator or the owner of the chat.

is_owner

</>

pub fn is_owner(
  cache cache: RoleCache,
  client client: client.TelegramClient,
  chat_id chat_id: Int,
  user_id user_id: Int,
) -> Bool

True if the user is the owner (creator) of the chat.

new_cache

</>

pub fn new_cache(ttl_ms ttl_ms: Int) -> RoleCache

Create a role cache. Entries expire after ttl_ms milliseconds; ttl_ms: 0 disables caching (every check hits the API).

The ETS table is owned by the calling process — create the cache from a long-lived process (bot setup), not from a handler.

require_admin

</>

pub fn require_admin(
  cache cache: RoleCache,
  on_denied on_denied: fn(
    bot.Context(session, error, dependencies),
  ) -> Result(bot.Context(session, error, dependencies), error),
) -> fn(
  fn(bot.Context(session, error, dependencies), update.Update) -> Result(
    bot.Context(session, error, dependencies),
    error,
  ),
) -> fn(bot.Context(session, error, dependencies), update.Update) -> Result(
  bot.Context(session, error, dependencies),
  error,
)

Router middleware that gates every route on admin/owner status. Non-admins are handed to on_denied instead of the matched handler.

Apply it to a dedicated admin sub-router and compose it with your main router so only those routes are gated:

let admin =
  router.new("admin")
  |> router.use_middleware(roles.require_admin(cache:, on_denied: deny))
  |> router.on_command("ban", ban_handler)

router.compose(main_router, admin)

require_owner

</>

pub fn require_owner(
  cache cache: RoleCache,
  on_denied on_denied: fn(
    bot.Context(session, error, dependencies),
  ) -> Result(bot.Context(session, error, dependencies), error),
) -> fn(
  fn(bot.Context(session, error, dependencies), update.Update) -> Result(
    bot.Context(session, error, dependencies),
    error,
  ),
) -> fn(bot.Context(session, error, dependencies), update.Update) -> Result(
  bot.Context(session, error, dependencies),
  error,
)

Router middleware that gates every route on owner (creator) status.