telegram_connector — Complete Project (2026-06-23)

telegram_connector — Complete & Deployed

Status: BUILT + DEPLOYED. Outbound relay LIVE in production (pin 3baa6e4). Inbound commands deployed and activated; operator round-trip live once position_manager’s ops consumer launches.

Repo: wowjeeez/telegram_connector main @ commit 7c8f179 (0f5e854 docs head). Read /Users/levander/coding/pmv2/telegram_connector/README.md for canonical service reference.

One-Sentence Summary

telegram_connector = the pmv2 fleet’s single Telegram I/O plane: sole bot-token holder, sole Telegram sender, only reader of operator commands. Two independent planes (outbound relay + inbound commands) both NATS-native, both deployed.

Two Planes

OUTBOUND: Durable JetStream Relay (LIVE)

Flow: TELEGRAM_OUTBOUND stream (subject telegram.outbound) → durable pull consumer tg_outbound → bot API sendMessage → Telegram.

Delivery guarantees:

• At-least-once: ack-after-send
• Sent → Ack (advance offset)
• Transient (429/5xx/network) → Nak with reactive backoff (honor 429 retry_after, capped to ack_wait)
• Terminal (400/403) → Term (drop) + loop-guarded meta-alert (no spam)

Contract: TelegramOutbound { msg_id: String, chat?: String, text: String, parse_mode?: String, kind?: String }. Verbatim relay — publisher owns escaping, connector never re-escapes.

NATS: stream TELEGRAM_OUTBOUND (1d retention, 5m dedup window on Nats-Msg-Id), consumer tg_outbound (durable, pull).

Resilience: Rebuilds consumer on stream-end/NATS disconnect; does not exit.

Related: telegram-connector-outbound-design, nats-jetstream-message-source, task7-lib-bin-split-relay-loop-tests, security-reliability-fixes-bab38c4

INBOUND: getUpdates → Ops Publishing (DEPLOYED)

Flow: getUpdates long-poll → authorize operator → validate command → publish to ops.commands → await ops.results → relay back to operator.

Activation: Only if TELEGRAM_OWNER_IDS env var is set (additive — outbound runs regardless).

Command set (code-defined):

Command	Effect	Min Role	Status
/autotrade <off|dry|live> [source] [confirm]	SetMode	Operator	Live
/halt	Halt (= mode off)	Operator	Live
/status	Status query	Viewer	Live
/enable <source>	Enable source	Operator	Live
/disable <source>	Disable source	Operator	Live
/reload-commands	Refresh perms cache	Admin	Live

Permissions: Viewer < Operator < Admin. DB-backed in tg_operators table (connector read-only); bootstrap via CREATE TABLE IF NOT EXISTS (not a sqlx migrator).

Kill-switch independence: TELEGRAM_OWNER_IDS (env) resolve to Admin in-memory BEFORE any DB lookup. If Supabase is down at boot, owners-only mode + no crash.

Restart-replay safety: Pre-startup messages (message.date < process_start) are dropped from dispatch but offset still advances → pre-restart /halt never re-executes.

Two-step confirm: Global live mode requires confirmation → /autotrade live returns needs_confirm → operator resends /autotrade live confirm (stateless).

Correlation: command_id = tg|<user_id>|<message_id> = reply key = Nats-Msg-Id. Results consumer tg_ops_results uses DeliverPolicy::New (skip OPS backlog).

Related: telegram-connector-inbound-framework, telegram-connector-inbound-commands

Wire Contracts (pmv2-contracts @ 4c7a54f)

Types vendored from shared pmv2-contracts crate (private git wowjeeez/pmv2-contracts). Connector owns TelegramOutbound + ops.* types it contributed.

Outbound

pub struct TelegramOutbound {
    pub msg_id: String,
    pub chat: Option<String>,
    pub text: String,
    pub parse_mode: Option<String>,
    pub kind: Option<String>,
}

Inbound — Ops Module

pub struct OpsCommand {
    pub schema_version: u32,
    pub command_id: String,      // reply key
    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, source: Option<String> },
    #[serde(rename = "enable")]
    Enable { source: String },
    #[serde(rename = "disable")]
    Disable { source: String },
    #[serde(rename = "halt")]
    Halt,
    #[serde(rename = "status")]
    Status,
}
 
pub enum Mode { Off, Dry, Live }
 
pub struct OpsResult {
    pub schema_version: u32,
    pub command_id: String,
    pub status: OpsStatus,
    pub state: Option<OpsState>,
    pub message: String,
}
 
pub enum OpsStatus { Applied, Rejected, NeedsConfirm, Error }
pub struct OpsState { /* PM-defined */ }
pub struct OpsCaps { /* PM-defined */ }

Subjects: ops.commands (publish), ops.results (consume). Stream OPS (7d retention, 5m dedup). See Reply Dedup Collision for a critical design lesson on distinct msg-ids in shared streams.

Co-design with position_manager: babylon #377, #378, #389, published #395. PM consumer built @ 2dd1fc8.

NATS Configuration

Subject	Stream	Direction	Durable	Retention	Dedup
telegram.outbound	TELEGRAM_OUTBOUND	consume	tg_outbound	1d	5m
ops.commands	OPS	publish	—	7d	5m
ops.results	OPS	consume	tg_ops_results	7d	5m

NKey permissions: SUB telegram.outbound, PUB ops.commands, SUB ops.results. Env var: NATS_TELEGRAM_NKEY_SEED.

Observability

OTLP → local collector → SigNoz. Never direct-to-SigNoz. Service name: pmv2-telegram-connector.

Logging discipline:

• ERROR = operator-visible failures
• WARN = transient/retry/rate-limit
• INFO = state transitions (metered, <1/s avg)
• DEBUG = per-message (behind RUST_LOG)

Structured attributes: subject, nats_msg_id, chat, kind, error_code, source.

Security: Bot token never in logs. reqwest errors are without_url()-stripped (token in URL).

Related: pmv2-fleet-infra-observability-nats-conventions

Configuration Matrix

Var	Required	Secret	Note
TELEGRAM_BOT_TOKEN	yes	yes	sole holder; dormant-exit if unset
TELEGRAM_CHAT_ID	yes	no	default chat for chat: None messages
NATS_URL	yes	no	fleet NATS spine; dormant-exit if unset
NATS_TELEGRAM_NKEY_SEED	yes	yes	NATS auth seed
TELEGRAM_OWNER_IDS	inbound only	no	comma-sep i64; presence activates inbound; implicit Admin
SUPABASE_DB_URL	inbound only	yes	perms store (read-only)
OTEL_EXPORTER_OTLP_ENDPOINT	optional	no	local OTel collector

Tunables (env): TG_DURABLE_NAME=tg_outbound, TG_ACK_WAIT_SECS=60, TG_MAX_DELIVER=5, TG_NAK_BACKOFF_SECS=30, META_ALERT_ON_DROP=true, TG_GETUPDATES_TIMEOUT_SECS=30.

Dormant-safe: Missing TELEGRAM_BOT_TOKEN or NATS_URL → log + exit 0 (no panic).

Build & Testing

Rust 2024 edition, unsafe_code = "forbid". Vendored pmv2-contracts (private git) requires CARGO_NET_GIT_FETCH_WITH_CLI + clone token at build.

Tests: 34 passing (unit + wiremock integration). Ignored: testcontainer tests for Postgres + ops.results round-trip.

• Unit: trait-seamed fakes (FakeTelegram, FakeSource, FakePublisher)
• Integration: wiremock mocks Telegram API (Ack→200, 429→retry_after, 400→terminal)
• Dormant smoke: unset token → exit 0

CLI: cargo build, cargo test, cargo clippy --all-targets -- -D warnings, cargo fmt --all -- --check.

Deployment

Pattern A (cargo + systemd, ansible). Inbound activates by adding TELEGRAM_OWNER_IDS + SUPABASE_DB_URL to env + re-pin.

Schema bootstrap: tg_operators table applied out-of-band (connector bootstraps via CREATE TABLE IF NOT EXISTS, never runs migrations).

Adding a Command

Commands are code-defined — not self-serve. Request via COMMAND_REQUEST.md template → @pmv2_telegram implements in connector. Requester owns consumer-side implementation.

Code Organization

crates/telegram_connector/src/
  config.rs         env parsing + Dormant guard
  outbound.rs       TelegramOutbound (from pmv2-contracts) + resolve_chat
  telegram.rs       Bot API client (Sender) + classify response + method_url
  relay.rs          outbound loop (run + run_resilient): Ack/Nak/Term + meta-alert
  consumer.rs       JetStream connect + bind_pull_consumer + outbound MessageSource
  telemetry.rs      OTLP Guard init
  version.rs        build-time GIT_HASH/GIT_SUBJECT (build.rs) → startup announce
  main.rs           wiring: outbound + (gated) inbound spawn + graceful shutdown
  inbound/
    role.rs         Role enum { Viewer, Operator, Admin }
    telegram_api.rs getUpdates client (UpdateSource) + Update/Message types
    poller.rs       drain_filter + next_offset (restart-replay safety)
    perms.rs        tg_operators read + PermsSnapshot + reload_into (atomic swap)
    auth.rs         effective_role (env-owner bootstrap)
    command.rs      Command trait + Registry
    ops_commands.rs 5 ops commands → typed OpsCommand
    publisher.rs    Publisher trait + NatsPublisher (ops.commands)
    replies.rs      PendingReplies (correlation) + ops.results consumer + resilient
    dispatch.rs     handle_update / dispatch / format_result / reload routing

Key patterns:

• Trait seams: MessageSource, UpdateSource, Publisher, TelegramClient
• Graceful shutdown: #[cfg(unix)] SIGTERM handler, await pending tasks
• Resilience: reconnect on stream-end, abort+retry on consumer errors
• DRY: bind_pull_consumer(), method_url(), optional parse_mode shared helpers

Engineering Process

Built via Superpowers workflow: brainstorm → spec → writing plan → subagent-driven dev (fresh subagent per task + per-task review + whole-branch review) in git worktrees, merged FF → main.

Validation passes:

1. Correctness pass: data flow, contract compliance, state machine
2. DRY-SOLID pass: trait extraction, no duplication, SRP
3. Integration pass: outbound relay self-heal, inbound ack/nak/term, ops round-trip, graceful shutdown

Bug caught in review: reqwest Error::Display embeds URL (= token). Fix: without_url() on error. Pre-commit hook prevents token leaks.

Final cleanups: strict /autotrade arg parsing, resilient relay, await-on-shutdown, drop dead code (run_poller), testable handle_update, lowercase format_result.

Final State (pin 3baa6e4)

✅ Outbound relay: LIVE in production. Inbound commands: deployed + activated. Ready for ops consumer integration.

✅ All 34 tests passing, clippy clean, startup announce working.

✅ Zero data-correctness defects, zero contract violations, zero silent failures.

✅ Dormant-safe: missing config → clean exit.

Related

• telegram-connector-overview — service summary + wire contracts
• telegram-connector-outbound-design — outbound relay detailed design
• telegram-connector-inbound-framework — inbound framework + auth + restart-replay
• telegram-connector-inbound-commands — command implementations
• nats-jetstream-message-source — pull consumer trait seam
• task7-lib-bin-split-relay-loop-tests — relay loop + integration tests
• security-reliability-fixes-bab38c4 — token leak fix + config hardening
• per-source-spend-caps-730 — #730 per-source spend caps: reply_is_verbatim, DRY config-key ownership, #732 deferral
• connector-contracts-verification-playbook — verify against un-pushed contracts via local [patch]; authoritative cargo output
• pmv2-fleet-infra-observability-nats-conventions — fleet NATS + OTel conventions
• polymarket-fetch — pmv2 trading fleet