Handle Telegram Bot API Rate Limits and 429 Errors in PHP

<?php

declare(strict_types=1);

class TelegramRateLimitException extends Exception 
{
    private int $retryAfter;

    public function __construct(int $retryAfter, string $message = "Rate limit exceeded") 
    {
        parent::__construct($message, 429);
        $this->retryAfter = $retryAfter;
    }

    public function getRetryAfter(): int 
    {
        return $this->retryAfter;
    }
}

class TelegramClient 
{
    private string $token;

    public function __construct()
    {
        $this->token = (string) getenv('TELEGRAM_BOT_TOKEN');
        if ($this->token === '') {
            throw new RuntimeException('TELEGRAM_BOT_TOKEN environment variable is not set.');
        }
    }

    public function sendRequest(string $method, array $payload): array
    {
        $url = "https://api.telegram.org/bot{$this->token}/{$method}";
        $jsonPayload = json_encode($payload);

        if ($jsonPayload === false) {
            throw new InvalidArgumentException('Failed to encode payload to JSON: ' . json_last_error_msg());
        }

        $ch = curl_init($url);
        if ($ch === false) {
            throw new RuntimeException('Failed to initialize cURL.');
        }

        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        curl_setopt($ch, CURLOPT_POST, true);
        curl_setopt($ch, CURLOPT_POSTFIELDS, $jsonPayload);
        curl_setopt($ch, CURLOPT_HTTPHEADER, [
            'Content-Type: application/json',
        ]);
        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) {
            throw new RuntimeException("cURL request failed: {$curlError}");
        }

        $data = json_decode($response, true);
        if (json_last_error() !== JSON_ERROR_NONE) {
            throw new RuntimeException('Failed to parse Telegram API response as JSON.');
        }

        if ($httpCode === 429) {
            $retryAfter = $data['parameters']['retry_after'] ?? 1;
            throw new TelegramRateLimitException((int) $retryAfter);
        }

        if (isset($data['ok']) && $data['ok'] === false) {
            $description = $data['description'] ?? 'Unknown error';
            throw new RuntimeException("Telegram API Error: {$description} (Code: {$httpCode})");
        }

        return $data;
    }
}

CREATE TABLE telegram_message_queue (
    id INT AUTO_INCREMENT PRIMARY KEY,
    chat_id BIGINT NOT NULL,
    payload JSON NOT NULL,
    status VARCHAR(20) DEFAULT 'pending', -- pending, sent, failed, rate_limited
    retry_after_time DATETIME NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_status_retry (status, retry_after_time),
    INDEX idx_chat_status (chat_id, status)
);

<?php

declare(strict_types=1);

// Assuming PDO connection $pdo and TelegramClient $client are initialized

$client = new TelegramClient();
$lastSentToChat = []; // Tracks last send timestamp per chat_id

while (true) {
    // Fetch pending messages that are not locked by a retry delay
    $stmt = $pdo->prepare("
        SELECT id, chat_id, payload 
        FROM telegram_message_queue 
        WHERE status = 'pending' 
          AND (retry_after_time IS NULL OR retry_after_time <= NOW())
        ORDER BY id ASC 
        LIMIT 30
    ");
    $stmt->execute();
    $messages = $stmt->fetchAll(PDO::FETCH_ASSOC);

    if (empty($messages)) {
        sleep(1); // No messages, idle
        continue;
    }

    $sentThisBatch = 0;

    foreach ($messages as $msg) {
        $chatId = (int) $msg['chat_id'];
        $payload = json_decode($msg['payload'], true);
        $now = microtime(true);

        // Enforce the 3-second delay per chat to prevent hitting the 20 msg/min limit
        if (isset($lastSentToChat[$chatId]) && ($now - $lastSentToChat[$chatId]) < 3.0) {
            continue; // Skip this message for this iteration to let other chats process
        }

        try {
            $client->sendRequest('sendMessage', $payload);

            // Update status to sent
            $update = $pdo->prepare("UPDATE telegram_message_queue SET status = 'sent' WHERE id = ?");
            $update->execute([$msg['id']]);

            $lastSentToChat[$chatId] = microtime(true);
            $sentThisBatch++;

            // Enforce global limit: max 30 messages per second
            if ($sentThisBatch >= 30) {
                sleep(1);
                $sentThisBatch = 0;
            }
        } catch (TelegramRateLimitException $e) {
            $retryAfter = $e->getRetryAfter();

            // Reschedule this message and set status back to pending with a delay
            $delayUntil = (new DateTime())->modify("+{$retryAfter} seconds")->format('Y-m-d H:i:s');
            $update = $pdo->prepare("
                UPDATE telegram_message_queue 
                SET status = 'pending', retry_after_time = ? 
                WHERE id = ?
            ");
            $update->execute([$delayUntil, $msg['id']]);

            // Pause the worker for the duration of the rate limit to avoid spamming the API
            sleep($retryAfter);
            break; // Break the foreach loop to fetch fresh data after the sleep
        } catch (Exception $e) {
            // Handle permanent failures (e.g., user blocked bot)
            $update = $pdo->prepare("UPDATE telegram_message_queue SET status = 'failed' WHERE id = ?");
            $update->execute([$msg['id']]);
        }
    }

    // Clean up memory for inactive chats
    if (count($lastSentToChat) > 1000) {
        $lastSentToChat = array_slice($lastSentToChat, -500, null, true);
    }

    usleep(100000); // 100ms sleep between batches
}