Architecting Telegram Bot Broadcasts in PHP: Queues, Grouping, and 429 Mitigation

{
  "ok": false,
  "error_code": 429,
  "description": "Too Many Requests: retry after 9",
  "parameters": {
    "retry_after": 9
  }
}

CREATE TABLE telegram_outbox (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    chat_id BIGINT NOT NULL,
    method VARCHAR(64) DEFAULT 'sendMessage',
    payload TEXT NOT NULL, -- JSON-encoded parameters for the Bot API
    status VARCHAR(20) DEFAULT 'pending', -- pending, sending, sent, failed, blocked
    retry_after_until DATETIME DEFAULT NULL, -- Lock individual chat if 429 occurs
    attempts INTEGER DEFAULT 0,
    error_message TEXT DEFAULT NULL,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_outbox_status_retry ON telegram_outbox (status, retry_after_until);
CREATE INDEX idx_outbox_chat_status ON telegram_outbox (chat_id, status);

<?php

function queueMessage(PDO $db, int $chatId, string $newText): void
{
    $db->beginTransaction();

    // Check for an existing pending sendMessage payload for this user
    $stmt = $db->prepare("
        SELECT id, payload 
        FROM telegram_outbox 
        WHERE chat_id = :chat_id 
          AND status = 'pending' 
          AND method = 'sendMessage' 
        LIMIT 1 FOR UPDATE
    ");
    $stmt->execute([':chat_id' => $chatId]);
    $existing = $stmt->fetch(PDO::FETCH_ASSOC);

    if ($existing) {
        $payload = json_decode($existing['payload'], true);

        // Append the new text with a double newline separator
        $payload['text'] .= "\n\n" . $newText;

        $update = $db->prepare("
            UPDATE telegram_outbox 
            SET payload = :payload, updated_at = CURRENT_TIMESTAMP 
            WHERE id = :id
        ");
        $update->execute([
            ':payload' => json_encode($payload),
            ':id' => $existing['id']
        ]);
    } else {
        // Insert a fresh pending message
        $payload = [
            'chat_id' => $chatId,
            'text' => $newText,
            'parse_mode' => 'HTML'
        ];

        $insert = $db->prepare("
            INSERT INTO telegram_outbox (chat_id, method, payload) 
            VALUES (:chat_id, 'sendMessage', :payload)
        ");
        $insert->execute([
            ':chat_id' => $chatId,
            ':payload' => json_encode($payload)
        ]);
    }

    $db->commit();
}

<?php

class TelegramOutboxProcessor
{
    private PDO $db;
    private string $botToken;
    private int $globalLimitPerSecond = 30;

    public function __construct(PDO $db, string $botToken)
    {
        $this->db = $db;
        $this->botToken = $botToken;
    }

    public function processQueue(): void
    {
        // Fetch pending messages that are not locked by a retry_after timer
        $stmt = $this->db->prepare("
            SELECT * FROM telegram_outbox 
            WHERE status = 'pending' 
              AND (retry_after_until IS NULL OR retry_after_until < CURRENT_TIMESTAMP)
            ORDER BY id ASC 
            LIMIT :limit
        ");

        // Bind limit to prevent exceeding the global 30 msg/sec limit in this batch
        $stmt->bindValue(':limit', $this->globalLimitPerSecond, PDO::PARAM_INT);
        $stmt->execute();
        $messages = $stmt->fetchAll(PDO::FETCH_ASSOC);

        if (empty($messages)) {
            return;
        }

        $startTime = microtime(true);

        foreach ($messages as $msg) {
            $this->sendMessage($msg);
        }

        // Maintain the global rate limit of 30 requests per second
        $elapsedTime = microtime(true) - $startTime;
        $minimumDuration = 1.0; // 1 second
        if ($elapsedTime < $minimumDuration) {
            usleep((int)(($minimumDuration - $elapsedTime) * 1000000));
        }
    }

    private function sendMessage(array $msg): void
    {
        $url = sprintf('https://api.telegram.org/bot%s/%s', $this->botToken, $msg['method']);
        $payload = json_decode($msg['payload'], true);

        $ch = curl_init();
        curl_setopt($ch, CURLOPT_URL, $url);
        curl_setopt($ch, CURLOPT_POST, true);
        curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($payload));
        curl_setopt($ch, CURLOPT_HTTPHEADER, ['Content-Type: application/json']);
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_TIMEOUT, 10);

        $response = curl_exec($ch);
        $httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
        $curlError = curl_error($ch);
        curl_close($ch);

        if ($response === false) {
            $this->markAsFailed($msg['id'], "cURL Error: " . $curlError);
            return;
        }

        $result = json_decode($response, true);
        if (json_last_error() !== JSON_ERROR_NONE) {
            $this->markAsFailed($msg['id'], "Invalid JSON response from Telegram");
            return;
        }

        if ($httpCode === 200 && isset($result['ok']) && $result['ok'] === true) {
            $this->markAsSent($msg['id']);
            return;
        }

        $this->handleApiError($msg, $httpCode, $result);
    }

    private function handleApiError(array $msg, int $httpCode, array $result): void
    {
        $description = $result['description'] ?? 'Unknown error';

        if ($httpCode === 429) {
            // We hit a rate limit. Extract retry_after.
            $retryAfter = $result['parameters']['retry_after'] ?? 5;
            $this->lockChat($msg['chat_id'], $retryAfter);
            $this->resetToPending($msg['id'], "Rate limited. Retry after {$retryAfter}s");
            return;
        }

        if ($httpCode === 403 || strpos($description, 'bot was blocked by the user') !== false) {
            // User blocked the bot. Mark as blocked to prevent future attempts.
            $this->markAsBlocked($msg['chat_id'], $msg['id'], $description);
            return;
        }

        // Other API errors (e.g., chat not found, bad request)
        $this->markAsFailed($msg['id'], "API Error {$httpCode}: {$description}");
    }

    private function markAsSent(int $id): void
    {
        $stmt = $this->db->prepare("UPDATE telegram_outbox SET status = 'sent', updated_at = CURRENT_TIMESTAMP WHERE id = :id");
        $stmt->execute([':id' => $id]);
    }

    private function markAsFailed(int $id, string $error): void
    {
        $stmt = $this->db->prepare("
            UPDATE telegram_outbox 
            SET status = 'failed', 
                error_message = :error, 
                attempts = attempts + 1, 
                updated_at = CURRENT_TIMESTAMP 
            WHERE id = :id
        ");
        $stmt->execute([':id' => $id, ':error' => $error]);
    }

    private function resetToPending(int $id, string $reason): void
    {
        $stmt = $this->db->prepare("
            UPDATE telegram_outbox 
            SET status = 'pending', 
                error_message = :reason, 
                attempts = attempts + 1, 
                updated_at = CURRENT_TIMESTAMP 
            WHERE id = :id
        ");
        $stmt->execute([':id' => $id, ':reason' => $reason]);
    }

    private function lockChat(int $chatId, int $seconds): void
    {
        // Lock all pending messages for this specific chat
        $stmt = $this->db->prepare("
            UPDATE telegram_outbox 
            SET retry_after_until = datetime('now', '+' || :seconds || ' seconds') 
            WHERE chat_id = :chat_id 
              AND status = 'pending'
        ");
        $stmt->execute([
            ':seconds' => $seconds,
            ':chat_id' => $chatId
        ]);
    }

    private function markAsBlocked(int $chatId, int $msgId, string $reason): void
    {
        $this->db->beginTransaction();

        // Mark current message as failed
        $stmt1 = $this->db->prepare("UPDATE telegram_outbox SET status = 'blocked', error_message = :reason WHERE id = :id");
        $stmt1->execute([':id' => $msgId, ':reason' => $reason]);

        // Cancel all other pending messages for this user
        $stmt2 = $this->db->prepare("UPDATE telegram_outbox SET status = 'blocked' WHERE chat_id = :chat_id AND status = 'pending'");
        $stmt2->execute([':chat_id' => $chatId]);

        $this->db->commit();
    }
}

<?php
// worker.php

require_once 'TelegramOutboxProcessor.php';

$db = new PDO('sqlite:bot_database.sqlite');
$db->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);

$botToken = getenv('TELEGRAM_BOT_TOKEN');
if (!$botToken) {
    die("TELEGRAM_BOT_TOKEN environment variable is missing.\n");
}

$processor = new TelegramOutboxProcessor($db, $botToken);

// Set up basic signal handling for graceful termination
$keepRunning = true;
if (function_exists('pcntl_signal')) {
    pcntl_async_signals(true);
    pcntl_signal(SIGTERM, function() use (&$keepRunning) {
        $keepRunning = false;
    });
    pcntl_signal(SIGINT, function() use (&$keepRunning) {
        $keepRunning = false;
    });
}

while ($keepRunning) {
    try {
        $processor->processQueue();
    } catch (Exception $e) {
        // Log database or runtime exceptions here
        error_log("Worker error: " . $e->getMessage());
    }

    // Sleep for 1 second before the next batch
    sleep(1);
}