[REFACTOR] 遷移遠端管理指令至 MQTT 非同步架構與邏輯優化

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

@@ -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 查詢路由保留 |

.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`)。
 

.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 連線正常 | MQTT 發布 (1min) | MQTT 訂閱 | MQTT 發布 |
-| WSS 斷線 < 3 分鐘 | 暫停，等待重連 | 不可用 | 暫存本地，重連後補發 |
+| WSS 斷線 | 暫停，等待重連 | 不可用 | 暫存至本地 SQLite 佇列 |
-| WSS 斷線 ≥ 3 分鐘 | HTTP B010 輪詢 | 不可用 (降級) | HTTP B600 回報 |
+| WSS 恢復 | 恢復 MQTT 發布 | 恢復 MQTT 訂閱 | 自動逐筆補發暫存訊息 |
-| WSS 恢復 | 切回 MQTT | 恢復 MQTT 訂閱 | 切回 MQTT |
+
+**本地暫存補發規則：**
+- 重連後延遲 2 秒開始補發，每筆間隔 200ms
+- 重試上限 10 次，超過標記為 failed
+- 暫存超過 72 小時的訊息直接丟棄
+
+> [!CAUTION]
+> **已遷移至 MQTT 的端點 (B600/B601/B602/B010/B013) 不保留 HTTP 路由。** 斷線時依賴 App 本地暫存機制保證資料完整性。
 
 ---
 

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;
+        }
+    }
 }

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()) {

app/Jobs/Machine/ProcessCommandAck.php

@@ -0,0 +1,155 @@
+<?php
+
+namespace App\Jobs\Machine;
+
+use App\Models\Machine\Machine;
+use App\Models\Machine\RemoteCommand;
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+use Illuminate\Support\Facades\Log;
+
+class ProcessCommandAck implements ShouldQueue
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+
+    protected string $serialNo;
+    protected array $payload;
+
+    /**
+     * 建立 Job 實例
+     *
+     * @param string $serialNo 機台序號（由 Go Gateway 從 Topic 路徑注入）
+     * @param array $payload MQTT Payload，包含 command_id, result, stock
+     */
+    public function __construct(string $serialNo, array $payload)
+    {
+        $this->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,
+        ]);
+    }
+}

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
+        );
     }
 
     /**

app/Services/Machine/MqttService.php

@@ -0,0 +1,54 @@
+<?php
+
+namespace App\Services\Machine;
+
+use Illuminate\Support\Facades\Redis;
+use Illuminate\Support\Facades\Log;
+
+class MqttService
+{
+    /**
+     * 推播雲端指令 (Downstream Command) 給機台
+     * 遵循 framework.md 中的 Redis 橋接機制，推入 mqtt_outgoing_commands 佇列，由 Go Gateway 接手發佈。
+     *
+     * @param string $serialNo 機台序號
+     * @param string $command 指令名稱 (例如: update_inventory, dispense)
+     * @param array $payload 指令的內容資料
+     * @param string $commandId 唯一指令識別碼 (通常對應 RemoteCommand 的 ID)
+     * @return bool
+     */
+    public function pushCommand(string $serialNo, string $command, array $payload, string $commandId): bool
+    {
+        try {
+            $message = [
+                'serial_no' => $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;
+        }
+    }
+}

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)
 		}
 	}
 }

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)

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 {
+		if token.Error() != nil || ackToken.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())
+			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")
 		}
 	})
 

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 () {