molong/laravel_swoole
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

更新

Browse Source
This commit is contained in:
molong
2026-02-18 19:41:03 +08:00
parent a0c2350662
commit 6543e2ccdd
18 changed files with 4885 additions and 1196 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
app/Console/Commands/RetryUnsentNotifications.php Normal file
View File

@@ -0,0 +1,74 @@
<?php
namespace App\Console\Commands;
use Illuminate\Console\Command;
use App\Services\System\NotificationService;
use Illuminate\Support\Facades\Log;
class RetryUnsentNotifications extends Command
{
/**
* The name and signature of the console command.
*
* @var string
*/
protected $signature = 'notifications:retry-unsent {--limit=100 : Maximum number of notifications to retry}';
/**
* The console command description.
*
* @var string
*/
protected $description = 'Retry sending unsent notifications via WebSocket';
/**
* @var NotificationService
*/
protected $notificationService;
/**
* RetryUnsentNotifications constructor
*/
public function __construct(NotificationService $notificationService)
{
parent::__construct();
$this->notificationService = $notificationService;
}
/**
* Execute the console command.
*/
public function handle(): int
{
$limit = (int) $this->option('limit');
$this->info('开始重试未发送的通知...');
$this->info("最大处理数量: {$limit}");
try {
$sentCount = $this->notificationService->retryUnsentNotifications($limit);
if ($sentCount > 0) {
$this->info("成功发送 {$sentCount} 条通知");
} else {
$this->info('没有需要重试的通知');
}
Log::info('重试未发送通知完成', [
'sent_count' => $sentCount,
'limit' => $limit
]);
return self::SUCCESS;
} catch (\Exception $e) {
$this->error('重试未发送通知失败: ' . $e->getMessage());
Log::error('重试未发送通知失败', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
return self::FAILURE;
}
}
}

361
app/Http/Controllers/System/Admin/Notification.php Normal file
View File

@@ -0,0 +1,361 @@
<?php
namespace App\Http\Controllers\System\Admin;
use Illuminate\Http\Request;
use Illuminate\Http\JsonResponse;
use App\Services\System\NotificationService;
use App\Http\Controllers\Controller;
class Notification extends Controller
{
/**
* @var NotificationService
*/
protected $notificationService;
/**
* Notification constructor
*/
public function __construct(NotificationService $notificationService)
{
$this->notificationService = $notificationService;
}
/**
* 获取通知列表
*
* @param Request $request
* @return JsonResponse
*/
public function index(Request $request): JsonResponse
{
$params = $request->all();
// 如果没有指定user_id，使用当前登录用户
if (empty($params['user_id'])) {
$params['user_id'] = auth('admin')->id();
}
$result = $this->notificationService->getList($params);
return response()->json([
'code' => 200,
'message' => 'success',
'data' => $result
]);
}
/**
* 获取未读通知
*
* @param Request $request
* @return JsonResponse
*/
public function unread(Request $request): JsonResponse
{
$userId = auth('admin')->id();
$limit = $request->input('limit', 10);
$notifications = $this->notificationService->getUnreadNotifications($userId, $limit);
return response()->json([
'code' => 200,
'message' => 'success',
'data' => [
'list' => $notifications
]
]);
}
/**
* 获取未读通知数量
*
* @return JsonResponse
*/
public function unreadCount(): JsonResponse
{
$userId = auth('admin')->id();
$count = $this->notificationService->getUnreadCount($userId);
return response()->json([
'code' => 200,
'message' => 'success',
'data' => [
'count' => $count
]
]);
}
/**
* 获取通知详情
*
* @param int $id
* @return JsonResponse
*/
public function show(int $id): JsonResponse
{
$notification = $this->notificationService->getById($id);
if (!$notification) {
return response()->json([
'code' => 404,
'message' => 'Notification not found',
'data' => null
], 404);
}
// 检查权限
if ($notification->user_id !== auth('admin')->id()) {
return response()->json([
'code' => 403,
'message' => 'Access denied',
'data' => null
], 403);
}
return response()->json([
'code' => 200,
'message' => 'success',
'data' => $notification
]);
}
/**
* 标记通知为已读
*
* @param Request $request
* @param int $id
* @return JsonResponse
*/
public function markAsRead(Request $request, int $id): JsonResponse
{
$userId = auth('admin')->id();
$result = $this->notificationService->markAsRead($id, $userId);
if (!$result) {
return response()->json([
'code' => 404,
'message' => 'Notification not found or access denied',
'data' => null
], 404);
}
return response()->json([
'code' => 200,
'message' => 'Notification marked as read',
'data' => null
]);
}
/**
* 批量标记通知为已读
*
* @param Request $request
* @return JsonResponse
*/
public function batchMarkAsRead(Request $request): JsonResponse
{
$request->validate([
'ids' => 'required|array',
'ids.*' => 'integer'
]);
$userId = auth('admin')->id();
$ids = $request->input('ids');
$count = $this->notificationService->batchMarkAsRead($ids, $userId);
return response()->json([
'code' => 200,
'message' => 'Notifications marked as read',
'data' => [
'count' => $count
]
]);
}
/**
* 标记所有通知为已读
*
* @return JsonResponse
*/
public function markAllAsRead(): JsonResponse
{
$userId = auth('admin')->id();
$count = $this->notificationService->markAllAsRead($userId);
return response()->json([
'code' => 200,
'message' => 'All notifications marked as read',
'data' => [
'count' => $count
]
]);
}
/**
* 删除通知
*
* @param int $id
* @return JsonResponse
*/
public function destroy(int $id): JsonResponse
{
$userId = auth('admin')->id();
$result = $this->notificationService->delete($id, $userId);
if (!$result) {
return response()->json([
'code' => 404,
'message' => 'Notification not found or access denied',
'data' => null
], 404);
}
return response()->json([
'code' => 200,
'message' => 'Notification deleted',
'data' => null
]);
}
/**
* 批量删除通知
*
* @param Request $request
* @return JsonResponse
*/
public function batchDelete(Request $request): JsonResponse
{
$request->validate([
'ids' => 'required|array',
'ids.*' => 'integer'
]);
$userId = auth('admin')->id();
$ids = $request->input('ids');
$count = $this->notificationService->batchDelete($ids, $userId);
return response()->json([
'code' => 200,
'message' => 'Notifications deleted',
'data' => [
'count' => $count
]
]);
}
/**
* 清空已读通知
*
* @return JsonResponse
*/
public function clearRead(): JsonResponse
{
$userId = auth('admin')->id();
$count = $this->notificationService->clearReadNotifications($userId);
return response()->json([
'code' => 200,
'message' => 'Read notifications cleared',
'data' => [
'count' => $count
]
]);
}
/**
* 获取通知统计
*
* @return JsonResponse
*/
public function statistics(): JsonResponse
{
$userId = auth('admin')->id();
$stats = $this->notificationService->getStatistics($userId);
return response()->json([
'code' => 200,
'message' => 'success',
'data' => $stats
]);
}
/**
* 发送通知（管理员功能）
*
* @param Request $request
* @return JsonResponse
*/
public function send(Request $request): JsonResponse
{
$request->validate([
'user_ids' => 'nullable|array',
'user_ids.*' => 'integer',
'title' => 'required|string|max:200',
'content' => 'required|string',
'type' => 'required|string|in:info,success,warning,error,task,system',
'category' => 'nullable|string|in:system,task,message,reminder,announcement',
'data' => 'nullable|array',
'action_type' => 'nullable|string|in:link,modal,none',
'action_data' => 'nullable|array',
]);
$userIds = $request->input('user_ids');
$title = $request->input('title');
$content = $request->input('content');
$type = $request->input('type', 'info');
$category = $request->input('category', 'system');
$extraData = $request->input('data', []);
// 如果没有指定user_ids，则发送给所有用户
if (empty($userIds)) {
$result = $this->notificationService->broadcast(
$title,
$content,
$type,
$category,
$extraData
);
} else {
$result = $this->notificationService->sendToUsers(
$userIds,
$title,
$content,
$type,
$category,
$extraData
);
}
return response()->json([
'code' => 200,
'message' => 'Notification sent successfully',
'data' => [
'count' => count($result)
]
]);
}
/**
* 重试发送未发送的通知（管理员功能）
*
* @param Request $request
* @return JsonResponse
*/
public function retryUnsent(Request $request): JsonResponse
{
$limit = $request->input('limit', 100);
$count = $this->notificationService->retryUnsentNotifications($limit);
return response()->json([
'code' => 200,
'message' => 'Unsent notifications retried',
'data' => [
'count' => $count
]
]);
}
}

154
app/Models/System/Notification.php Normal file
View File

@@ -0,0 +1,154 @@
<?php
namespace App\Models\System;
use App\Traits\ModelTrait;
use Illuminate\Database\Eloquent\Model;
use Illuminate\Database\Eloquent\SoftDeletes;
use Illuminate\Database\Eloquent\Relations\BelongsTo;
class Notification extends Model
{
use ModelTrait, SoftDeletes;
protected $table = 'system_notifications';
protected $fillable = [
'user_id',
'title',
'content',
'type',
'category',
'data',
'action_type',
'action_data',
'is_read',
'read_at',
'sent_via_websocket',
'sent_at',
'retry_count',
];
protected $casts = [
'data' => 'array',
'is_read' => 'boolean',
'sent_via_websocket' => 'boolean',
'read_at' => 'datetime',
'sent_at' => 'datetime',
'retry_count' => 'integer',
];
/**
* 通知类型常量
*/
const TYPE_INFO = 'info';
const TYPE_SUCCESS = 'success';
const TYPE_WARNING = 'warning';
const TYPE_ERROR = 'error';
const TYPE_TASK = 'task';
const TYPE_SYSTEM = 'system';
/**
* 通知分类常量
*/
const CATEGORY_SYSTEM = 'system';
const CATEGORY_TASK = 'task';
const CATEGORY_MESSAGE = 'message';
const CATEGORY_REMINDER = 'reminder';
const CATEGORY_ANNOUNCEMENT = 'announcement';
/**
* 操作类型常量
*/
const ACTION_LINK = 'link';
const ACTION_MODAL = 'modal';
const ACTION_NONE = 'none';
/**
* 关联用户
*/
public function user(): BelongsTo
{
return $this->belongsTo(\App\Models\Auth\User::class, 'user_id');
}
/**
* 标记为已读
*/
public function markAsRead(): bool
{
$this->is_read = true;
$this->read_at = now();
return $this->save();
}
/**
* 标记为未读
*/
public function markAsUnread(): bool
{
$this->is_read = false;
$this->read_at = null;
return $this->save();
}
/**
* 标记已通过WebSocket发送
*/
public function markAsSent(): bool
{
$this->sent_via_websocket = true;
$this->sent_at = now();
return $this->save();
}
/**
* 增加重试次数
*/
public function incrementRetry(): bool
{
$this->increment('retry_count');
return true;
}
/**
* 获取通知类型选项
*/
public static function getTypeOptions(): array
{
return [
self::TYPE_INFO => '信息',
self::TYPE_SUCCESS => '成功',
self::TYPE_WARNING => '警告',
self::TYPE_ERROR => '错误',
self::TYPE_TASK => '任务',
self::TYPE_SYSTEM => '系统',
];
}
/**
* 获取通知分类选项
*/
public static function getCategoryOptions(): array
{
return [
self::CATEGORY_SYSTEM => '系统通知',
self::CATEGORY_TASK => '任务通知',
self::CATEGORY_MESSAGE => '消息通知',
self::CATEGORY_REMINDER => '提醒通知',
self::CATEGORY_ANNOUNCEMENT => '公告通知',
];
}
/**
* 获取操作类型选项
*/
public static function getActionTypeOptions(): array
{
return [
self::ACTION_LINK => '跳转链接',
self::ACTION_MODAL => '弹窗显示',
self::ACTION_NONE => '无操作',
];
}
}

548
app/Services/System/NotificationService.php Normal file
View File

@@ -0,0 +1,548 @@
<?php
namespace App\Services\System;
use App\Models\System\Notification;
use App\Services\WebSocket\WebSocketService;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
class NotificationService
{
/**
* @var WebSocketService
*/
protected $webSocketService;
/**
* NotificationService constructor
*/
public function __construct()
{
$this->webSocketService = app(WebSocketService::class);
}
/**
* 获取通知列表
*
* @param array $params
* @return array
*/
public function getList(array $params): array
{
$query = Notification::query();
// 过滤用户ID
if (!empty($params['user_id'])) {
$query->where('user_id', $params['user_id']);
}
// 关键字搜索
if (!empty($params['keyword'])) {
$query->where(function ($q) use ($params) {
$q->where('title', 'like', '%' . $params['keyword'] . '%')
->orWhere('content', 'like', '%' . $params['keyword'] . '%');
});
}
// 过滤阅读状态
if (isset($params['is_read']) && $params['is_read'] !== '') {
$query->where('is_read', $params['is_read']);
}
// 过滤通知类型
if (!empty($params['type'])) {
$query->where('type', $params['type']);
}
// 过滤通知分类
if (!empty($params['category'])) {
$query->where('category', $params['category']);
}
// 日期范围
if (!empty($params['start_date'])) {
$query->where('created_at', '>=', $params['start_date']);
}
if (!empty($params['end_date'])) {
$query->where('created_at', '<=', $params['end_date']);
}
$query->orderBy('created_at', 'desc');
$pageSize = $params['page_size'] ?? 20;
$list = $query->paginate($pageSize);
return [
'list' => $list->items(),
'total' => $list->total(),
'page' => $list->currentPage(),
'page_size' => $list->perPage(),
];
}
/**
* 获取未读通知列表
*
* @param int $userId
* @param int $limit
* @return array
*/
public function getUnreadNotifications(int $userId, int $limit = 10): array
{
return Notification::where('user_id', $userId)
->where('is_read', false)
->orderBy('created_at', 'desc')
->limit($limit)
->get()
->toArray();
}
/**
* 获取未读通知数量
*
* @param int $userId
* @return int
*/
public function getUnreadCount(int $userId): int
{
return Notification::where('user_id', $userId)
->where('is_read', false)
->count();
}
/**
* 根据ID获取通知
*
* @param int $id
* @return Notification|null
*/
public function getById(int $id): ?Notification
{
return Notification::find($id);
}
/**
* 创建通知
*
* @param array $data
* @return Notification
*/
public function create(array $data): Notification
{
$notification = Notification::create($data);
// 如果用户在线，立即通过WebSocket发送
if ($this->webSocketService->isUserOnline($data['user_id'])) {
$this->sendViaWebSocket($notification);
}
return $notification;
}
/**
* 批量创建通知
*
* @param array $notificationsData
* @return array
*/
public function batchCreate(array $notificationsData): array
{
$notifications = [];
DB::beginTransaction();
try {
foreach ($notificationsData as $data) {
$notification = Notification::create($data);
$notifications[] = $notification;
// 如果用户在线，立即通过WebSocket发送
if ($this->webSocketService->isUserOnline($data['user_id'])) {
$this->sendViaWebSocket($notification);
}
}
DB::commit();
} catch (\Exception $e) {
DB::rollBack();
Log::error('批量创建通知失败', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
throw $e;
}
return $notifications;
}
/**
* 发送系统通知给单个用户
*
* @param int $userId
* @param string $title
* @param string $content
* @param string $type
* @param string $category
* @param array $extraData
* @return Notification
*/
public function sendToUser(
int $userId,
string $title,
string $content,
string $type = Notification::TYPE_INFO,
string $category = Notification::CATEGORY_SYSTEM,
array $extraData = []
): Notification {
$data = [
'user_id' => $userId,
'title' => $title,
'content' => $content,
'type' => $type,
'category' => $category,
'data' => $extraData,
'is_read' => false,
];
return $this->create($data);
}
/**
* 发送系统通知给多个用户
*
* @param array $userIds
* @param string $title
* @param string $content
* @param string $type
* @param string $category
* @param array $extraData
* @return array
*/
public function sendToUsers(
array $userIds,
string $title,
string $content,
string $type = Notification::TYPE_INFO,
string $category = Notification::CATEGORY_SYSTEM,
array $extraData = []
): array {
$notificationsData = [];
foreach ($userIds as $userId) {
$notificationsData[] = [
'user_id' => $userId,
'title' => $title,
'content' => $content,
'type' => $type,
'category' => $category,
'data' => $extraData,
'is_read' => false,
'created_at' => now(),
'updated_at' => now(),
];
}
return $this->batchCreate($notificationsData);
}
/**
* 发送系统广播通知（所有用户）
*
* @param string $title
* @param string $content
* @param string $type
* @param string $category
* @param array $extraData
* @return array
*/
public function broadcast(
string $title,
string $content,
string $type = Notification::TYPE_INFO,
string $category = Notification::CATEGORY_ANNOUNCEMENT,
array $extraData = []
): array {
// 获取所有用户ID
$userIds = \App\Models\Auth\User::where('status', 1)->pluck('id')->toArray();
return $this->sendToUsers($userIds, $title, $content, $type, $category, $extraData);
}
/**
* 通过WebSocket发送通知
*
* @param Notification $notification
* @return bool
*/
protected function sendViaWebSocket(Notification $notification): bool
{
$data = [
'type' => 'notification',
'data' => [
'id' => $notification->id,
'title' => $notification->title,
'content' => $notification->content,
'type' => $notification->type,
'category' => $notification->category,
'data' => $notification->data,
'action_type' => $notification->action_type,
'action_data' => $notification->action_data,
'timestamp' => $notification->created_at->timestamp,
]
];
$result = $this->webSocketService->sendToUser($notification->user_id, $data);
if ($result) {
$notification->markAsSent();
} else {
$notification->incrementRetry();
}
return $result;
}
/**
* 重试发送未发送的通知
*
* @param int $limit
* @return int
*/
public function retryUnsentNotifications(int $limit = 100): int
{
$notifications = Notification::where('sent_via_websocket', false)
->where('retry_count', '<', 3)
->where('created_at', '>', now()->subHours(24))
->orderBy('created_at', 'desc')
->limit($limit)
->get();
$sentCount = 0;
foreach ($notifications as $notification) {
if ($this->webSocketService->isUserOnline($notification->user_id)) {
if ($this->sendViaWebSocket($notification)) {
$sentCount++;
}
}
}
return $sentCount;
}
/**
* 标记通知为已读
*
* @param int $id
* @param int $userId
* @return bool
*/
public function markAsRead(int $id, int $userId): bool
{
$notification = Notification::where('id', $id)
->where('user_id', $userId)
->first();
if (!$notification) {
return false;
}
return $notification->markAsRead();
}
/**
* 批量标记通知为已读
*
* @param array $ids
* @param int $userId
* @return int
*/
public function batchMarkAsRead(array $ids, int $userId): int
{
return Notification::where('user_id', $userId)
->whereIn('id', $ids)
->update([
'is_read' => true,
'read_at' => now(),
]);
}
/**
* 标记所有通知为已读
*
* @param int $userId
* @return int
*/
public function markAllAsRead(int $userId): int
{
return Notification::where('user_id', $userId)
->where('is_read', false)
->update([
'is_read' => true,
'read_at' => now(),
]);
}
/**
* 删除通知
*
* @param int $id
* @param int $userId
* @return bool
*/
public function delete(int $id, int $userId): bool
{
$notification = Notification::where('id', $id)
->where('user_id', $userId)
->first();
if (!$notification) {
return false;
}
return $notification->delete();
}
/**
* 批量删除通知
*
* @param array $ids
* @param int $userId
* @return int
*/
public function batchDelete(array $ids, int $userId): int
{
return Notification::where('user_id', $userId)
->whereIn('id', $ids)
->delete();
}
/**
* 清空已读通知
*
* @param int $userId
* @return int
*/
public function clearReadNotifications(int $userId): int
{
return Notification::where('user_id', $userId)
->where('is_read', true)
->delete();
}
/**
* 获取通知统计
*
* @param int $userId
* @return array
*/
public function getStatistics(int $userId): array
{
$total = Notification::where('user_id', $userId)->count();
$unread = Notification::where('user_id', $userId)->where('is_read', false)->count();
$read = Notification::where('user_id', $userId)->where('is_read', true)->count();
// 按类型统计
$byType = Notification::where('user_id', $userId)
->selectRaw('type, COUNT(*) as count')
->groupBy('type')
->pluck('count', 'type')
->toArray();
// 按分类统计
$byCategory = Notification::where('user_id', $userId)
->selectRaw('category, COUNT(*) as count')
->groupBy('category')
->pluck('count', 'category')
->toArray();
return [
'total' => $total,
'unread' => $unread,
'read' => $read,
'by_type' => $byType,
'by_category' => $byCategory,
];
}
/**
* 发送任务通知
*
* @param int $userId
* @param string $title
* @param string $content
* @param array $taskData
* @return Notification
*/
public function sendTaskNotification(int $userId, string $title, string $content, array $taskData = []): Notification
{
return $this->sendToUser(
$userId,
$title,
$content,
Notification::TYPE_TASK,
Notification::CATEGORY_TASK,
array_merge(['task' => $taskData], $taskData)
);
}
/**
* 发送系统维护通知
*
* @param string $title
* @param string $content
* @param array $maintenanceData
* @return array
*/
public function sendMaintenanceNotification(string $title, string $content, array $maintenanceData = []): array
{
return $this->broadcast(
$title,
$content,
Notification::TYPE_WARNING,
Notification::CATEGORY_ANNOUNCEMENT,
array_merge(['maintenance' => $maintenanceData], $maintenanceData)
);
}
/**
* 发送新消息通知
*
* @param int $userId
* @param string $title
* @param string $content
* @param array $messageData
* @return Notification
*/
public function sendNewMessageNotification(int $userId, string $title, string $content, array $messageData = []): Notification
{
return $this->sendToUser(
$userId,
$title,
$content,
Notification::TYPE_INFO,
Notification::CATEGORY_MESSAGE,
array_merge(['message' => $messageData], $messageData)
);
}
/**
* 发送提醒通知
*
* @param int $userId
* @param string $title
* @param string $content
* @param array $reminderData
* @return Notification
*/
public function sendReminderNotification(int $userId, string $title, string $content, array $reminderData = []): Notification
{
return $this->sendToUser(
$userId,
$title,
$content,
Notification::TYPE_WARNING,
Notification::CATEGORY_REMINDER,
array_merge(['reminder' => $reminderData], $reminderData)
);
}
}

124
app/Services/System/TaskService.php
View File

@@ -3,10 +3,25 @@
namespace App\Services\System;
use App\Models\System\Task;
use App\Services\System\NotificationService;
use Illuminate\Support\Facades\Validator;
use Illuminate\Support\Facades\Log;
class TaskService
{
/**
* @var NotificationService
*/
protected $notificationService;
/**
* TaskService constructor
*/
public function __construct(NotificationService $notificationService)
{
$this->notificationService = $notificationService;
}
public function getList(array $params): array
{
$query = Task::query();
@@ -144,6 +159,9 @@ class TaskService
'last_output' => substr($output, 0, 10000),
]);
// 发送任务执行结果通知
$this->sendTaskNotification($task, $status, $errorMessage, $executionTime);
return [
'status' => $status,
'output' => $output,
@@ -164,4 +182,110 @@ class TaskService
'inactive' => $inactive,
];
}
/**
* 发送任务执行结果通知
*
* @param Task $task
* @param string $status
* @param string|null $errorMessage
* @param int $executionTime
* @return void
*/
protected function sendTaskNotification(Task $task, string $status, ?string $errorMessage, int $executionTime): void
{
try {
// 只对失败的任务或重要的成功任务发送通知
// 这里可以根据实际需求调整通知策略
if ($status === 'error') {
// 任务失败通知
$title = '任务执行失败: ' . $task->name;
$content = sprintf(
"任务 %s 执行失败，错误信息：%s\n执行时间：%d 毫秒",
$task->name,
$errorMessage ?: '未知错误',
$executionTime
);
// 获取管理员用户ID列表
$adminUserIds = $this->getAdminUserIds();
if (!empty($adminUserIds)) {
$this->notificationService->sendToUsers(
$adminUserIds,
$title,
$content,
\App\Models\System\Notification::TYPE_ERROR,
\App\Models\System\Notification::CATEGORY_TASK,
[
'task_id' => $task->id,
'task_name' => $task->name,
'command' => $task->command,
'error_message' => $errorMessage,
'execution_time' => $executionTime,
'last_run_at' => $task->last_run_at?->toDateTimeString()
]
);
}
} elseif ($executionTime > 60000) {
// 执行时间超过1分钟的成功任务，发送通知
$title = '任务执行完成: ' . $task->name;
$content = sprintf(
"任务 %s 执行成功，耗时：%.2f 秒",
$task->name,
$executionTime / 1000
);
$adminUserIds = $this->getAdminUserIds();
if (!empty($adminUserIds)) {
$this->notificationService->sendToUsers(
$adminUserIds,
$title,
$content,
\App\Models\System\Notification::TYPE_SUCCESS,
\App\Models\System\Notification::CATEGORY_TASK,
[
'task_id' => $task->id,
'task_name' => $task->name,
'execution_time' => $executionTime,
'last_run_at' => $task->last_run_at?->toDateTimeString()
]
);
}
}
} catch (\Exception $e) {
Log::error('发送任务通知失败', [
'task_id' => $task->id,
'error' => $e->getMessage()
]);
}
}
/**
* 获取管理员用户ID列表
*
* @return array
*/
protected function getAdminUserIds(): array
{
try {
// 获取拥有管理员权限的用户
// 这里可以根据实际业务逻辑调整，例如获取特定角色的用户
$adminUserIds = \App\Models\Auth\User::where('status', 1)
->whereHas('roles', function ($query) {
$query->where('name', 'admin');
})
->pluck('id')
->toArray();
return $adminUserIds;
} catch (\Exception $e) {
Log::error('获取管理员用户列表失败', [
'error' => $e->getMessage()
]);
return [];
}
}
}

32
app/Services/WebSocket/WebSocketHandler.php
View File

@@ -21,6 +21,16 @@ class WebSocketHandler implements WebSocketHandlerInterface
*/
protected $userOnlineService;
/**
* Get wsTable instance
*
* @return \Swoole\Table
*/
protected function getWsTable(): \Swoole\Table
{
return app('swoole')->wsTable;
}
/**
* WebSocketHandler constructor
*/
@@ -54,12 +64,12 @@ class WebSocketHandler implements WebSocketHandlerInterface
if ($userId && $token) {
// Store user connection mapping
$server->wsTable->set('uid:' . $userId, [
$this->getWsTable()->set('uid:' . $userId, [
'value' => $fd,
'expiry' => time() + 3600, // 1 hour expiry
]);
$server->wsTable->set('fd:' . $fd, [
$this->getWsTable()->set('fd:' . $fd, [
'value' => $userId,
'expiry' => time() + 3600
]);
@@ -231,7 +241,7 @@ class WebSocketHandler implements WebSocketHandlerInterface
$token = $data['token'] ?? null;
// Get the user ID from wsTable (set during connection)
$storedUserId = $server->wsTable->get('fd:' . $fd)['value'] ?? null;
$storedUserId = $this->getWsTable()->get('fd:' . $fd)['value'] ?? null;
if ($storedUserId && $storedUserId == $userId) {
// Authentication confirmed, send success response
@@ -291,13 +301,13 @@ class WebSocketHandler implements WebSocketHandlerInterface
}
// Get target user's connection
$targetFd = $server->wsTable->get('uid:' . $toUserId);
$targetFd = $this->getWsTable()->get('uid:' . $toUserId);
if ($targetFd && $targetFd['value']) {
$server->push((int)$targetFd['value'], json_encode([
'type' => 'chat',
'data' => [
'from_user_id' => $server->wsTable->get('fd:' . $fd)['value'] ?? null,
'from_user_id' => $this->getWsTable()->get('fd:' . $fd)['value'] ?? null,
'content' => $content,
'timestamp' => time()
]
@@ -334,7 +344,7 @@ class WebSocketHandler implements WebSocketHandlerInterface
protected function handleBroadcast(Server $server, int $fd, array $data): void
{
$message = $data['message'] ?? '';
$userId = $server->wsTable->get('fd:' . $fd)['value'] ?? null;
$userId = $this->getWsTable()->get('fd:' . $fd)['value'] ?? null;
// TODO: Check if user has admin permission to broadcast
// For now, allow any authenticated user
@@ -400,7 +410,7 @@ class WebSocketHandler implements WebSocketHandlerInterface
}
// Store subscription in wsTable
$server->wsTable->set('channel:' . $channel . ':fd:' . $fd, [
$this->getWsTable()->set('channel:' . $channel . ':fd:' . $fd, [
'value' => 1,
'expiry' => time() + 7200 // 2 hours
]);
@@ -443,7 +453,7 @@ class WebSocketHandler implements WebSocketHandlerInterface
}
// Remove subscription from wsTable
$server->wsTable->del('channel:' . $channel . ':fd:' . $fd);
$this->getWsTable()->del('channel:' . $channel . ':fd:' . $fd);
$server->push($fd, json_encode([
'type' => 'unsubscribed',
@@ -476,12 +486,12 @@ class WebSocketHandler implements WebSocketHandlerInterface
]);
// Get user ID from wsTable
$userId = $server->wsTable->get('fd:' . $fd)['value'] ?? null;
$userId = $this->getWsTable()->get('fd:' . $fd)['value'] ?? null;
if ($userId) {
// Remove user connection mapping
$server->wsTable->del('uid:' . $userId);
$server->wsTable->del('fd:' . $fd);
$this->getWsTable()->del('uid:' . $userId);
$this->getWsTable()->del('fd:' . $fd);
// Update user online status
$this->userOnlineService->updateUserOnlineStatus($userId, $fd, false);

40
app/Services/WebSocket/WebSocketService.php
View File

@@ -50,12 +50,14 @@ class WebSocketService
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
if (!$server) {
Log::warning('WebSocket server not available', ['user_id' => $userId]);
return false;
}
$fdInfo = $server->wsTable->get('uid:' . $userId);
$wsTable = app('swoole')->wsTable;
$fdInfo = $wsTable->get('uid:' . $userId);
if (!$fdInfo || !$fdInfo['value']) {
Log::info('User not connected to WebSocket', ['user_id' => $userId]);
@@ -67,8 +69,8 @@ class WebSocketService
if (!$server->isEstablished($fd)) {
Log::info('WebSocket connection not established', ['user_id' => $userId, 'fd' => $fd]);
// Clean up stale connection
$server->wsTable->del('uid:' . $userId);
$server->wsTable->del('fd:' . $fd);
$wsTable->del('uid:' . $userId);
$wsTable->del('fd:' . $fd);
return false;
}
@@ -119,6 +121,7 @@ class WebSocketService
return 0;
}
$wsTable = app('swoole')->wsTable;
$message = json_encode($data);
$count = 0;
@@ -129,7 +132,7 @@ class WebSocketService
// Check if we should exclude this user
if ($excludeUserId) {
$fdInfo = $server->wsTable->get('fd:' . $fd);
$fdInfo = $wsTable->get('fd:' . $fd);
if ($fdInfo && $fdInfo['value'] == $excludeUserId) {
continue;
}
@@ -159,11 +162,12 @@ class WebSocketService
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
if (!$server) {
Log::warning('WebSocket server not available for channel broadcast', ['channel' => $channel]);
return 0;
}
$wsTable = app('swoole')->wsTable;
$count = 0;
$message = json_encode($data);
@@ -173,7 +177,7 @@ class WebSocketService
continue;
}
$subscription = $server->wsTable->get('channel:' . $channel . ':fd:' . $fd);
$subscription = $wsTable->get('channel:' . $channel . ':fd:' . $fd);
if ($subscription) {
$server->push($fd, $message);
@@ -224,11 +228,13 @@ class WebSocketService
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
if (!$server) {
return false;
}
$fdInfo = $server->wsTable->get('uid:' . $userId);
$wsTable = app('swoole')->wsTable;
$fdInfo = $wsTable->get('uid:' . $userId);
if (!$fdInfo || !$fdInfo['value']) {
return false;
@@ -249,11 +255,13 @@ class WebSocketService
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
if (!$server) {
return false;
}
$fdInfo = $server->wsTable->get('uid:' . $userId);
$wsTable = app('swoole')->wsTable;
$fdInfo = $wsTable->get('uid:' . $userId);
if (!$fdInfo || !$fdInfo['value']) {
return false;
@@ -274,8 +282,8 @@ class WebSocketService
$server->disconnect($fd);
// Clean up
$server->wsTable->del('uid:' . $userId);
$server->wsTable->del('fd:' . $fd);
$wsTable->del('uid:' . $userId);
$wsTable->del('fd:' . $fd);
Log::info('User disconnected from WebSocket by server', [
'user_id' => $userId,
@@ -297,10 +305,12 @@ class WebSocketService
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
if (!$server) {
return [];
}
$wsTable = app('swoole')->wsTable;
$userIds = [];
foreach ($server->connections as $fd) {
@@ -308,7 +318,7 @@ class WebSocketService
continue;
}
$fdInfo = $server->wsTable->get('fd:' . $fd);
$fdInfo = $wsTable->get('fd:' . $fd);
if ($fdInfo && $fdInfo['value']) {
$userIds[] = (int)$fdInfo['value'];