xinyouhe/fdesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加消息推送

Browse Source
This commit is contained in:
mzl
2025-10-12 06:22:29 +00:00
parent 8e89ffdcf5
commit 7e820ee9c4
8 changed files with 441 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/common/utils/WsEvents.php
+31 -15
View File
@@ -53,6 +53,16 @@ class WsEvents
}
}
protected function eventToMethod(string $event): string
{
$parts = explode('.', $event);
$method = array_shift($parts);
foreach ($parts as $part) {
$method .= ucfirst($part);
}
return $method;
}
// ================= 系统事件 =================
/**
@@ -112,6 +122,14 @@ class WsEvents
'system.auth',
['uid' => $user->uid]
));
// 初始化推送
$wdata = [
'is_first' => true,
'event' => 'approval.todo.counts',
'user_id' => $user->uid
];
\Webman\RedisQueue\Redis::send('ws-msg', $wdata);
} catch (\Throwable $th) {
flg('[x] [TOKEN认证] TOKEN认证失败：'.$th->getMessage(), false);
Gateway::sendToClient($client_id, WsResponse::error(
@@ -235,22 +253,20 @@ class WsEvents
));
}
/**
* 推送消息通知
*/
public function pushWsNotice(array $wsData, int $userId)
{
Gateway::sendToUid($userId, WsResponse::success('notify.receive', $wsData));
}
// ================= 辅助方法 =================
protected function eventToMethod(string $event): string
/*
approval.todo.counts 审批待办数量
*/
// ================= 主动推送 =================
public function pushWebsocket(string $event, int|string $uid, array $data)
{
$parts = explode('.', $event);
$method = array_shift($parts);
foreach ($parts as $part) {
$method .= ucfirst($part);
}
return $method;
return match (gettype($uid)) {
'integer' => Gateway::sendToUid($uid, WsResponse::success($event, $data)),
'string' => Gateway::sendToClient($uid, WsResponse::success($event, $data)),
default => throw new \InvalidArgumentException('UID must be integer or string'),
};
}
}
app/common/utils/approval/service/ApprovalFlowService.php
+151 -12
View File
@@ -70,6 +70,8 @@ class ApprovalFlowService
$this->notifyCCs($firstStep, $instance, 'pending');
$this->pushTodoChanged($approverIds);
return ['instance_id' => $instanceId];
});
}
@@ -152,12 +154,18 @@ class ApprovalFlowService
$outcome = $this->getStepOutcome($instanceId, $round, $idx, $type); // 'pending' | 'approve' | 'reject'
if ($outcome === 'pending') {
$this->pushTodoChanged([$approverId]);
return ['status' => 'step_pending', 'msg' => '审批待定'];
}
$otherPendingUids = $this->getPendingApproverIds($instanceId, $round, $idx);
// 3) 有结论：关闭本步其余 pending
$this->closeOtherPending($instanceId, $round, $idx, [$approverId]);
// 这些人 pending→skipped，待办数减少
$this->pushTodoChanged($otherPendingUids);
// 4) 拒绝 => 结束
if ($outcome === 'reject') {
Db::table('approval_instances')->where('id', $instanceId)->update([
@@ -167,6 +175,7 @@ class ApprovalFlowService
]);
$this->onFlowFinalized($instance, 'rejected');
$this->sendNotification([$instance->initiator_id], $instance, 'reject', false);
$this->pushTodoChanged([$approverId]);
return ['status' => 'completed', 'msg' => '拒绝'];
}
@@ -185,7 +194,7 @@ class ApprovalFlowService
]);
$this->onFlowFinalized($instance, $finalStatus);
$this->sendNotification([$instance->initiator_id], $instance, $notifyAction, false);
$this->pushTodoChanged([$approverId]);
return ['status' => 'completed', 'msg' => ($finalStatus === 'voided' ? '已作废' : '同意')];
}
@@ -207,6 +216,7 @@ class ApprovalFlowService
}
$this->sendNotification($nextApprovers, $instance, 'pending', false);
$this->notifyCCs($nextStep, $instance, 'pending');
$this->pushTodoChanged(array_merge([$approverId], $nextApprovers));
return ['status' => 'next_step', 'msg' => '进行下一步：'.$nextStep['name'] ?? ('Step '.$nextIndex)];
});
}
@@ -282,12 +292,18 @@ class ApprovalFlowService
$this->onFlowFinalized($instance, 'revoked');
$pendingUids = Db::table('approval_records')
->where('instance_id', $instanceId)
->where('status', 'pending')
->pluck('approver_id')->toArray();
Db::table('approval_records')
->where('instance_id', $instanceId)
->where('status', 'pending')
->update(['status' => 'skipped', 'approved_at' => $now, 'updated_at' => $now]);
$this->sendNotification([$initiatorId], $instance, 'revoked', false);
$this->pushTodoChanged($pendingUids);
return ['status' => 'completed', 'result' => 'revoked'];
});
}
@@ -378,6 +394,13 @@ class ApprovalFlowService
$this->onFlowFinalized((object)$instance, $final);
$pendingUids = Db::table('approval_records')
->where('instance_id', $instance->id)
->where('round', (int)$instance->round)
->where('step_index', $instance->current_step)
->where('status', 'pending')
->pluck('approver_id')->toArray();
Db::table('approval_records')
->where('instance_id', $instance->id)
->where('round', (int)$instance->round)
@@ -386,6 +409,8 @@ class ApprovalFlowService
->update(['status' => 'timeout', 'approved_at' => $now, 'updated_at' => $now]);
$this->notifyInitiator($instance, $finalAction);
$this->pushTodoChanged($pendingUids);
});
}
}
@@ -586,7 +611,7 @@ class ApprovalFlowService
{
$records = Db::table('approval_records')
->where('instance_id', $instanceId)
->where('round', $round) // ★
->where('round', $round) //
->where('step_index', $stepIndex)
->get();
@@ -620,7 +645,7 @@ class ApprovalFlowService
$now = date('Y-m-d H:i:s');
$q = Db::table('approval_records')
->where('instance_id', $instanceId)
->where('round', $round) // ★
->where('round', $round) //
->where('step_index', $stepIndex)
->where('status', 'pending');
@@ -637,7 +662,7 @@ class ApprovalFlowService
$exists = Db::table('approval_records')
->where('instance_id', (int)$instance->id)
->where('round', $round) // ★ 限当前轮
->where('round', $round) // 限当前轮
->where('step_index', $stepIndex)
->where('approver_id', $approverId)
->where('status', 'pending')
@@ -649,7 +674,7 @@ class ApprovalFlowService
'instance_id' => (int)$instance->id,
'approver_id' => $approverId,
'step_index' => $stepIndex,
'round' => $round, // ★
'round' => $round, //
'action' => null,
'status' => 'pending',
'comment' => null,
@@ -995,7 +1020,7 @@ class ApprovalFlowService
$finishedAt = null;
$durationSec = null;
// ★ 每一步的“审批意见”（只展示有内容的 comment）
// 每一步的“审批意见”（只展示有内容的 comment）
$opinions = []; // [{approver_id, approver_name?, action, status, comment, time}...]
if ($hasRecs) {
@@ -1082,7 +1107,7 @@ class ApprovalFlowService
'started_at' => $startedAt,
'finished_at' => $finishedAt,
'duration_seconds' => $durationSec,
'opinions' => $opinions, // ★ 这就是每个审批意见
'opinions' => $opinions, // 这就是每个审批意见
];
// 把申请人信息并到第一步（可选）
@@ -1262,13 +1287,18 @@ class ApprovalFlowService
$deadlineAt = date('Y-m-d H:i:s', time() + $flow->timeout_hours * 3600);
}
$oldPendingUids = Db::table('approval_records')
->where('instance_id', $instanceId)
->where('status', 'pending')
->pluck('approver_id')->toArray();
// 清理可能遗留的 pending
Db::table('approval_records')
->where('instance_id', $instanceId)
->where('status', 'pending')
->update(['status'=>'skipped','approved_at'=>$now,'updated_at'=>$now]);
// ★ 轮次 +1
// 轮次 +1
$newRound = ((int)$instance->round) + 1;
// 重新开启实例
@@ -1303,6 +1333,7 @@ class ApprovalFlowService
$extra = $appealText !== '' ? ['appeal' => $appealText] : [];
$this->sendNotification($approvers, $instanceLite, 'pending', false, $extra);
$this->notifyCCs($firstStep, $instanceLite, 'pending');
$this->pushTodoChanged(array_merge($oldPendingUids, $approvers));
return [
'status' => 'resubmitted',
@@ -1595,13 +1626,14 @@ class ApprovalFlowService
'created_at' => $now,
]);
$this->pushTodoChanged([$fromUserId, $toUserId]);
return ['status' => 'ok', 'record_id' => $effectiveRecordId];
});
}
/** ★ 新增：发起“作废审批” */
/** 新增：发起“作废审批” */
public function startNullify(
string $voidFlowType, // 例如 'stock_orders_nullify'
string $bizType, // 原业务类型，如 'stock_orders'
@@ -1686,7 +1718,7 @@ class ApprovalFlowService
'initiator_id' => $initiatorId,
'status' => 'processing',
'current_step' => $firstIndex,
'round' => 1, // ★
'round' => 1, //
'deadline_at' => $deadlineAt,
'created_at' => $now,
'updated_at' => $now,
@@ -1701,7 +1733,7 @@ class ApprovalFlowService
'initiator_id' => $initiatorId,
'current_step' => $firstIndex,
'status' => 'processing',
'round' => 1, // ★
'round' => 1, //
];
// 6) 生成首步待办 + 通知（带上作废原因）
@@ -1715,7 +1747,7 @@ class ApprovalFlowService
}
$this->sendNotification($approverIds, $instance, 'pending', false, $extra);
$this->notifyCCs($firstStep, $instance, 'pending');
$this->pushTodoChanged($approverIds);
return ['instance_id' => $instanceId];
});
}
@@ -1810,4 +1842,111 @@ class ApprovalFlowService
return $final;
}
/** WebSocket：把一组用户的“待办数量”推送给前端 */
private function pushTodoChanged22(array $userIds): void
{
$userIds = array_values(array_unique(array_map('intval', $userIds)));
if (empty($userIds)) return;
// 统计这些用户当前的待办数量（以 approval_records.status = 'pending' 为准）
$rows = Db::table('approval_records')
->whereIn('approver_id', $userIds)
->where('status', 'pending')
->groupBy('approver_id')
->select('approver_id', Db::raw('COUNT(*) as cnt'))
->get();
// 默认都置 0，再填充真实值
$payload = array_fill_keys($userIds, 0);
foreach ($rows as $r) {
$payload[(int)$r->approver_id] = (int)$r->cnt;
}
try {
foreach ($payload as $uid => $cnt) {
$data = [
'event' => 'approval.todo.counts',
'user_id' => $uid,
'num' => $cnt,
];
\Webman\RedisQueue\Redis::send('ws-msg', $data);
}
} catch (\Throwable $e) {
Log::error('ws todo push error: '.$e->getMessage());
}
}
/** 取“某实例、某轮次、某步骤”里当前 pending 的审批人（用于推送受影响的用户） */
private function getPendingApproverIds(int $instanceId, int $round, int $stepIndex): array
{
$ids = Db::table('approval_records')
->where('instance_id', $instanceId)
->where('round', $round)
->where('step_index', $stepIndex)
->where('status', 'pending')
->pluck('approver_id')->toArray();
return array_values(array_unique(array_map('intval', $ids)));
}
/** 登录/首屏初始化：获取当前用户的待办总数（仅统计当前步骤、进行中的实例） */
public function getInitTodoCount(int $userId): int
{
$userId = (int)$userId;
if ($userId <= 0) {
throw new BusinessException('无效的用户ID');
}
$row = Db::table('approval_records as ar')
->join('approval_instances as ai', 'ai.id', '=', 'ar.instance_id')
->where('ar.approver_id', $userId)
->where('ar.status', 'pending')
->where('ai.status', 'processing')
// 只统计实例“当前轮次、当前步骤”的待办，避免历史残留干扰
->whereRaw('ar.round = ai.round')
->whereRaw('ar.step_index = ai.current_step')
->selectRaw('COUNT(*) as c')
->first();
return (int)($row->c ?? 0);
}
/** WebSocket：把一组用户的“待办数量”推送给前端（与 getInitTodoCount 统一口径） */
private function pushTodoChanged(array $userIds): void
{
$userIds = array_values(array_unique(array_map('intval', $userIds)));
if (empty($userIds)) return;
// 只统计：实例进行中 + 当前轮次 + 当前步骤 的 pending
$rows = Db::table('approval_records as ar')
->join('approval_instances as ai', 'ai.id', '=', 'ar.instance_id')
->whereIn('ar.approver_id', $userIds)
->where('ar.status', 'pending')
->where('ai.status', 'processing')
->whereRaw('ar.round = ai.round')
->whereRaw('ar.step_index = ai.current_step')
->groupBy('ar.approver_id')
->select('ar.approver_id', Db::raw('COUNT(*) as cnt'))
->get();
// 默认置 0，再填充真实值
$payload = array_fill_keys($userIds, 0);
foreach ($rows as $r) {
$payload[(int)$r->approver_id] = (int)$r->cnt;
}
try {
foreach ($payload as $uid => $cnt) {
\Webman\RedisQueue\Redis::send('ws-msg', [
'event' => 'approval.todo.counts',
'user_id' => $uid,
'num' => $cnt,
]);
}
} catch (\Throwable $e) {
Log::error('ws todo push error: '.$e->getMessage());
}
}
}