README

Broadcast (send / copy / forward) with progress statistics for Reactor bots.

This package is built specifically for the Reactor PHP framework. It requires afaztech/reactor-framework ^0.5.0 and will not work standalone.

Overview

Reactor Broadcast delivers mass messages to every user of a Telegram bot without blocking the bot's main loop. Jobs are processed asynchronously through the Reactor queue, progress is tracked per recipient, and the whole flow can be paused, resumed or cancelled at any time.

Three delivery strategies are available:

Strategy	Description
`send`	Send a fresh text message (optional inline/reply keyboard).
`copy`	Copy an existing message from any chat. No "Forwarded from" label.
`forward`	Forward an existing message from any chat, with the "Forwarded from" label.

Every recipient is recorded in a dedicated table with one of four states: pending, sent, failed, or blocked. Aggregate counters are updated live so the operator can watch progress in real time.

Requirements

PHP >= 8.1
afaztech/reactor-framework ^0.5.0
A Reactor application with migrations and the queue worker configured
A running queue worker: php reactor.php queue:work --queue=broadcast

Installation

Install the package through Composer. The service provider is auto-discovered through Composer metadata (extra.reactor.providers), so no manual registration is required.

composer require afaztech/reactor-broadcast

Run the migrations to create the broadcasts and broadcast_recipients tables:

php reactor.php migrate

Start the queue worker that will process the broadcasts:

php reactor.php queue:work --queue=broadcast

In production, run the worker under a supervisor (systemd, supervisord, ...) so it restarts automatically after a crash.

Configuration

The package is configured programmatically through the fluent BroadcastConfig value object. There is no config file to publish.

Bind BroadcastConfig in your application's service provider. The package registers an empty default when the application does not override it, so a bare setup works out of the box:

// app/Providers/AppServiceProvider.php

namespace App\Providers;

use Reactor\Core\Container;
use Reactor\Contracts\ServiceProviderInterface;
use AfazTech\ReactorBroadcast\Support\BroadcastConfig;

class AppServiceProvider implements ServiceProviderInterface
{
    public function register(Container $container): void
    {
        $container->singleton(BroadcastConfig::class, function () {
            return (new BroadcastConfig())
                ->table('users')
                ->column('user_id')
                ->onlyActive('status', 1)
                ->onlyRecent('last_interaction_at', 14)
                ->queue('broadcast');
        });
    }

    public function boot(Container $container): void
    {
    }
}

Available setters

Method	Default	Description
`table(string $name)`	`users`	Table that stores your users.
`column(string $name)`	`user_id`	Column holding the Telegram user ID.
`onlyActive(string $column, mixed $value)`	disabled	Only include rows where `$column = $value`.
`withoutActiveFilter()`	—	Disable the active-user filter.
`onlyRecent(string $column, int $days)`	disabled	Only include rows newer than `$days`.
`withoutRecentFilter()`	—	Disable the recent-interaction filter.
`where(string $col, string $op, mixed $val)`	—	Add a raw WHERE clause. Callable multiple times.
`withoutRawFilters()`	—	Clear all raw filters.
`chunkSize(int $size)`	`500`	Recipients per batch when collecting IDs.
`batchDelayMs(int $ms)`	`200`	Delay between batches in milliseconds.
`cancelCheckEvery(int $n)`	`50`	Check pause/cancel status every N messages.
`queue(string $name)`	`broadcast`	Queue name used by the processing job.

Usage

Getting the manager

use AfazTech\ReactorBroadcast\Contracts\BroadcastManagerInterface;

$manager = $container->get(BroadcastManagerInterface::class);

Start a text broadcast

$id = $manager->start('send', [
    'text'     => 'Hello everyone!',
    'keyboard' => [
        'inline_keyboard' => [[
            ['text' => 'Learn more', 'url' => 'https://example.com'],
        ]],
    ],
    'extra' => ['parse_mode' => 'HTML'],
]);

Copy an existing message

The message will be delivered to every recipient without the "Forwarded from" label.

$id = $manager->start('copy', [
    'from_chat_id' => '@my_channel',
    'message_id'   => 1234,
]);

Forward an existing message

The message will be delivered with the "Forwarded from" label.

$id = $manager->start('forward', [
    'from_chat_id' => -1001234567890,
    'message_id'   => 5678,
]);

Send to a custom list of recipients

Pass $recipientIds to bypass the configured RecipientProviderInterface and target a specific set of users.

$id = $manager->start(
    strategy: 'send',
    payload: ['text' => 'Special announcement'],
    recipientIds: [111111, 222222, 333333],
);

Get progress

$stats = $manager->stats($id);

// [
//   'id'          => 1,
//   'strategy'    => 'send',
//   'status'      => 'running',  // pending | running | paused | completed | cancelled
//   'total'       => 1500,
//   'sent'        => 870,
//   'failed'      => 12,
//   'blocked'     => 43,
//   'pending'     => 575,
//   'started_at'  => '2026-10-09T10:00:00+00:00',
//   'finished_at' => null,
//   'created_at'  => '2026-10-09T09:59:55+00:00',
// ]

Control a running broadcast

$manager->pause($id);   // Temporarily halt. Progress is preserved.
$manager->resume($id);  // Continue from the first pending recipient.
$manager->cancel($id);  // Stop permanently. Already-sent messages are kept.

A paused broadcast keeps its progress. Resuming re-queues the processing job and continues from the first pending recipient, so no user receives a message twice.

Status lifecycle

pending -> running -> completed
pending -> running -> cancelled
pending -> running -> paused -> running -> completed
pending -> running -> paused -> cancelled

Example handlers

Two ready-to-use handlers for a test bot. No access control is applied — add an admin check before deploying to production.

Start a broadcast by replying to a message

<?php
namespace App\Handlers\Admin;

use Neili\Client;
use Reactor\Attributes\Text;
use Reactor\Attributes\OnUpdate;
use Reactor\Contracts\LanguageInterface;
use Reactor\Contracts\LoggerInterface;
use App\Contracts\Repository\UserRepositoryInterface;
use App\Handlers\BaseHandler;
use AfazTech\ReactorBroadcast\Contracts\BroadcastManagerInterface;

#[Text(name: '/send', isCommand: true, priority: 100)]
#[OnUpdate('message')]
class SendBroadcastHandler extends BaseHandler
{
    public function __construct(
        Client $client,
        LanguageInterface $language,
        LoggerInterface $logger,
        UserRepositoryInterface $userRepository,
        private BroadcastManagerInterface $broadcast,
    ) {
        parent::__construct($client, $language, $logger, $userRepository);
    }

    protected function handle(array $params = []): void
    {
        $message = $this->update['message'] ?? [];
        $replyTo = $message['reply_to_message'] ?? null;

        if (!$replyTo || !isset($replyTo['message_id'])) {
            $this->reply(
                "Reply to the message you want to broadcast, then send /send."
            );
            return;
        }

        try {
            $broadcastId = $this->broadcast->start('copy', [
                'from_chat_id' => $message['chat']['id'],
                'message_id'   => (int) $replyTo['message_id'],
            ]);
        } catch (\Throwable $e) {
            $this->logger->error('Broadcast start failed', [
                'error' => $e->getMessage(),
            ]);
            $this->reply('Failed to start broadcast: ' . $e->getMessage());
            return;
        }

        $stats = $this->broadcast->stats($broadcastId);

        $this->reply(
            "Broadcast #{$broadcastId} queued.\n"
            . "Recipients: {$stats['total']}\n\n"
            . "Send /stats {$broadcastId} to watch progress."
        );
    }
}

Check broadcast progress

<?php
namespace App\Handlers\Admin;

use Neili\Client;
use Reactor\Attributes\Text;
use Reactor\Attributes\OnUpdate;
use Reactor\Contracts\LanguageInterface;
use Reactor\Contracts\LoggerInterface;
use App\Contracts\Repository\UserRepositoryInterface;
use App\Handlers\BaseHandler;
use AfazTech\ReactorBroadcast\Contracts\BroadcastManagerInterface;

#[Text(name: '/stats', isCommand: true, priority: 100)]
#[OnUpdate('message')]
class BroadcastStatsHandler extends BaseHandler
{
    public function __construct(
        Client $client,
        LanguageInterface $language,
        LoggerInterface $logger,
        UserRepositoryInterface $userRepository,
        private BroadcastManagerInterface $broadcast,
    ) {
        parent::__construct($client, $language, $logger, $userRepository);
    }

    protected function handle(array $params = []): void
    {
        $id = $params[0] ?? null;
        if (!is_string($id) || !ctype_digit($id)) {
            $this->reply('Usage: /stats <broadcast_id>');
            return;
        }

        try {
            $stats = $this->broadcast->stats((int) $id);
        } catch (\Throwable $e) {
            $this->reply("Broadcast #{$id} not found.");
            return;
        }

        $percent = $stats['total'] > 0
            ? round(($stats['sent'] + $stats['failed'] + $stats['blocked']) * 100 / $stats['total'], 1)
            : 0.0;

        $this->reply(
            "Broadcast #{$stats['id']}\n"
            . "Status: {$stats['status']}\n"
            . "Progress: {$percent}%\n\n"
            . "Total:    {$stats['total']}\n"
            . "Sent:     {$stats['sent']}\n"
            . "Failed:   {$stats['failed']}\n"
            . "Blocked:  {$stats['blocked']}\n"
            . "Pending:  {$stats['pending']}"
        );
    }
}

Custom recipient providers

If your users are not stored in a single table, or require logic that the ConfigurableRecipientProvider cannot express, implement RecipientProviderInterface and bind it in your application's service provider. The binding overrides the package default.

<?php
namespace App\Broadcast;

use AfazTech\ReactorBroadcast\Contracts\RecipientProviderInterface;

class ActivePayingUsersProvider implements RecipientProviderInterface
{
    public function __construct(private \PDO $pdo) {}

    public function count(): int
    {
        return (int) $this->pdo
            ->query("SELECT COUNT(*) FROM premium_users WHERE expires_at > NOW()")
            ->fetchColumn();
    }

    public function chunk(int $offset, int $limit): array
    {
        $stmt = $this->pdo->prepare(
            "SELECT telegram_id FROM premium_users WHERE expires_at > NOW() ORDER BY telegram_id LIMIT :limit OFFSET :offset"
        );
        $stmt->bindValue(':limit', $limit, \PDO::PARAM_INT);
        $stmt->bindValue(':offset', $offset, \PDO::PARAM_INT);
        $stmt->execute();
        return $stmt->fetchAll(\PDO::FETCH_COLUMN);
    }
}

Then in AppServiceProvider:

use AfazTech\ReactorBroadcast\Contracts\RecipientProviderInterface;

$container->singleton(RecipientProviderInterface::class, function ($c) {
    return new \App\Broadcast\ActivePayingUsersProvider($c->get(\PDO::class));
});

Error handling

The processing job classifies every failure automatically:

Exception	Behavior
`RateLimitException`	Waits exactly `retry_after` seconds, then retries the same recipient.
`PermanentException` with "blocked" / "deactivated" / "kicked"	Marks recipient as `blocked`. Increments the `blocked` counter.
`PermanentException` (other)	Marks recipient as `failed`. Stores the error message.
`TransientException`	Marks recipient as `failed`.
Any other `Throwable`	Marks recipient as `failed` and logs the error.

Testing the package locally

To develop against a local checkout, add a path repository in your application's composer.json:

"repositories": [
    {
        "type": "path",
        "url": "../reactor-broadcast",
        "options": { "symlink": true }
    }
],
"require": {
    "afaztech/reactor-broadcast": "@dev"
}

The framework must also be a path repository or a tagged release of ^0.5.0. When the framework is checked out locally, add a branch-alias to its composer.json so dev-main is treated as 0.5.x-dev:

"extra": {
    "branch-alias": {
        "dev-main": "0.5.x-dev"
    }
}

License

MIT. See LICENSE.

afaztech / reactor-broadcast

Package info

Security

Statistics