نظام النشر والاشتراك في Redis

1. الناشرون - إرسال الرسائل إلى القنوات
2. القنوات - تدفقات الرسائل المسماة
3. المشتركون - تلقي الرسائل من القنوات

الخصائص الرئيسية:
✓ النار والنسيان: الرسائل لا تُحفظ
✓ الوقت الفعلي: التسليم الفوري للمشتركين النشطين
✓ فك الاقتران: الناشرون لا يعرفون عن المشتركين
✗ لا يوجد سجل رسائل: المشتركون الجدد يفقدون الرسائل القديمة

# الاشتراك في قناة (يحجب حتى استلام رسالة)
SUBSCRIBE notifications

# الاشتراك في قنوات متعددة
SUBSCRIBE notifications alerts updates

# نشر رسالة إلى قناة
PUBLISH notifications "New user registered"
# يعيد: (integer) 2  # عدد المشتركين الذين تلقوا الرسالة

# إلغاء الاشتراك من القنوات
UNSUBSCRIBE notifications
UNSUBSCRIBE  # إلغاء الاشتراك من الكل

use Illuminate\Support\Facades\Redis;

// نشر رسالة بسيطة
Redis::publish('notifications', 'User registered');

// نشر بيانات JSON
$data = [
    'event' => 'user.registered',
    'user_id' => 123,
    'email' => 'user@example.com',
    'timestamp' => time()
];
Redis::publish('notifications', json_encode($data));

// النشر إلى قنوات متعددة
$channels = ['notifications', 'admin-alerts', 'analytics'];
foreach ($channels as $channel) {
    Redis::publish($channel, json_encode($data));
}

// الاشتراك في قناة واحدة
Redis::subscribe(['notifications'], function ($message) {
    echo "تم الاستلام: {$message}\n";

    $data = json_decode($message, true);
    // معالجة الرسالة
    logger()->info('تم استلام الإشعار', $data);
});

// الاشتراك في قنوات متعددة
Redis::subscribe(['notifications', 'alerts'], function ($message, $channel) {
    echo "القناة: {$channel}, الرسالة: {$message}\n";
});

// Redis CLI
PSUBSCRIBE user.*        # يطابق user.created, user.updated, user.deleted
PSUBSCRIBE order.*.paid  # يطابق order.123.paid, order.456.paid
PSUBSCRIBE *             # الاشتراك في جميع القنوات (استخدم بحذر!)

// تنفيذ Laravel
Redis::psubscribe(['user.*', 'order.*'], function ($message, $channel) {
    echo "مطابقة النمط - القناة: {$channel}, الرسالة: {$message}\n";

    // استخراج نوع الحدث من اسم القناة
    $parts = explode('.', $channel);
    $entity = $parts[0];  // user, order
    $action = $parts[1] ?? null;  // created, updated, deleted

    match($channel) {
        'user.created' => $this->handleUserCreated(json_decode($message, true)),
        'user.updated' => $this->handleUserUpdated(json_decode($message, true)),
        'order.completed' => $this->handleOrderCompleted(json_decode($message, true)),
        default => logger()->debug("قناة غير معالجة: {$channel}")
    };
});

namespace App\Services;

use Illuminate\Support\Facades\Redis;

class NotificationService
{
    public function sendNotification(int $userId, string $type, array $data)
    {
        $notification = [
            'user_id' => $userId,
            'type' => $type,
            'data' => $data,
            'timestamp' => now()->toIso8601String(),
            'id' => uniqid('notif_')
        ];

        // النشر إلى قناة خاصة بالمستخدم
        Redis::publish("notifications.user.{$userId}", json_encode($notification));

        // أيضاً النشر إلى قناة خاصة بالنوع للتحليلات
        Redis::publish("notifications.type.{$type}", json_encode($notification));

        return $notification['id'];
    }

    public function broadcastToAll(string $type, array $data)
    {
        $notification = [
            'type' => $type,
            'data' => $data,
            'timestamp' => now()->toIso8601String()
        ];

        Redis::publish('notifications.broadcast', json_encode($notification));
    }
}

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\Redis;

class NotificationSubscriber extends Command
{
    protected $signature = 'notifications:subscribe';
    protected $description = 'الاشتراك في قنوات الإشعارات';

    public function handle()
    {
        $this->info('الاشتراك في قنوات الإشعارات...');

        Redis::psubscribe(['notifications.*'], function ($message, $channel) {
            $this->processNotification($message, $channel);
        });
    }

    private function processNotification(string $message, string $channel)
    {
        $data = json_decode($message, true);

        // تسجيل الإشعار
        logger()->info("تم استلام الإشعار", [
            'channel' => $channel,
            'data' => $data
        ]);

        // الإرسال عبر قنوات مختلفة بناءً على النوع
        if (str_contains($channel, 'user')) {
            $this->sendWebSocketNotification($data);
            $this->sendEmailIfImportant($data);
        }

        if (str_contains($channel, 'broadcast')) {
            $this->sendToAllWebSockets($data);
        }
    }
}

class ChatController extends Controller
{
    public function sendMessage(Request $request)
    {
        $validated = $request->validate([
            'room_id' => 'required|string',
            'message' => 'required|string|max:1000'
        ]);

        $messageData = [
            'user_id' => auth()->id(),
            'username' => auth()->user()->name,
            'message' => $validated['message'],
            'timestamp' => now()->toIso8601String()
        ];

        // النشر إلى قناة الغرفة
        Redis::publish(
            "chat.room.{$validated['room_id']}",
            json_encode($messageData)
        );

        // تخزين الرسالة في مجموعة مرتبة Redis للسجل (اختياري)
        Redis::zadd(
            "chat:history:{$validated['room_id']}",
            time(),
            json_encode($messageData)
        );

        return response()->json(['status' => 'sent']);
    }
}

// التشغيل كعملية منفصلة: php artisan chat:subscribe
class ChatSubscriber extends Command
{
    protected $signature = 'chat:subscribe';

    public function handle()
    {
        Redis::psubscribe(['chat.room.*'], function ($message, $channel) {
            // استخراج معرف الغرفة من اسم القناة
            preg_match('/chat\.room\.(\w+)/', $channel, $matches);
            $roomId = $matches[1] ?? null;

            if (!$roomId) return;

            // البث إلى عملاء WebSocket في هذه الغرفة
            $this->broadcastToWebSocket($roomId, json_decode($message, true));
        });
    }

    private function broadcastToWebSocket(string $roomId, array $message)
    {
        // الإرسال إلى خادم WebSocket (مثل Laravel Echo Server, Socket.io)
        event(new ChatMessageReceived($roomId, $message));
    }
}

// سرد القنوات النشطة مع مشترك واحد على الأقل
PUBSUB CHANNELS
PUBSUB CHANNELS user.*  # مطابقة الأنماط

// عد المشتركين في قناة
PUBSUB NUMSUB notifications alerts
# يعيد: 1) "notifications" 2) (integer) 5 3) "alerts" 4) (integer) 3

// عد الأنماط الفريدة المشترك فيها
PUBSUB NUMPAT
# يعيد: (integer) 7

Redis Pub/Sub:
✓ خفيف الوزن، اتصال مباشر
✓ لا يوجد عبء استمرارية
✓ رائع للمراسلة من خادم إلى خادم
✗ لا توجد مصادقة عميل مدمجة
✗ لا يوجد سجل رسائل
✗ تكامل WebSocket يدوي

Laravel Broadcasting (يستخدم Pub/Sub داخلياً):
✓ دعم WebSocket مدمج (Pusher، Echo)
✓ مصادقة عميل تلقائية
✓ تفويض القنوات
✓ قنوات الحضور
✗ إعداد أكثر تعقيداً
✗ تبعيات إضافية