Levandor

telegram_connector Inbound Commands — Implementation (2026-06-23)

9 min read

Inbound Commands — Operator Interface to pmv2 Fleet

Status: LIVE, merged main @ 7c8f179. Operator commands deployable + activable via env; round-trip complete upon position_manager ops consumer launch.

Commands Overview

All commands are code-defined — new requests via COMMAND_REQUEST.md template. Dispatch is stateless (no state machine in telegram_connector; services own their state). Responses are two-step for global live mode (PM confirms state before accepting).

Command Reference

CommandEffectMin RoleFormat
/autotrade <mode> [source] [confirm]SetModeOperatorsee below
/haltHalt (= off)Operatorbare command
/statusStatus queryViewerbare command
/enable <source>Enable{source}Operator/enable crypto
/disable <source>Disable{source}Operator/disable polymarket
/reload-commandsRefresh permsAdminbare command

/autotrade <mode> [source] [confirm]

Sets trading mode globally or per-source.

Format:

  • /autotrade off — global off
  • /autotrade dry — global dry-run
  • /autotrade live — global live (→ needs_confirm, operator must resend /autotrade live confirm)
  • /autotrade off crypto — crypto source only: off
  • /autotrade live polymarket confirm — global live with confirmation in one step (still needs PM approval; used if operator knows it’s safe)

Args validation (strict):

  • mode must be one of off, dry, live (lowercase)
  • source optional but if present must be non-empty string, alphanumeric + underscore
  • confirm if present must be exactly confirm (guard against typos)

Publishes: OpsCommand { command_id, issued_at, actor, command: SetMode { mode, source } }

Response scenarios:

  • Applied — PM updated state immediately (off/dry always; live + confirm)
  • NeedsConfirm — PM returned needs_confirm on global live without confirm (operator resends with confirm)
  • Error — PM rejected the command (invalid source, mode locked, etc.)

/halt

Emergency stop. Equivalent to /autotrade off.

Args: none (bare command)

Publishes: OpsCommand { command_id, issued_at, actor, command: Halt }

Response: Applied (PM always accepts halt).

/status

Query current trading mode, breaker state, and source status.

Args: none (bare command)

Publishes: OpsCommand { command_id, issued_at, actor, command: Status }

Response: Applied with state snapshot (PM defines the structure).

/enable <source>, /disable <source>

Enable or disable a specific trading source (e.g., crypto, polymarket).

Args: source name (non-empty, alphanumeric + underscore).

Publishes: OpsCommand { command_id, issued_at, actor, command: Enable { source } } (or Disable).

Response: Applied or Error (if source unknown).

/reload-commands

Force a refresh of operator roles from the DB (tg_operators table). Used after adding new operators without restarting.

Args: none (bare command)

Effect: Reads fresh tg_operators table, atomically swaps the in-memory PermsSnapshot, returns status to operator.

DB failure mode: If read fails, keep old snapshot + log warning + do NOT reply to operator (silent fail-safe — owners still have access via env).

Architecture: Trait Seams & Dispatch

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<()>;
}
 
pub struct Update {
  pub update_id: u64,           // Telegram update ID (monotonic)
  pub message: Option<Message>,
}
 
pub struct Message {
  pub message_id: i32,
  pub from: Option<User>,
  pub chat: Chat,
  pub date: i64,               // unix timestamp (seconds)
  pub text: Option<String>,
}
 
pub struct User {
  pub id: i64,  // telegram_user_id
  pub is_bot: bool,
  pub username: Option<String>,
}
 
pub struct Chat {
  pub id: i64,
  pub chat_type: String,  // "private" | "group" | "supergroup"
}

Impl: GetUpdatesClient (Telegram Bot API getUpdates poller).

Publisher Trait

trait Publisher {
  async fn publish(&self, subject: &str, data: &[u8], headers: Option<Headers>) -> Result<String>;
}

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

Main Poll Loop

async fn run_inbound(
  source: impl UpdateSource,
  db: &Database,
  owner_ids: &[u64],
  publisher: &Publisher,
) -> Result<()> {
  let mut process_start = Instant::now();
  
  loop {
    match source.poll().await {
      Ok(updates) => {
        for update in updates {
          match handle_update(&update, &db, &owner_ids, &publisher, process_start).await {
            Ok(()) => { source.ack(update.update_id).await?; }
            Err(e) => {
              log::warn!("dispatch error: {}", e);
              source.nak(update.update_id).await?;
            }
          }
        }
      }
      Err(e) => {
        log::warn!("poll error, backoff: {}", e);
        tokio::time::sleep(Duration::from_secs(5)).await;
      }
    }
  }
}

handle_update() — Pure Logic

async fn handle_update(
  update: &Update,
  db: &Database,
  owner_ids: &[u64],
  publisher: &Publisher,
  process_start: Instant,
) -> Result<()> {
  // 1. Extract message
  let Some(msg) = &update.message else { return Ok(()); };
  
  // 2. Restart-replay safety: drop pre-start messages
  if msg.date < process_start.elapsed().as_secs() as i64 {
    return Ok(()); // dropped from dispatch, but ack() will advance offset
  }
  
  // 3. Extract sender and command
  let Some(from) = &msg.from else { return Ok(()); };
  let Some(text) = &msg.text else { return Ok(()); };
  
  // 4. Authorize
  let role = effective_role(from.id, &db, owner_ids).await?;
  if role == Role::Unknown {
    return Ok(()); // silent ignore
  }
  
  // 5. Parse command
  let cmd_text = text.trim_start_matches('@').trim();
  if !cmd_text.starts_with('/') { return Ok(()); }
  
  // 6. Dispatch
  return dispatch(cmd_text, from.id, &msg.chat, role, &db, &publisher).await;
}

dispatch() — Command Routing

async fn dispatch(
  text: &str,
  user_id: i64,
  chat: &Chat,
  role: Role,
  db: &Database,
  publisher: &Publisher,
) -> Result<()> {
  // Parse command name + args
  let parts: Vec<&str> = text.split_whitespace().collect();
  let cmd_name = parts[0].trim_start_matches('/').to_lowercase();
  let args = &parts[1..];
  
  // Strip @botname suffix
  let cmd_name = if cmd_name.contains('@') {
    cmd_name.split('@').next().unwrap_or(cmd_name)
  } else {
    &cmd_name
  };
  
  // Check min role + dispatch
  let cmd_id = format!("tg|{}|{}", user_id, update.message.message_id);
  
  match cmd_name {
    "autotrade" => {
      if role < Role::Operator { return reply_not_authorized(chat); }
      return ops_autotrade(args, user_id, &cmd_id, &publisher).await;
    }
    "halt" => {
      if role < Role::Operator { return reply_not_authorized(chat); }
      return ops_halt(user_id, &cmd_id, &publisher).await;
    }
    "status" => {
      if role < Role::Viewer { return reply_not_authorized(chat); }
      return ops_status(user_id, &cmd_id, &publisher).await;
    }
    "enable" => { /* ... */ }
    "disable" => { /* ... */ }
    "reload-commands" => {
      if role < Role::Admin { return reply_not_authorized(chat); }
      return reload_and_reply(&db, chat).await;
    }
    _ => return Ok(()), // unknown command, silently ignore
  }
}

Ops Command Publishing

Each command is serialized to JSON and published to ops.commands with Nats-Msg-Id = command_id:

pub async fn ops_autotrade(
  args: &[&str],
  user_id: i64,
  command_id: &str,
  publisher: &Publisher,
) -> Result<()> {
  // Parse args
  let mode_str = args.get(0).ok_or(anyhow::anyhow!("missing mode"))?;
  let mode = match *mode_str {
    "off" => Mode::Off,
    "dry" => Mode::Dry,
    "live" => Mode::Live,
    _ => return Err(anyhow::anyhow!("invalid mode: {}", mode_str)),
  };
  
  let source = args.get(1).map(|s| s.to_string());
  let confirm = args.get(2).map(|s| *s == "confirm").unwrap_or(false);
  
  // Validate source (if present)
  if let Some(ref src) = source {
    if !src.chars().all(|c| c.is_alphanumeric() || c == '_') {
      return Err(anyhow::anyhow!("invalid source chars"));
    }
  }
  
  // Build OpsCommand
  let cmd = OpsCommand {
    schema_version: OPS_SCHEMA_VERSION,
    command_id: command_id.to_string(),
    issued_at: std::time::SystemTime::now()
      .duration_since(UNIX_EPOCH)?
      .as_millis() as i64,
    actor: user_id.to_string(),
    command: Command::SetMode { mode, source },
  };
  
  // Publish
  let payload = serde_json::to_vec(&cmd)?;
  publisher.publish("ops.commands", &payload, Some(nats_msg_id(command_id))).await?;
  
  Ok(())
}

Reply Correlation & OpsResult Consumer

Correlation on Message Body, Not Msg-Id

telegram_connector correlates results via OpsResult.command_id in the message body, never via Nats-Msg-Id. This is intentional: position_manager publishes results with a distinct msg-id (<command_id>-result) to avoid JetStream dedup collisions — see Reply Dedup Collision.

PendingReplies Map

pub struct PendingReplies {
  pending: RwLock<HashMap<String, PendingCmd>>,
}
 
pub struct PendingCmd {
  pub chat_id: i64,
  pub issued_at: Instant,
  pub timeout: Duration,
}

When a command is published, an entry is added to pending. When an OpsResult arrives (matching command_id), the reply is sent to chat and the entry removed.

OpsResults Consumer

pub async fn consume_ops_results(
  stream: &jetstream::Stream,
  pending: Arc<PendingReplies>,
  telegram: &TelegramClient,
) -> Result<()> {
  let consumer = stream
    .create_consumer(PushConsumer {
      durable_name: Some("tg_ops_results".to_string()),
      config: ConsumerConfig {
        deliver_policy: DeliverPolicy::New, // skip OPS backlog
        ..Default::default()
      }
    })
    .await?;
  
  let mut messages = consumer.messages().await?;
  
  while let Some(msg) = messages.next().await {
    match msg {
      Ok(msg) => {
        let result: OpsResult = serde_json::from_slice(&msg.payload)?;
        
        // Find pending command
        if let Some(pending_cmd) = pending.pending.write().remove(&result.command_id) {
          // Format and send reply
          let reply_text = format_result(&result);
          telegram.send_message(pending_cmd.chat_id, &reply_text).await.ok();
        }
        
        msg.ack().await?;
      }
      Err(e) => {
        log::error!("consumer error: {}", e);
        tokio::time::sleep(Duration::from_secs(5)).await;
      }
    }
  }
  
  Ok(())
}

Result Formatting

pub fn format_result(result: &OpsResult) -> String {
  let status_str = match result.status {
    OpsStatus::Applied => "applied",
    OpsStatus::Rejected => "rejected",
    OpsStatus::NeedsConfirm => "needs confirmation",
    OpsStatus::Error => "error",
  };
  
  format!("{}: {}", status_str, result.message)
}

Authorization & Role Hierarchy

Role Enum

#[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Clone, Copy)]
pub enum Role {
  Unknown = 0,
  Viewer = 1,
  Operator = 2,
  Admin = 3,
}

Effective Role Lookup

pub async fn effective_role(
  user_id: i64,
  db: &Database,
  env_owner_ids: &[u64],
) -> Result<Role> {
  // 1. Check env override (in-memory, always succeeds)
  if env_owner_ids.contains(&user_id) {
    return Ok(Role::Admin);
  }
  
  // 2. Check DB (may fail)
  match db.query_operator_role(user_id).await {
    Ok(Some(role)) => Ok(role),
    Ok(None) => Ok(Role::Unknown),
    Err(e) => {
      log::warn!("DB lookup failed: {}, defaulting to Unknown", e);
      Ok(Role::Unknown)
    }
  }
}

Kill-switch design: Env owners are resolved before any DB call. If DB is unreachable at boot, owners still have full access.

Permissions Snapshot (Atomic Reload)

pub struct PermsSnapshot {
  pub tg_operators: HashMap<i64, Role>,
  pub fetched_at: Instant,
}
 
pub async fn reload_into(
  db: &Database,
  snapshot: &RwLock<PermsSnapshot>,
) -> Result<()> {
  match db.fetch_all_operators().await {
    Ok(operators) => {
      let mut map = HashMap::new();
      for (user_id, role) in operators {
        map.insert(user_id, role);
      }
      
      let new_snapshot = PermsSnapshot {
        tg_operators: map,
        fetched_at: Instant::now(),
      };
      
      *snapshot.write() = new_snapshot;
      log::info!("perms snapshot refreshed");
      Ok(())
    }
    Err(e) => {
      log::warn!("failed to reload perms: {}", e);
      Err(e)
    }
  }
}

Restart-Replay Safety

Problem: Telegram’s getUpdates can re-deliver unconfirmed updates for up to 24 hours after a restart. A pre-restart /halt could re-execute.

Solution:

  1. Track process_start = Instant::now() at boot
  2. For each incoming update, check message.date < process_start
  3. If pre-restart: drop from dispatch but still ack the offset
  4. Telegram sees the offset advance → stops re-delivering

Why it works: Telegram uses the offset to know which updates were processed. By advancing the offset even for dropped messages, we tell Telegram “we’ve seen this” without executing it.

if msg.date < (now() - process_start_duration) as i64 {
  // Pre-restart: drop from dispatch, but ack so offset advances
  return Ok(()); // ack() will be called by the poll loop
}

Inbound Activation Gating

Inbound only activates if TELEGRAM_OWNER_IDS env var is set (comma-separated list of Telegram user IDs).

pub fn main() {
  let config = Config::from_env()?;
  
  // Outbound always runs
  tokio::spawn(run_outbound_relay(config.clone()));
  
  // Inbound only if owners set
  if !config.telegram_owner_ids.is_empty() {
    tokio::spawn(run_inbound_commands(config));
  }
  
  // ... wait for shutdown
}

If TELEGRAM_OWNER_IDS is not set or empty, the connector runs outbound-only (unchanged behavior).

Graph View

Backlinks