PHP code example of afaztech / reactor-broadcast

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

afaztech / reactor-broadcast example snippets

// 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
    {
    }
}

use AfazTech\ReactorBroadcast\Contracts\BroadcastManagerInterface;

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

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

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

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

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

$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',
// ]

$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.

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."
        );
    }
}

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']}"
        );
    }
}

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);
    }
}

use AfazTech\ReactorBroadcast\Contracts\RecipientProviderInterface;

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

bash
php reactor.php migrate

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