From 1d5aff2f2ee8e900301dd15954e3a905aba9dc4a Mon Sep 17 00:00:00 2001 From: sky121113 Date: Wed, 22 Apr 2026 11:30:39 +0800 Subject: [PATCH] =?UTF-8?q?[REFACTOR]=20=E9=81=B7=E7=A7=BB=E9=81=A0?= =?UTF-8?q?=E7=AB=AF=E7=AE=A1=E7=90=86=E6=8C=87=E4=BB=A4=E8=87=B3=20MQTT?= =?UTF-8?q?=20=E9=9D=9E=E5=90=8C=E6=AD=A5=E6=9E=B6=E6=A7=8B=E8=88=87?= =?UTF-8?q?=E9=82=8F=E8=BC=AF=E5=84=AA=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 實作 MqttService 統一管理 MQTT 指令發送。 2. 遷移「遠端庫存與效期中心」至 MQTT 流程，並加入 batch_no (批號) 支援。 3. 修改 RemoteController 指令中心所有指令同步改由 MQTT 推送。 4. 修正 Go Gateway redis_consumer.go 確保 JSON 格式符合最新規格（移除寫死的 target/message_id）。 5. 實作 ProcessCommandAck 自動更新指令結果、異常回滾 (Rollback) 與機台自動亮燈 (Online)。 6. 完善 mqtt-communication-specs 規格文件，補齊所有指令定義與 ACK 範例。 7. 強化指令狀態保護，確保只有 pending 狀態的指令能被更新結果，防止重複處理。 --- .agents/skills/api-technical-specs/SKILL.md | 15 ++ .agents/skills/iot-communication/SKILL.md | 6 +- .../skills/mqtt-communication-specs/SKILL.md | 131 +++++++++++++-- app/Console/Commands/ListenMqttQueue.php | 43 ++++- .../Controllers/Admin/RemoteController.php | 21 ++- app/Jobs/Machine/ProcessCommandAck.php | 155 ++++++++++++++++++ app/Services/Machine/MachineService.php | 23 ++- app/Services/Machine/MqttService.php | 54 ++++++ .../internal/bridge/redis_consumer.go | 24 +-- mqtt-gateway/internal/handler/mqtt_handler.go | 9 +- mqtt-gateway/main.go | 12 +- routes/api.php | 7 +- 12 files changed, 459 insertions(+), 41 deletions(-) create mode 100644 app/Jobs/Machine/ProcessCommandAck.php create mode 100644 app/Services/Machine/MqttService.php diff --git a/.agents/skills/api-technical-specs/SKILL.md b/.agents/skills/api-technical-specs/SKILL.md index 0eac408..6d5aa56 100644 --- a/.agents/skills/api-technical-specs/SKILL.md +++ b/.agents/skills/api-technical-specs/SKILL.md @@ -303,3 +303,18 @@ description: 本技能規範定義了 Star Cloud 系統中所有機台 (IoT) 與 | id | String | 是 | 雲端下發的指令 ID | "99" | | type | String | 是 | 出貨類型代碼 (0: 成功, 其他: 失敗) | "0" | | stock | String | 是 | 出貨後的貨道剩餘數量 | "9" | + +--- + +## 4. 已遷移至 MQTT 的端點索引 + +以下端點的 HTTP REST API 已廢棄，改為透過 MQTT 即時通訊架構處理。詳細的 Payload 定義請參閱 [MQTT 通訊規範](file:///home/mama/projects/star-cloud/.agents/skills/mqtt-communication-specs/SKILL.md)。 + +| 原 API 代碼 | 名稱 | MQTT Topic | 備註 | +| :--- | :--- | :--- | :--- | +| B010 | 心跳上報 | `machine/{serial_no}/heartbeat` | [已廢棄](#312-b010-deprecated--removed-心跳上報與狀態同步) | +| B013 | 異常上報 | `machine/{serial_no}/error` | [已廢棄](#36-b013-deprecated--removed-機台故障與異常狀態上報) | +| B600 | 交易建立 | `machine/{serial_no}/transaction` | `action: create` | +| B601 | 發票紀錄 | `machine/{serial_no}/transaction` | `action: invoice` | +| B602 | 出貨結果 | `machine/{serial_no}/transaction` | `action: dispense` | +| B055 (PUT) | 遠端出貨回報 | `machine/{serial_no}/command/ack` | GET 查詢路由保留 | diff --git a/.agents/skills/iot-communication/SKILL.md b/.agents/skills/iot-communication/SKILL.md index 7170f46..f031e62 100644 --- a/.agents/skills/iot-communication/SKILL.md +++ b/.agents/skills/iot-communication/SKILL.md @@ -12,14 +12,14 @@ description: 規範智能販賣機與 Cloud 平台間的高頻通訊處理流程 所有來自機台的非即時性資料（日誌、心跳、狀態上報）必須遵循以下 pipeline。依據通訊協議不同，進入點有兩條路徑： ### 1.1 HTTP 管線 (低頻/大檔操作) -適用於 B000, B009, B012, B014, B017 等低頻、同步需求或大資料量的端點： +適用於 B000, B005, B009, B012, B014, B017, B018 等低頻、同步需求或大資料量的端點： 1. **API Controller (接收層)**：驗證 Request 合法性，隨即分派 (Dispatch) 任務至 Queue，並回傳 `202 Accepted`。 2. **Job (異步層)**：由背景 Worker 讀取隊列任務，呼叫對應 Service 處理。 3. **Service (邏輯層)**：封裝商業邏輯，更新資料庫。 4. **Model (儲存層)**：執行資料存取。 ### 1.2 MQTT 管線 (高頻/即時操作) -適用於 B010 (心跳), B013 (異常), B600 (交易) 等高頻或即時性端點： +適用於 B010 (心跳), B013 (異常), B600/B601/B602 (交易生命週期) 等高頻或即時性端點： 1. **Go Gateway (接收層)**：訂閱 EMQX Topic，提取 `serial_no`，包裝成標準 JSON。 2. **Redis List (橋接層)**：Go 執行 `RPUSH starcloud_database_mqtt_incoming_jobs {json}`。 3. **Laravel `mqtt:listen` (消費層)**：常駐指令 `BLPOP` 取出 JSON，根據 `type` 分派至對應 Job。 @@ -92,7 +92,7 @@ public function handle(MachineService $service): void 1. **$SYS Events**：由 Broker 自動觸發，走 **MQTT 管線**。用於毫秒級更新機台的 `online/offline` 狀態與 `last_heartbeat_at`。 2. **B010 (心跳)**：高頻點，走 **MQTT 管線** (`machine/+/heartbeat`)。更新 `last_heartbeat_at` 與感測器快照。 3. **B013 (異常)**：事件驅動點，走 **MQTT 管線** (`machine/+/error`)。寫入 `machine_logs` 並觸發告警。 -4. **B600 (交易)**：高價值點，走 **MQTT 管線** (`machine/+/transaction`)。建立 `Transaction` 紀錄並支援重試。 +4. **B600/B601/B602 (交易生命週期)**：高價值點，走 **MQTT 管線** (`machine/+/transaction`)。透過 `action` 欄位區分 `create` (交易建立) / `invoice` (發票) / `dispense` (出貨結果)。 5. **B012 (商品同步)**：大資料量，走 **HTTP 管線**。應確保 Service 層具備緩存 (Cache) 機制。 6. **B055 (遠端出貨)**：雲端下發指令，走 **MQTT 下行管線** (`machine/{id}/command`)。 diff --git a/.agents/skills/mqtt-communication-specs/SKILL.md b/.agents/skills/mqtt-communication-specs/SKILL.md index 979f8a3..461ae46 100644 --- a/.agents/skills/mqtt-communication-specs/SKILL.md +++ b/.agents/skills/mqtt-communication-specs/SKILL.md @@ -100,10 +100,11 @@ Mqtt3Client client = MqttClient.builder() | 主題名稱 (Topic) | 方向 | QoS | 用途說明 | | :--- | :--- | :--- | :--- | -| `machine/{serial_no}/heartbeat` | 設備 ➜ 雲端 | 0 | 每 10 秒上報狀態大包裝（含心跳、溫度、韌體版本等）。 | +| `machine/{serial_no}/heartbeat` | 設備 ➜ 雲端 | 0 | 每 1 分鐘上報狀態大包裝（溫度、韌體版本等）。 | | `machine/{serial_no}/error` | 設備 ➜ 雲端 | 1 | 發生硬體故障、卡貨或門未關時立即上報。 | -| `machine/{serial_no}/transaction` | 設備 ➜ 雲端 | 1 | 交易完成、出貨結果的回報。 | +| `machine/{serial_no}/transaction` | 設備 ➜ 雲端 | 1 | 交易生命週期上報 (B600 建立 / B601 發票 / B602 出貨結果)，透過 `action` 欄位區分。 | | `machine/{serial_no}/command` | 雲端 ➜ 設備 | 1 | 雲端下發的即時指令（出貨、更新、重啟）。 | +| `machine/{serial_no}/command/ack` | 設備 ➜ 雲端 | 1 | 機台回覆指令執行結果（B055 出貨回報）。 | --- @@ -137,7 +138,49 @@ Mqtt3Client client = MqttClient.builder() ``` *(註：`error_code` 應遵循 `MachineService::ERROR_CODE_MAP` 定義，雲端將自動轉換為對應語言標籤。`tid` 為貨道或任務 ID。)* -### 3.3 雲端指令 (Downstream Commands) - `machine/{id}/command` +### 3.3 交易生命週期上報 (Transaction) - `machine/{serial_no}/transaction` + +此 Topic 整合了交易生命週期的三個階段，透過 `action` 欄位區分： + +#### B600 交易建立 (`action: create`) +```json +{ + "action": "create", + "payment_type": "credit_card", + "amount": 25.0, + "slot_no": 3, + "product_id": "5" +} +``` + +#### B601 發票紀錄 (`action: invoice`) +```json +{ + "action": "invoice", + "invoice_no": "AB12345678", + "transaction_ref": "T202604140001", + "invoice_date": "2026-04-22", + "buyer_tax_id": "", + "amount": 25.0 +} +``` + +#### B602 出貨結果 (`action: dispense`) +```json +{ + "action": "dispense", + "slot_no": 3, + "result": "success", + "transaction_ref": "T202604140001", + "remaining_stock": 9 +} +``` + +> [!NOTE] +> - `serial_no` 不需放在 Payload 中，Go Gateway 自動從 Topic 路徑提取並注入。 +> - 若 Payload 不含 `action` 欄位，系統預設視為 `create`（向後相容既有 B600 格式）。 + +### 3.4 雲端指令 (Downstream Commands) - `machine/{serial_no}/command` 這是雲端主動下發給機台的訊息，取代原本 B010 Response 的輪詢等待。 ```json @@ -147,16 +190,69 @@ Mqtt3Client client = MqttClient.builder() "slot_no": 5, "transaction_id": "T202604140001" }, - "message_id": "MSG_123456789" + "command_id": "MSG_123456789" } ``` **常用指令集：** -- `reboot`: 機台重啟。 +- `reboot`: 遠端重啟機台系統。 +- `reboot_card`: 遠端重啟讀卡機。 +- `checkout`: 遠端結帳/登出指令。 +- `lock`: 遠端鎖定機台（暫停服務）。 +- `unlock`: 遠端解鎖機台（恢復服務）。 +- `change`: 遠端修改數值（夾帶 amount 參數）。 - `reload_config`: 重新下載參數 (B014)。 -- `reload_products`: 重新同步商品 (B012)。 +- `reload_products`: 重新同步商品 (B012，適用於商品主檔、圖文變更)。 +- `update_inventory`: 遠端即時修改庫存 (直接夾帶數量更新 SQLite，取代 HTTP)。 - `dispense`: 遠端出貨指令 (B055)。 -### 3.4 屬性上報 (Attribute Updates) +**`update_inventory` Payload 範例：** + ```json +{ + "command": "update_inventory", + "command_id": "3", + "payload": { + "data": [ + { + "slot_no": "3", + "stock": 6, + "expiry_date": null, + "batch_no": "B2026042201" + } + ] + } +} + ``` + +### 3.5 指令回覆 (Command ACK) - `machine/{serial_no}/command/ack` + +機台執行完雲端下發的指令後，透過此 Topic 回報執行結果。雲端收到後會根據 `command_id` 更新操作紀錄狀態，若結果為 `failed` 且為庫存指令，將觸發自動回滾 (Rollback)。 + +**Payload 格式：** +```json +{ + "command_id": "3", + "result": "success", + "stock": 6, + "message": "Update successful" +} +``` + +**欄位說明：** +- `command_id`: (Required) 雲端下發指令時夾帶的 ID。 +- `result`: (Required) 執行結果，固定為 `success` 或 `failed`。 +- `stock`: (Optional) 執行該指令後，該貨道的最終剩餘庫存。 +- `message`: (Optional) 額外的執行訊息或失敗原因。 + +| 欄位 | 類型 | 必填 | 說明 | +| :--- | :--- | :--- | :--- | +| `command_id` | String | 是 | 雲端下發的指令 ID（對應 `remote_commands.id`）| +| `result` | String | 是 | 執行結果：`"success"` 或 `"0"` 代表成功，其他代表失敗 | +| `stock` | String | 否 | 出貨後的貨道剩餘庫存數量 | + +> [!NOTE] +> 此 Topic 取代原本 B055 的 HTTP PUT 回報路由。Go Gateway 將此 4 層 Topic 識別為 `type: command_ack` 推入 Redis。 + +### 3.6 屬性上報 (Attribute Updates) 過去曾支援獨立的 `temperature` 與 `firmware_version` Topic，現已**廢棄**。 機台應將所有屬性更新集中於 `machine/{serial_no}/heartbeat` 的 JSON Payload 中發送，以降低連線開銷並簡化後端處理。 @@ -207,16 +303,23 @@ Laravel `mqtt:listen` 使用 `BLPOP` 長時間阻塞等待。為防止 PHP 端 --- -## 6. 與 REST API 的同步關係 +## 6. 斷線保障機制 -App 端的通訊策略為 **MQTT 優先、HTTP 兜底**： +App 端的通訊策略為 **MQTT 唯一通道 + 本地暫存補發**，不維護 HTTP Fallback： -| 狀態 | 心跳 | 指令接收 | 交易回報 | +| 狀態 | 心跳 | 指令接收 | 交易/發票/出貨 | | :--- | :--- | :--- | :--- | -| WSS 連線正常 | MQTT 發布 (10s) | MQTT 訂閱 | MQTT 發布 | -| WSS 斷線 < 3 分鐘 | 暫停，等待重連 | 不可用 | 暫存本地，重連後補發 | -| WSS 斷線 ≥ 3 分鐘 | HTTP B010 輪詢 | 不可用 (降級) | HTTP B600 回報 | -| WSS 恢復 | 切回 MQTT | 恢復 MQTT 訂閱 | 切回 MQTT | +| WSS 連線正常 | MQTT 發布 (1min) | MQTT 訂閱 | MQTT 發布 | +| WSS 斷線 | 暫停，等待重連 | 不可用 | 暫存至本地 SQLite 佇列 | +| WSS 恢復 | 恢復 MQTT 發布 | 恢復 MQTT 訂閱 | 自動逐筆補發暫存訊息 | + +**本地暫存補發規則：** +- 重連後延遲 2 秒開始補發，每筆間隔 200ms +- 重試上限 10 次，超過標記為 failed +- 暫存超過 72 小時的訊息直接丟棄 + +> [!CAUTION] +> **已遷移至 MQTT 的端點 (B600/B601/B602/B010/B013) 不保留 HTTP 路由。** 斷線時依賴 App 本地暫存機制保證資料完整性。 --- diff --git a/app/Console/Commands/ListenMqttQueue.php b/app/Console/Commands/ListenMqttQueue.php index 30f8a91..af6fcb3 100644 --- a/app/Console/Commands/ListenMqttQueue.php +++ b/app/Console/Commands/ListenMqttQueue.php @@ -9,6 +9,9 @@ use App\Jobs\Machine\ProcessHeartbeat; use App\Jobs\Machine\ProcessMachineError; use App\Jobs\Machine\ProcessTransaction; use App\Jobs\Machine\ProcessStatus; +use App\Jobs\Transaction\ProcessInvoice; +use App\Jobs\Transaction\ProcessDispenseRecord; +use App\Jobs\Machine\ProcessCommandAck; class ListenMqttQueue extends Command { @@ -90,11 +93,49 @@ class ListenMqttQueue extends Command ProcessMachineError::dispatch($serialNo, $payload); break; case 'transaction': - ProcessTransaction::dispatch($serialNo, $payload); + $this->dispatchTransactionByAction($serialNo, $payload); + break; + case 'command_ack': + // B055: 機台出貨結果 ACK (machine/{serial_no}/command/ack) + ProcessCommandAck::dispatch($serialNo, $payload); break; default: Log::notice("MQTT Listen: Unknown message type [{$type}]", ['data' => $data]); break; } } + + /** + * 根據交易 Payload 中的 action 欄位，分派至對應的 Job + * - create (或無 action): B600 交易建立 + * - invoice: B601 發票紀錄 + * - dispense: B602 出貨結果 + */ + protected function dispatchTransactionByAction(string $serialNo, array $payload): void + { + $action = $payload['action'] ?? 'create'; + + switch ($action) { + case 'invoice': + // B601: 發票紀錄 — ProcessInvoice 期望接收含 serial_no 的 data 陣列 + $payload['serial_no'] = $serialNo; + ProcessInvoice::dispatch($payload); + Log::info("MQTT Transaction [invoice] dispatched", ['serial_no' => $serialNo]); + break; + + case 'dispense': + // B602: 出貨結果 — ProcessDispenseRecord 期望接收含 serial_no 的 data 陣列 + $payload['serial_no'] = $serialNo; + ProcessDispenseRecord::dispatch($payload); + Log::info("MQTT Transaction [dispense] dispatched", ['serial_no' => $serialNo]); + break; + + case 'create': + default: + // B600: 交易建立（既有邏輯，向後相容無 action 欄位的舊格式） + ProcessTransaction::dispatch($serialNo, $payload); + Log::info("MQTT Transaction [create] dispatched", ['serial_no' => $serialNo]); + break; + } + } } diff --git a/app/Http/Controllers/Admin/RemoteController.php b/app/Http/Controllers/Admin/RemoteController.php index 2a17873..83dd0ac 100644 --- a/app/Http/Controllers/Admin/RemoteController.php +++ b/app/Http/Controllers/Admin/RemoteController.php @@ -6,10 +6,18 @@ use App\Http\Controllers\Controller; use Illuminate\Http\Request; use App\Models\Machine\Machine; use App\Models\Machine\RemoteCommand; +use App\Services\Machine\MqttService; use Illuminate\Support\Facades\Auth; class RemoteController extends Controller { + protected $mqttService; + + public function __construct(MqttService $mqttService) + { + $this->mqttService = $mqttService; + } + /** * 遠端管理指揮中心 */ @@ -147,7 +155,7 @@ class RemoteController extends Controller 'executed_at' => now(), ]); - RemoteCommand::create([ + $command = RemoteCommand::create([ 'machine_id' => $validated['machine_id'], 'user_id' => auth()->id(), 'command_type' => $validated['command_type'], @@ -156,6 +164,17 @@ class RemoteController extends Controller 'note' => $validated['note'] ?? null, ]); + // 【新增】同步推播 MQTT 指令 + $machine = Machine::find($validated['machine_id']); + if ($machine) { + $this->mqttService->pushCommand( + $machine->serial_no, + $command->command_type, + $command->payload, + $command->id + ); + } + session()->flash('success', __('Command has been queued successfully.')); if ($request->expectsJson()) { diff --git a/app/Jobs/Machine/ProcessCommandAck.php b/app/Jobs/Machine/ProcessCommandAck.php new file mode 100644 index 0000000..3f6af2f --- /dev/null +++ b/app/Jobs/Machine/ProcessCommandAck.php @@ -0,0 +1,155 @@ +serialNo = $serialNo; + $this->payload = $payload; + } + + /** + * 處理 B055 出貨回報 ACK + * 原始邏輯移植自 MachineController::reportDispenseResult() + */ + public function handle(): void + { + $commandId = $this->payload['command_id'] ?? null; + $resultCode = $this->payload['result'] ?? null; + $stockReported = $this->payload['stock'] ?? null; + + if (empty($commandId)) { + Log::warning('MQTT CommandAck: Missing command_id', [ + 'serial_no' => $this->serialNo, + 'payload' => $this->payload + ]); + return; + } + + $command = RemoteCommand::find($commandId); + + if (!$command) { + Log::warning('MQTT CommandAck: Command not found', [ + 'serial_no' => $this->serialNo, + 'command_id' => $commandId + ]); + return; + } + + $machine = $command->machine; + + // 【優先執行】更新機台為在線狀態 (只要收到訊息就代表在線，不受指令狀態影響) + if ($machine) { + $machine->update([ + 'last_heartbeat_at' => now(), + 'status' => 'online', + ]); + } + + // 核心邏輯：只有處於 pending 狀態的指令才處理後續的業務邏輯 (如庫存異動) + if ($command->status !== 'pending') { + Log::info('MQTT CommandAck: Command already processed, ignoring business logic', [ + 'command_id' => $commandId, + 'current_status' => $command->status, + ]); + return; + } + + // 判定執行結果 (result: "success" 或 "0" 代表成功) + $status = in_array($resultCode, ['success', '0'], true) ? 'success' : 'failed'; + + $payloadUpdates = [ + 'reported_stock' => $stockReported, + 'report_result' => $resultCode, + ]; + + $machine = $command->machine; + + if ($command->command_type === 'dispense') { + // 若成功，連動更新貨道庫存 + if ($status === 'success' && isset($command->payload['slot_no'])) { + $slotNo = $command->payload['slot_no']; + $slot = $machine->slots()->where('slot_no', $slotNo)->first(); + + if ($slot) { + $oldStock = $slot->stock; + // 若 APP 回傳的庫存大於 0 則使用回傳值，否則執行扣減 + $newStock = (int) ($stockReported ?? 0); + if ($newStock <= 0 && $oldStock > 0) { + $newStock = $oldStock - 1; + } + + $finalStock = max(0, $newStock); + $slot->update(['stock' => $finalStock]); + + // 紀錄庫存變化供前端顯示 + $payloadUpdates['old_stock'] = $oldStock; + $payloadUpdates['new_stock'] = $finalStock; + + ProcessStateLog::dispatch( + $machine->id, + $machine->company_id, + "Remote dispense successful for slot :slot", + 'info', + ['slot' => $slotNo] + ); + } + } + } elseif ($command->command_type === 'reload_stock') { + // 若同步失敗，自動 Rollback 回滾庫存 + if ($status === 'failed' && isset($command->payload['slot_no'], $command->payload['old'])) { + $slotNo = $command->payload['slot_no']; + $slot = $machine->slots()->where('slot_no', $slotNo)->first(); + + if ($slot) { + $oldData = $command->payload['old']; + $slot->update([ + 'stock' => $oldData['stock'], + 'expiry_date' => $oldData['expiry_date'] ?? null, + 'batch_no' => $oldData['batch_no'] ?? null, + ]); + + Log::warning('MQTT CommandAck failed, rolled back inventory', [ + 'serial_no' => $this->serialNo, + 'slot_no' => $slotNo, + 'rolled_back_to' => $oldData + ]); + } + } + } + + $command->update([ + 'status' => $status, + 'executed_at' => now(), + 'payload' => array_merge($command->payload, $payloadUpdates), + ]); + + Log::info("MQTT CommandAck processed", [ + 'serial_no' => $this->serialNo, + 'command_id' => $commandId, + 'status' => $status, + ]); + } +} diff --git a/app/Services/Machine/MachineService.php b/app/Services/Machine/MachineService.php index 2cb3677..ff42d46 100644 --- a/app/Services/Machine/MachineService.php +++ b/app/Services/Machine/MachineService.php @@ -203,7 +203,7 @@ class MachineService */ public function updateSlot(Machine $machine, array $data, ?int $userId = null): void { - DB::transaction(function () use ($machine, $data, $userId) { + $command = DB::transaction(function () use ($machine, $data, $userId) { $slotNo = $data['slot_no']; $stock = $data['stock'] ?? null; $expiryDate = $data['expiry_date'] ?? null; @@ -236,7 +236,7 @@ class MachineService ]); // 建立遠端指令紀錄 (Unified Command Concept) - RemoteCommand::create([ + return RemoteCommand::create([ 'machine_id' => $machine->id, 'user_id' => $userId, 'command_type' => 'reload_stock', @@ -252,6 +252,25 @@ class MachineService ] ]); }); + + // 交易成功後，推播 MQTT 遠端指令 + $mqttPayload = [ + 'data' => [ + [ + 'slot_no' => $command->payload['slot_no'], + 'stock' => $command->payload['new']['stock'], + 'expiry_date' => $command->payload['new']['expiry_date'], + 'batch_no' => $command->payload['new']['batch_no'] ?? null, + ] + ] + ]; + + app(\App\Services\Machine\MqttService::class)->pushCommand( + $machine->serial_no, + 'update_inventory', + $mqttPayload, + (string) $command->id + ); } /** diff --git a/app/Services/Machine/MqttService.php b/app/Services/Machine/MqttService.php new file mode 100644 index 0000000..a102c86 --- /dev/null +++ b/app/Services/Machine/MqttService.php @@ -0,0 +1,54 @@ + $serialNo, + 'command' => $command, + 'payload' => $payload, + 'command_id' => (string) $commandId, + 'timestamp' => now()->toIso8601String(), + ]; + + // 系統預設 outgoing queue 名稱 (Go Gateway 預設讀取此佇列) + $queueKey = config('app.mqtt_outgoing_queue', 'mqtt_outgoing_commands'); + + // Laravel 的 Redis Facade 會自動加上 database prefix + Redis::rpush($queueKey, json_encode($message)); + + Log::info("MQTT Command Pushed to Queue", [ + 'queue' => $queueKey, + 'serial_no' => $serialNo, + 'command' => $command, + 'command_id' => $commandId + ]); + + return true; + } catch (\Exception $e) { + Log::error("Failed to push MQTT command", [ + 'serial_no' => $serialNo, + 'command' => $command, + 'error' => $e->getMessage() + ]); + return false; + } + } +} diff --git a/mqtt-gateway/internal/bridge/redis_consumer.go b/mqtt-gateway/internal/bridge/redis_consumer.go index 0bf4e3a..c8ca6a4 100644 --- a/mqtt-gateway/internal/bridge/redis_consumer.go +++ b/mqtt-gateway/internal/bridge/redis_consumer.go @@ -15,13 +15,6 @@ type RedisConsumer struct { queueKey string } -type CommandPayload struct { - Target string `json:"target"` - Command string `json:"command"` - Payload interface{} `json:"payload"` - MessageID string `json:"message_id"` -} - func NewRedisConsumer(rdb *redis.Client, mqttClient mqtt.Client, queueKey string) *RedisConsumer { return &RedisConsumer{rdb: rdb, mqttClient: mqttClient, queueKey: queueKey} } @@ -43,20 +36,31 @@ func (c *RedisConsumer) Start(ctx context.Context) { } // result[0] 是 key，result[1] 是 value - var cmd CommandPayload + var cmd map[string]interface{} err = json.Unmarshal([]byte(result[1]), &cmd) if err != nil { log.Printf("Failed to unmarshal command: %v", err) continue } - topic := "machine/" + cmd.Target + "/command" + serialNo, ok := cmd["serial_no"].(string) + if !ok || serialNo == "" { + log.Printf("Invalid or missing serial_no in outgoing command: %v", string(result[1])) + continue + } + + // 將發送方才需要的欄位移除，保持 Payload 乾淨 + delete(cmd, "serial_no") + delete(cmd, "timestamp") + + topic := "machine/" + serialNo + "/command" payload, _ := json.Marshal(cmd) token := c.mqttClient.Publish(topic, 1, false, payload) token.Wait() - log.Printf("Sent command [%s] to machine [%s] via topic: %s", cmd.Command, cmd.Target, topic) + commandType, _ := cmd["command"].(string) + log.Printf("Sent command [%s] to machine [%s] via topic: %s", commandType, serialNo, topic) } } } diff --git a/mqtt-gateway/internal/handler/mqtt_handler.go b/mqtt-gateway/internal/handler/mqtt_handler.go index 2207486..cef03c4 100644 --- a/mqtt-gateway/internal/handler/mqtt_handler.go +++ b/mqtt-gateway/internal/handler/mqtt_handler.go @@ -68,7 +68,9 @@ func (h *MqttHandler) HandleMessage(client mqtt.Client, msg mqtt.Message) { return } } else { - // 一般業務主題 (machine/{serialNo}/{type}) + // 一般業務主題 + // 3 層: machine/{serialNo}/{type} (如 heartbeat, error, transaction) + // 4 層: machine/{serialNo}/{type}/{subType} (如 command/ack) parts := strings.Split(topic, "/") if len(parts) < 3 { return @@ -76,6 +78,11 @@ func (h *MqttHandler) HandleMessage(client mqtt.Client, msg mqtt.Message) { serialNo = parts[1] msgType = parts[2] + // 4 層 Topic：將 type/subType 組合為 type_subType (如 command/ack → command_ack) + if len(parts) >= 4 { + msgType = parts[2] + "_" + parts[3] + } + err := json.Unmarshal(payload, &jsonPayload) if err != nil { // 如果不是有效的 JSON，則視為純文字/數值 (Raw Value) diff --git a/mqtt-gateway/main.go b/mqtt-gateway/main.go index 714b018..6e7d2f8 100644 --- a/mqtt-gateway/main.go +++ b/mqtt-gateway/main.go @@ -55,20 +55,24 @@ func main() { opts.SetOnConnectHandler(func(c mqtt.Client) { log.Printf("Connected to MQTT Broker as Gateway [%s] at %s", cfg.MQTTClientID, cfg.MQTTAddr) - // 訂閱所有機台的上行 Topic + // 訂閱所有機台的上行 Topic (3 層: machine/{serial_no}/{type}) token := c.Subscribe("machine/+/+", 1, mqttHandler.HandleMessage) token.Wait() + // 訂閱 command/ack 子 Topic (4 層: machine/{serial_no}/command/ack) + ackToken := c.Subscribe("machine/+/command/ack", 1, mqttHandler.HandleMessage) + ackToken.Wait() + // 訂閱 EMQX 5 系統主題 (連線與斷線事件) sysToken := c.Subscribe("$SYS/brokers/+/clients/+/connected", 1, mqttHandler.HandleMessage) sysToken.Wait() sysToken2 := c.Subscribe("$SYS/brokers/+/clients/+/disconnected", 1, mqttHandler.HandleMessage) sysToken2.Wait() - if token.Error() != nil || sysToken.Error() != nil || sysToken2.Error() != nil { - log.Printf("Failed to subscribe to MQTT topics: %v %v %v", token.Error(), sysToken.Error(), sysToken2.Error()) + if token.Error() != nil || ackToken.Error() != nil || sysToken.Error() != nil || sysToken2.Error() != nil { + log.Printf("Failed to subscribe to MQTT topics: %v %v %v %v", token.Error(), ackToken.Error(), sysToken.Error(), sysToken2.Error()) } else { - log.Println("Successfully subscribed to machine/+/+ and $SYS topics") + log.Println("Successfully subscribed to machine/+/+, machine/+/command/ack and $SYS topics") } }) diff --git a/routes/api.php b/routes/api.php index 40f7d96..9fe1a39 100644 --- a/routes/api.php +++ b/routes/api.php @@ -72,15 +72,12 @@ Route::prefix('v1')->middleware(['throttle:api'])->group(function () { Route::match(['get', 'patch'], 'machine/products/B012', [App\Http\Controllers\Api\V1\App\MachineController::class, 'getProducts']); - // 遠端指令出貨控制 (B055) + // 遠端指令出貨控制 (B055) — PUT 回報已遷移至 MQTT command/ack Topic Route::get('machine/dispense/B055', [App\Http\Controllers\Api\V1\App\MachineController::class, 'getDispenseQueue']); - Route::put('machine/dispense/B055', [App\Http\Controllers\Api\V1\App\MachineController::class, 'reportDispenseResult']); - // 交易、發票與出貨 (B600, B601, B602) + // 交易 (B600) — B601/B602 已遷移至 MQTT transaction Topic Route::post('machine/restock/B018', [App\Http\Controllers\Api\V1\App\MachineController::class, 'recordRestock']); Route::post('B600', [App\Http\Controllers\Api\V1\App\TransactionController::class, 'store']); - Route::post('B601', [App\Http\Controllers\Api\V1\App\TransactionController::class, 'recordInvoice']); - Route::post('B602', [App\Http\Controllers\Api\V1\App\TransactionController::class, 'recordDispense']); }); Route::prefix('machines')->group(function () {