Levandor

telegram_connector Inbound Framework — Built & Merged

6 min read

Built + merged to main (commit e048062) on 2026-06-23. Inbound = operator Telegram commands → NATS publish, role-based authorization, restart-replay safety, Admin in-memory kill-switch independence.

High-Level Design

Inbound flow: operator issues /command in Telegram → poller (getUpdates) receives update → authorize (role check) → validate command shape → publish to ops.commands via NATS JetStream (with Nats-Msg-Id dedup) → ops consumer handles.

Key principle: Commands are CODE-DEFINED (not self-serve DB registry). New commands requested via COMMAND_REQUEST.md → @pmv2_telegram implements in the connector.

Commands are dispatch-only; there is NO command oracle. Unknown user → silently ignored (no error/permission message). Insufficient role → “not authorized” reply, no publish.

Activation & Config

Inbound turns on ONLY when TELEGRAM_OWNER_IDS is set. This is additive — outbound relay is unaffected by inbound activation.

Required env:

  • TELEGRAM_BOT_TOKEN — sole holder
  • TELEGRAM_CHAT_ID — default chat for outbound
  • TELEGRAM_OWNER_IDS — comma-separated Telegram user IDs (at least one = Admin)
  • NATS_URL + NATS_*_NKEY_SEED — fleet NATS spine
  • SUPABASE_DB_URL — optional; for role-based perms lookup (DB unreachable at boot = start owners-only, never crash)

Role-Based Permissions

Role hierarchy: Viewer < Operator < Admin

Bootstrap on startup (idempotent CREATE TABLE IF NOT EXISTS):

CREATE TABLE IF NOT EXISTS tg_operators (
  telegram_user_id BIGINT PRIMARY KEY,
  role TEXT NOT NULL CHECK (role IN ('Viewer', 'Operator', 'Admin')),
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);

The connector NEVER writes to this table. It is READ-ONLY from the connector. Bootstrap is idempotent; the connector is NOT a sqlx migrator — this avoids colliding with the shared DB’s _sqlx_migrations table (fleet decision #333).

Admin Kill-Switch Independence

CRITICAL: Admin user IDs from TELEGRAM_OWNER_IDS are resolved IN-MEMORY before any DB lookup. If DB is unreachable at boot:

  • Owners from env are still Admin
  • Connector starts normally, owners-only mode
  • Connector never crashes; outbound untouched
  • DB reconnect on next poll; lazy role refresh

This is why owners are in env: unconditional access even if Supabase is down.

Restart-Replay Safety

CRITICAL Design

Telegram re-delivers unconfirmed updates ≤24h after a restart. The poller MUST handle this correctly.

The fix:

  • Poll loop tracks process_start = now() at boot
  • For each incoming update, check message.date < process_start
  • Pre-restart messages are DROPPED from dispatch (no command execute)
  • BUT offset is still advanced past the dropped update — so a /halt or /autotrade issued before restart never re-executes
  • No offset persistence to SQLite — would reintroduce the replay risk

Why it works: Telegram’s unconfirmed delivery guarantees that re-delivered messages have date <= original_date. By advancing the offset, we acknowledge receipt without executing; Telegram stops re-sending. By dropping from dispatch, we prevent double-execute.

Authorization & Dispatch

Command execution is authorize-before-execute:

  1. Lookup user — find telegram_user_id in tg_operators table (or env override for owners)
  2. Check role — verify role >= command minimum (e.g., /status = Viewer, /halt = Admin)
  3. Unknown user — silently ignore (no “not authorized” message)
  4. Insufficient role — reply “not authorized”, do NOT publish
  5. Authorized — parse args, validate shape, publish ops.commands, await result for reply

Reload Commands (Admin)

/reload-commands (Admin-only) atomically swaps the role permissions snapshot:

  • Reads fresh tg_operators from DB
  • Replaces in-memory cache (prevents TOCTOU race)
  • If DB read fails → keep old snapshot, log warning, do NOT reply (silent fail-safe)

Group Chat Support

Command tokens strip the @BotName suffix automatically (e.g., /halt@my_bot_1234 → /halt).

Architecture: Trait Seams & Dispatch

Similar to outbound relay’s trait separation:

UpdateSource trait

trait UpdateSource {
  async fn poll(&mut self) -> Result<Vec<Update>>;
  async fn ack(&mut self, id: u64) -> Result<()>;
  async fn nak(&mut self, id: u64) -> Result<()>;
}

Impl: GetUpdatesClient (adapted from polymarket-data/notifier.rs).

Publisher trait

trait Publisher {
  async fn publish(&self, subject: &str, msg: &CommandMsg) -> Result<()>;
}

Impl: NatsPublisher (JetStream publish_with_headers; Nats-Msg-Id dedup key = command_id).

dispatch() — pure logic

async fn dispatch(
  update: Update,
  db: &Database,
  owner_ids: &[u64],
  publisher: &Publisher,
) -> Result<()>

Takes Telegram Update, checks auth, validates, publishes. No side effects except DB reads + publish.

Main poll loop

Async-native (no block_on):

loop {
  match source.poll().await {
    Ok(updates) => {
      for update in updates {
        if update.message.date < process_start { source.ack(update.id).await?; continue; }
        match dispatch(update, &db, &owner_ids, &publisher).await {
          Ok(()) => source.ack(update.id).await?,
          Err(e) => { log::warn!(...); source.nak(update.id).await?; }
        }
      }
    }
    Err(e) => { /* exponential backoff */ }
  }
}

Ops Command Contract (Co-Design with PM)

Status: co-design in progress (babylon 378, 2026-06-23).

PM proposed shape for pmv2_contracts::ops:

pub struct OpsCommand {
  pub command_id: String,        // reply correlation key; Nats-Msg-Id = command_id
  pub issued_at: i64,            // unix millis
  pub actor: String,             // telegram user_id as string
  pub command: Command,
}
 
pub enum Command {
  #[serde(rename = "set_mode")]
  SetMode { mode: Mode },
  #[serde(rename = "enable")]
  Enable { target: String },
  #[serde(rename = "disable")]
  Disable { target: String },
  #[serde(rename = "halt")]
  Halt,
  #[serde(rename = "status")]
  Status,
}
 
pub enum Mode {
  #[serde(rename = "off")]
  Off,
  #[serde(rename = "dry")]
  Dry,
  #[serde(rename = "live")]
  Live,
}
 
pub struct OpsResult {
  pub command_id: String,        // reply to OpsCommand.command_id
  pub status: OpsStatus,         // applied | rejected | needs_confirm | error
  pub state: OpsState,           // current state snapshot after command
  pub message: String,           // human-readable detail
}
 
pub enum OpsStatus {
  #[serde(rename = "applied")]
  Applied,
  #[serde(rename = "rejected")]
  Rejected,
  #[serde(rename = "needs_confirm")]
  NeedsConfirm,
  #[serde(rename = "error")]
  Error,
}

Six Adjustments Confirmed

  1. Add schema_version: u32 to OpsCommand + OpsResult (for forward compat)
  2. Add consts OPS_SCHEMA_VERSION, subjects::OPS_COMMANDS, subjects::OPS_RESULTS to pmv2_contracts
  3. actor = telegram user_id as string (not integer)
  4. Two-step confirm for global live mode — driven STATELESSLY by operator token, PM owns the policy
  5. All types serde with #[serde(rename = "...")] inner enum tag, snake_case keys
  6. OpsState/OpsCaps use f64 for status values (OK for status updates)

NATS Subjects

ops.commands              ← inbound publishes OpsCommand (with Nats-Msg-Id = command_id)
ops.results              ← inbound consumes OpsResult for reply correlation (DeliverPolicy::New + pending-map)

Deferred (Gated on PM Contract)

  • Concrete /autotrade, /halt, /status command impls
  • ops.results reply-correlation consumer (DeliverPolicy::New + in-memory pending-map keyed by command_id)
  • Currently, a synthetic /echo command exercises the framework (echoes back the command via ops.results for testing)

Known Limitations & Future Work

No Argument Parsing

Current framework dispatch is command name only (no args dict). This was a v1 decision to keep it simple. Future:

  • Add args: serde_json::Value to OpsCommand
  • Update /set-mode etc. to include target/parameters
  • Keep dispatch pure; let consuming service validate args

No Pending Commands Map

Once PM confirms ops.results contract, we’ll add:

  • In-memory pending: HashMap<String, PendingCmd> (keyed by command_id)
  • Timeout mechanism (if no result within X seconds, reply with “pending” or “timeout”)
  • Reply correlation for multi-service commands (one command may need results from multiple services)

Ops State Persistence

Current: stateless reply correlation (PM owns the state). Future may move to:

  • Postgres ops_commands (audit log)
  • Postgres ops_state (current state snapshot)

Current Build Status

  • Inbound framework: complete, merged to main (commit e048062)
  • Startup tests: verify config parsing, env override for owners
  • Integration tests: tbd (awaiting PM ops.* contract finalization)
  • Concrete commands (/autotrade, /halt, /status): deferred, gated on PM contract

Graph View

Backlinks