消息推送

app/common/model/MessageNotifications.php

@@ -46,4 +46,12 @@ class MessageNotifications extends Model
      * 不允许自动填充模型字段
      */
     protected $guarded = [];
+
+    // 消息创建
+    const SCENE = [
+        self::SCENE0 => '系统消息',
+        self::SCENE1 => '系统通知'
+    ];
+    const SCENE0 = 0;
+    const SCENE1 = 1;
 }

app/common/service/MessageNotificationsService.php

@@ -75,7 +75,7 @@ class MessageNotificationsService
      */
     public function markAsRead(array|string $msgIds, int $uid)
     {
-        $query = Db::table('message_user_status')->where('uid', $uid);
+        $query = Db::table('message_user_status')->where('uid', $uid)->where('status', 0);
         if (is_array($msgIds)) {
             $query->whereIn('msg_id', $msgIds);
         }
@@ -92,4 +92,39 @@ class MessageNotificationsService
     {
         return User::where('status', 1)->pluck('id')->toArray();
     }
+
+    /**
+     * 批量推送消息
+     */
+    public function pushNotice(array $data): int
+    {
+        // 数据结构：
+        // $data = [
+        //     'sender_id'     => 0,
+        //     'sender_type'   => 'system',
+        //     'title'         => '测试消息 - 标题',
+        //     'content'       => '测试消息 - 内容',
+        //     'scene'         => 1,
+        //     'recipient'     => [],
+        // ];
+        
+        // 写消息
+        $msgId = $this->createMessage([
+            'sender_id'   => $data['sender_id'],
+            'sender_type' => $data['sender_type'],
+            'title'       => $data['title'],
+            'content'     => $data['content'],
+            'scene'       => $data['scene'],
+            'recipient'   => $data['recipient'] ? implode(',', $data['recipient']) : '',
+        ]);
+
+        // 写用户状态
+        $uids = count($data['recipient']) > 0 ? $data['recipient'] : $this->getAllUserIds();
+        $this->addUserStatus($msgId, $uids);
+
+        // 执行推送
+        (new \app\common\utils\WsEvents())->pushNotice($msgId, $data['title'], $data['content'], $data['recipient']);
+
+        return $msgId;
+    }
 }

app/common/utils/WsEvents.php

@@ -222,6 +222,29 @@ class WsEvents
         ));
     }
 
+    /**
+     * 推送消息通知
+     */
+    public function pushNotice(int $msgId, string $title, string $content, array $recipient)
+    {
+        $payload = WsResponse::success('notify.push', [
+            'msg_id'        => $msgId,
+            'title'         => $title,
+            'content'       => $content,
+            'created_at'    => time()
+        ]);
+
+        if (count($recipient) > 0) {
+            // 推送给指定用户
+            foreach ($recipient as $uid) {
+                Gateway::sendToUid($uid, $payload);
+            }
+        } else {
+            // 推送给全体在线用户
+            Gateway::sendToAll($payload);
+        }
+    }
+
     // ================= 辅助方法 =================
     protected function eventToMethod(string $event): string
     {

app/controller/IndexController.php

@@ -6,6 +6,7 @@ use app\common\model\InquiryLog;
 use app\common\model\User;
 use app\common\model\UserRoles;
 use app\common\model\UserSystems;
+use app\common\service\MessageNotificationsService;
 use app\common\traits\DataPermission;
 use app\common\utils\email\MailSend;
 use app\common\utils\scaffold\Generate;
@@ -106,11 +107,21 @@ class IndexController
     public function test()
     {
 
-        $service = new \app\common\service\MessageNotificationsService();
+        // $data = [
-        $udis = $service->getAllUserIds();
+        //     'sender_id'     => 0,
+        //     'sender_type'   => 'system',
+        //     'title'         => '测试消息 - 标题',
+        //     'content'       => '测试消息 - 内容',
+        //     'scene'         => 1,
+        //     'recipient'     => [1],
+        // ];
 
-        var_dump($udis);
+        // // 同步推送
+        // (new MessageNotificationsService())->pushNotice($data);
 
-        return json($udis);
+        // // 异步推送
+        // \Webman\RedisQueue\Redis::send('notice', $data);
+
+        return json($data);
     }
 }

app/queue/redis/Notice.php

@@ -0,0 +1,48 @@
+<?php
+
+namespace app\queue\redis;
+
+use support\exception\BusinessException;
+use Webman\RedisQueue\Consumer;
+
+class Notice implements Consumer
+{
+    // 要消费的队列名
+    public $queue = 'notice';
+
+    // 连接名，对应 plugin/webman/redis-queue/redis.php 里的连接`
+    public $connection = 'default';
+
+    // 消费
+    public function consume($data)
+    {
+        try {
+            (new \app\common\service\MessageNotificationsService())->pushNotice($data);
+            return true;
+        } catch (\Throwable $th) {
+            var_dump('4 ：'.$th->getFile()."|".$th->getLine()."|".$th->getMessage());
+            throw new BusinessException($th->getFile()."|".$th->getLine()."|".$th->getMessage());
+        }
+    }
+
+    // 消费失败回调
+    /* 
+    $package = [
+        'id' => 1357277951, // 消息ID
+        'time' => 1709170510, // 消息时间
+        'delay' => 0, // 延迟时间
+        'attempts' => 2, // 消费次数
+        'queue' => 'send-mail', // 队列名
+        'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息内容
+        'max_attempts' => 5, // 最大重试次数
+        'error' => '错误信息' // 错误信息
+    ]
+    */
+    public function onConsumeFailure(\Throwable $e, $package)
+    {
+        echo "consume failure\n";
+        echo $e->getMessage() . "\n";
+        // 无需反序列化
+        var_export($package); 
+    }
+}