[FEAT] 實作 MQTT Gateway 整合與高併發 IoT 通訊架構

1. 導入 Go 語言撰寫的 MQTT Gateway (mqtt-gateway/)，負責訊息接收並透過 Redis List 轉發至 Laravel。 2. 更新 compose.yaml 以包含 EMQX Broker 與 mqtt-gateway 服務，並配置相關環境變數。 3. 新增 ListenMqttQueue 指令與對應的異步處理 Job (ProcessHeartbeat, ProcessTransaction, ProcessMachineError)。 4. 實作 MqttSyncAuth 指令與 Machine Model Observer，支援自動化的機台認證資訊同步至 Redis。 5. 新增 MqttCommandService 用於處理雲端至機台的即時指令下發。 6. 更新專案規範文件 (framework.md 與 Skills) 以反映新的 MQTT 通訊機制與安全性規範。 7. 提供 simulate-heartbeat.sh 指令碼以供開發與測試使用。
2026-04-16 14:28:42 +08:00 · 2026-04-16 14:28:42 +08:00 · 959625a640
commit 959625a640
parent 1301bf1cb8
24 changed files with 815 additions and 32 deletions
--- a/.agents/rules/framework.md
+++ b/.agents/rules/framework.md
@ -96,9 +96,13 @@ Go 專案以 Monorepo 形式置於專案根目錄下的 `mqtt-gateway/` 資料
 ### 4.4 Redis 橋接機制 (Go ↔ Laravel)
 Go Gateway 與 Laravel 之間透過 Redis List 進行單向異步橋接：

-*   **Redis Key**：`mqtt_incoming_jobs`
-*   **Go 端 (生產者)**：將 MQTT 收到的 Payload 包裝成標準 JSON 後，執行 `RPUSH mqtt_incoming_jobs {json}`。
-*   **Laravel 端 (消費者)**：常駐指令 `php artisan mqtt:listen` 持續執行 `BLPOP mqtt_incoming_jobs`，取得 JSON 後解碼並分派至對應的 Laravel Job (如 `ProcessHeartbeat`, `ProcessTransaction`)。
+*   **Redis Key (邏輯名)**：`mqtt_incoming_jobs`
+*   **Redis Key (實際名)**：`starcloud_database_mqtt_incoming_jobs` (含 Laravel 前綴)
+*   **Go 端 (生產者)**：將 MQTT 收到的 Payload 包裝成標準 JSON 後，執行 `RPUSH starcloud_database_mqtt_incoming_jobs {json}`。Go 程式碼中透過 `"starcloud_database_" + cfg.IncomingQueueKey` 手動加入前綴。
+*   **Laravel 端 (消費者)**：常駐指令 `php artisan mqtt:listen` 持續執行 `BLPOP mqtt_incoming_jobs`（Laravel Redis 驅動自動加上前綴），取得 JSON 後解碼並分派至對應的 Laravel Job (如 `ProcessHeartbeat`, `ProcessTransaction`)。
+
+> [!CAUTION]
+> **Redis 前綴一致性**：Laravel 的 Redis 驅動會自動在 Key 前加上 `config('database.redis.options.prefix')` (預設 `starcloud_database_`)。Go 端不使用 Laravel 的 Redis 客戶端，因此**必須在程式碼中手動加上此前綴**，否則雙方會操作不同的 Key，導致訊息永遠無法傳達。

 **JSON 橋接格式規範**：
 ```json
@ -118,11 +122,16 @@ Go Gateway 與 Laravel 之間透過 Redis List 進行單向異步橋接：
 > **為何不讓 Go 直接寫入 Laravel Queue？** 因為 Laravel Queue 的 Payload 包含 PHP 序列化物件字串 (`serialize()`)，Go 無法安全產生此格式。透過獨立的 Redis List + 純 JSON，可徹底解耦兩端的技術依賴。

 ### 4.5 MQTT 連線認證
-機台連線 EMQX 時，使用 `serial_no` 作為 Username、`api_token` 作為 Password。驗證流程：
+機台連線 EMQX 時，使用 `serial_no` 作為 Username、`api_token` (SHA256) 作為 Password。驗證流程：

-1.  **Laravel 端 (Token 派發時)**：B014 下發 `api_token` 時，同步執行 `Redis::set("machine_auth:{serial_no}", hash(api_token))`。
-2.  **EMQX 端 (連線驗證時)**：配置 Redis Auth Plugin，直接查詢 Redis 進行極速驗證 (毫秒級)，不經過 MySQL。
-3.  **Token 更新/撤銷時**：Laravel 更新或刪除機台 Token 時，必須同步更新或刪除 Redis 中的對應快取。
+1.  **Laravel 端 (Token 派發時)**：B014 下發 `api_token` 時，同步執行 `Redis::hset("machine_auth:{serial_no}", "password", hash('sha256', $api_token))`。
+2.  **EMQX 端 (連線驗證時)**：配置 Redis Auth Plugin，使用 `HMGET machine_auth:${username} password` 查詢 Redis Hash 進行極速驗證 (毫秒級)，不經過 MySQL。
+3.  **Token 更新/撤銷時**：Laravel 更新或刪除機台 Token 時，必須同步更新或刪除 Redis 中的對應 Hash。
+4.  **Go Gateway 認證**：Gateway 使用固定帳號 `star-cloud-gateway` 連線 EMQX，其認證資料同樣以 Hash 結構存放於 Redis (`machine_auth:star-cloud-gateway`)。
+
+> [!CAUTION]
+> **認證資料結構為 Redis Hash (HSET)**，非 String (SET)。EMQX 5.x 預設的 Redis 認證插件使用 `HMGET` 指令讀取，若誤用 `SET` 儲存為 String 類型，認證將永遠失敗。
+> 批量同步指令：`php artisan mqtt:sync-auth`。

 ## 5. 開發標準 (Coding Standards)
 *   **命名規範**：
--- a/.agents/skills/iot-communication/SKILL.md
+++ b/.agents/skills/iot-communication/SKILL.md
@ -21,13 +21,16 @@ description: 規範智能販賣機與 Cloud 平台間的高頻通訊處理流程
 ### 1.2 MQTT 管線 (高頻/即時操作)
 適用於 B010 (心跳), B013 (異常), B600 (交易) 等高頻或即時性端點：
 1.  **Go Gateway (接收層)**：訂閱 EMQX Topic，提取 `serial_no`，包裝成標準 JSON。
-2.  **Redis List (橋接層)**：Go 執行 `RPUSH mqtt_incoming_jobs {json}`。
+2.  **Redis List (橋接層)**：Go 執行 `RPUSH starcloud_database_mqtt_incoming_jobs {json}`。
 3.  **Laravel `mqtt:listen` (消費層)**：常駐指令 `BLPOP` 取出 JSON，根據 `type` 分派至對應 Job。
 4.  **Job ➜ Service ➜ Model**：與 HTTP 管線後半段相同。

 > [!TIP]
 > 兩條管線的 **Job / Service / Model 層完全共用**，差異僅在「進入點」。這確保了業務邏輯不會因為通訊協議不同而分裂。

+> [!CAUTION]
+> **Redis Key 前綴一致性**：Laravel 的 Redis 驅動會自動在 Key 前加上 `starcloud_database_` 前綴。Go Gateway 不使用 Laravel Redis 客戶端，因此必須在程式碼中手動加上此前綴。另外，EMQX 認證使用 Redis Hash (`HSET`)，嚴禁使用 String (`SET`)。
+
 > [!IMPORTANT]
 > **嚴禁**在 API Controller 直接進行資料庫寫入操作（針對機台發訊端點）。

--- a/.agents/skills/mqtt-communication-specs/SKILL.md
+++ b/.agents/skills/mqtt-communication-specs/SKILL.md
@ -16,10 +16,23 @@ description: 定義 Star Cloud 與終端機台 (IoT) 之間的 MQTT 全域通訊

 ### 1.2 身份認證 (Authentication)
 機台連線時必須提供以下憑據：
- **Username**：機台編號 (`serial_no`)，例如 `M-001`。
- **Password**：機台正式 Token (`api_token`)，由 B014 API 取得。
+- **Username**：機台編號 (`serial_no`)，例如 `SN00001`。
+- **Password**：機台 `api_token` 的 SHA256 雜湊值，由 B014 API 取得。
 - **Client ID**：建議格式為 `SC_{serial_no}_{random_suffix}`，確保唯一性。

+**EMQX 認證機制**：
+- **資料結構**：Redis Hash (`HSET`)。Key 格式為 `{prefix}machine_auth:{username}`，包含 `password` 欄位。
+- **查詢指令**：EMQX 使用 `HMGET {prefix}machine_auth:${username} password` 進行驗證。
+- **Laravel 同步**：透過 `MachineService::syncMqttAuth()` 或批量指令 `php artisan mqtt:sync-auth` 維護。
+
+> [!CAUTION]
+> **嚴禁使用 Redis String (`SET`) 儲存認證資料**。EMQX 5.x 的 Redis Auth Plugin 僅支援 `HMGET` 讀取 Hash 結構。若誤用 String，認證將永遠失敗。
+
+### 1.3 Go Gateway 認證
+- **Username**：`star-cloud-gateway` (固定值)。
+- **Password**：與 Redis Hash `machine_auth:star-cloud-gateway` 中的 `password` 欄位對應。
+- Go Gateway 的認證資料須透過 Laravel 側寫入 Redis。
+
 ---

 ## 2. 主題架構 (Topic Topology)
@ -96,5 +109,44 @@ description: 定義 Star Cloud 與終端機台 (IoT) 之間的 MQTT 全域通訊

 ---

-## 5. 與 REST API 的同步關係
+## 5. Redis 橋接注意事項
+
+### 5.1 Key 前綴一致性
+Laravel 的 Redis 驅動會自動在 Key 前加上前綴 (預設 `starcloud_database_`)。Go Gateway 不使用 Laravel Redis 客戶端，因此**必須在程式碼中手動加上此前綴**：
+
+```go
+// main.go
+prefixedIncomingKey := "starcloud_database_" + cfg.IncomingQueueKey
+pusher := bridge.NewRedisPusher(rdb, prefixedIncomingKey)
+```
+
+> [!CAUTION]
+> 若 Go 與 Laravel 操作不同前綴的 Key，訊息將永遠無法傳達。修改 Laravel 的 `config/database.php` 中 Redis prefix 後，必須同步修改 Go 端的前綴字串。
+
+### 5.2 長連線穩定性
+Laravel `mqtt:listen` 使用 `BLPOP` 長時間阻塞等待。為防止 PHP 端主動關閉連線，必須在 `config/database.php` 設定：
+```php
+'redis' => [
+    'default' => [
+        'read_timeout' => -1,  // 永不超時
+        'timeout' => 10.0,     // 初始連線超時
+    ],
+],
+```
+
+---
+
+## 6. 與 REST API 的同步關係
 當 MQTT 通訊正常時，機台應停止定時呼叫 B010 HTTP API。若 MQTT 斷線超過 3 分鐘，則退回 (Fallback) 使用 HTTP 輪詢模式以維持基礎通訊。
+
+---
+
+## 7. 常用維運指令
+
+| 指令 | 用途 |
+|------|------|
+| `sail artisan mqtt:sync-auth` | 將所有機台認證資料從 MySQL 同步至 Redis Hash |
+| `sail artisan mqtt:listen` | 啟動 MQTT 訊息消費者 (常駐執行) |
+| `docker compose build mqtt-gateway` | 重新編譯 Go Gateway |
+| `docker restart star-cloud-mqtt-gateway` | 重啟 Gateway 容器 |
+| `docker restart star-cloud-queue` | 重啟 Queue Worker (程式碼變更後必須執行) |
--- a/README.md
+++ b/README.md
@ -9,20 +9,28 @@ Star Cloud 是一個專為智能販賣機設計的後台管理系統，負責管
 ## 🚀 技術架構

 ### 核心架構
-本專案採用 **傳統單體式架構 (Monolithic Architecture)**，結合 Laravel Blade 引擎進行伺服器端渲染 (SSR)。
+本專案採用 **Monorepo 單體式架構**，以 Laravel 為核心進行伺服器端渲染 (SSR) 與 API 服務，並搭配 **Go MQTT Gateway** 作為高併發 IoT 通訊的前置接收層。兩者透過 **Redis** 進行異步橋接。

 | 服務 | 容器名稱 | 技術 | 用途 | 本地 Port |
 |------|---------|------|------|--------|
 | **應用程式** | `star-cloud-laravel` | Laravel 12 + PHP 8.5 | 核心業務與渲染 | 8090 |
+| **背景隊列** | `star-cloud-queue` | Laravel Queue Worker | 異步任務處理 | - |
 | **資料庫** | `star-cloud-mysql` | MySQL 8.0 | 數據持久化 | 3306 |
-| **快取/隊列** | `star-cloud-redis` | Redis Alpine | IoT 高併發隊列 | 6380 |
+| **快取/隊列** | `star-cloud-redis` | Redis Alpine | IoT 高併發隊列與 MQTT 橋接 | 6380 |
+| **MQTT Broker** | `star-cloud-emqx` | EMQX 5.10 | 機台長連線與訊息路由 | 1883 / 18083 |
+| **MQTT Gateway** | `star-cloud-mqtt-gateway` | Go | 訂閱 MQTT Topic、轉發至 Redis | - |

 ### 後端技術棧
 - **Framework**: Laravel 12.x
 - **Language**: PHP 8.5
- **Redis**: 用於 IoT 高併發隊列 (B010, B600 等)
+- **Redis**: 用於 IoT 高併發隊列與 MQTT 橋接
 - **Database**: MySQL 8.0

+### IoT 通訊層
+- **MQTT Broker**: EMQX 5.10 (維持機台長連線與訊息路由)
+- **MQTT Gateway**: Go (訂閱 MQTT Topic、預處理訊息、轉發至 Redis)
+- **橋接機制**: Redis List (`mqtt_incoming_jobs`)，由 Laravel 常駐指令 (`mqtt:listen`) 消費
+
 ### 前端技術棧
 - **View**: Laravel Blade
 - **CSS**: Tailwind CSS + **Preline UI**
@ -48,6 +56,8 @@ Star Cloud 是一個專為智能販賣機設計的後台管理系統，負責管
 6. `./vendor/bin/sail artisan migrate --seed`
 7. `./vendor/bin/sail npm install`
 8. `./vendor/bin/sail npm run dev`
+9. `./vendor/bin/sail artisan mqtt:sync-auth` (同步機台 MQTT 認證至 Redis)
+10. `./vendor/bin/sail artisan mqtt:listen` (啟動 MQTT 訊息消費者)

 ### 常用開發指令
 | 功能 | 指令 |
@ -58,6 +68,9 @@ Star Cloud 是一個專為智能販賣機設計的後台管理系統，負責管
 | Composer 指令 | `./vendor/bin/sail composer <cmd>` |
 | NPM 指令 | `./vendor/bin/sail npm <cmd>` |
 | 執行測試 | `./vendor/bin/sail test` |
+| MQTT 認證同步 | `./vendor/bin/sail artisan mqtt:sync-auth` |
+| MQTT 隊列監聽 | `./vendor/bin/sail artisan mqtt:listen` |
+| 重建 Gateway | `docker compose build mqtt-gateway && docker compose up -d mqtt-gateway` |

 ---

@ -75,6 +88,11 @@ Star Cloud 是一個專為智能販賣機設計的後台管理系統，負責管
 - **認證**: Header `Authorization: Bearer <api_token>`。
 - **高併發處理**: 核心日誌 (B010 心跳、B600 交易) **嚴禁直寫 DB**，必須進入 **Redis Queue** 背景異步處理。

+### 3. MQTT IoT 通訊
+- **對象**: 高頻/即時性機台通訊 (心跳、異常、交易)。
+- **認證**: EMQX 透過 Redis Hash 驗證 (`machine_auth:{serial_no}` → `password` 欄位)。
+- **資料流向**: `機台 ➜ EMQX ➜ Go Gateway ➜ Redis List ➜ Laravel mqtt:listen ➜ Job ➜ MySQL`
+
 ---

 ## 🌐 多語系支援 (I18n)
@ -88,14 +106,22 @@ Star Cloud 是一個專為智能販賣機設計的後台管理系統，負責管

 ## 📂 目錄結構

+### Laravel (PHP)
 - `app/Http/Controllers/`: 控制器
 - `app/Models/{Domain}/`: 分領域的模型 (如 `Machine`, `Member`)
 - `app/Services/{Domain}/`: 封裝商業邏輯與資料異動
 - `app/Jobs/{Domain}/`: IoT 異步隊列處理任務 (重要)
+- `app/Console/Commands/`: Artisan 指令 (含 `mqtt:listen`, `mqtt:sync-auth`)
 - `resources/views/`: Blade 模板 (按功能分資料夾)
 - `resources/views/components/`: 可重用的 UI 組件
 - `routes/`: 路由定義 (`web.php` 與 `api.php`)

+### MQTT Gateway (Go)
+- `mqtt-gateway/main.go`: 進入點 (MQTT 連線、Redis 初始化)
+- `mqtt-gateway/config/`: 環境變數與連線設定
+- `mqtt-gateway/internal/handler/`: MQTT Topic 訊息處理邏輯
+- `mqtt-gateway/internal/bridge/`: Redis 橋接層 (RPush / Consumer)
+
 ---

 ## 🚢 CI/CD 與部署
--- a/app/Console/Commands/ListenMqttQueue.php
+++ b/app/Console/Commands/ListenMqttQueue.php
@ -0,0 +1,90 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\Redis;
+use Illuminate\Support\Facades\Log;
+use App\Jobs\Machine\ProcessHeartbeat;
+use App\Jobs\Machine\ProcessMachineError;
+use App\Jobs\Machine\ProcessTransaction;
+
+class ListenMqttQueue extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'mqtt:listen';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Listen to the MQTT incoming queue from Redis and dispatch jobs';
+
+    /**
+     * Execute the console command.
+     */
+    public function handle()
+    {
+        $queueKey = config('mqtt.incoming_queue', 'mqtt_incoming_jobs');
+        $this->info("Listening to MQTT queue: {$queueKey}...");
+
+        while (true) {
+            try {
+                // BLPOP returns [key, value]. Timeout set to 30s to avoid persistent connection read errors.
+                $result = Redis::blpop($queueKey, 30);
+                
+                if (!$result || !isset($result[1])) {
+                    continue;
+                }
+
+                $data = json_decode($result[1], true);
+                if (json_last_error() !== JSON_ERROR_NONE) {
+                    Log::error("MQTT Listen: Failed to decode JSON", ['data' => $result[1]]);
+                    continue;
+                }
+
+                $this->dispatchJob($data);
+
+            } catch (\Exception $e) {
+                $this->error("MQTT Listen Error: " . $e->getMessage());
+                Log::error("MQTT Listen Error: " . $e->getMessage(), ['exception' => $e]);
+                sleep(2); // 防止出錯時無限循環噴錯
+            }
+        }
+    }
+
+    /**
+     * 分派任務至對應的 Job
+     */
+    protected function dispatchJob(array $data)
+    {
+        $type = $data['type'] ?? '';
+        $serialNo = $data['serial_no'] ?? '';
+        $payload = $data['payload'] ?? [];
+
+        if (empty($serialNo)) {
+            Log::warning("MQTT Listen: Missing serial_no in message", ['data' => $data]);
+            return;
+        }
+
+        switch ($type) {
+            case 'heartbeat':
+                ProcessHeartbeat::dispatch($serialNo, $payload);
+                break;
+            case 'error':
+                ProcessMachineError::dispatch($serialNo, $payload);
+                break;
+            case 'transaction':
+                ProcessTransaction::dispatch($serialNo, $payload);
+                break;
+            default:
+                Log::notice("MQTT Listen: Unknown message type [{$type}]", ['data' => $data]);
+                break;
+        }
+    }
+}
--- a/app/Console/Commands/MqttSyncAuth.php
+++ b/app/Console/Commands/MqttSyncAuth.php
@ -0,0 +1,45 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use App\Models\Machine\Machine;
+use App\Services\Machine\MachineService;
+
+class MqttSyncAuth extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'mqtt:sync-auth';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Sync all machine API tokens to Redis for MQTT authentication';
+
+    /**
+     * Execute the console command.
+     */
+    public function handle(MachineService $machineService)
+    {
+        $machines = Machine::withoutGlobalScopes()->get();
+        $this->info("Syncing " . $machines->count() . " machines to Redis...");
+
+        $bar = $this->output->createProgressBar($machines->count());
+        $bar->start();
+
+        foreach ($machines as $machine) {
+            $machineService->syncMqttAuth($machine);
+            $bar->advance();
+        }
+
+        $bar->finish();
+        $this->newLine();
+        $this->info("Sync completed.");
+    }
+}
--- a/app/Jobs/Machine/ProcessHeartbeat.php
+++ b/app/Jobs/Machine/ProcessHeartbeat.php
@ -2,7 +2,7 @@

 namespace App\Jobs\Machine;

-use App\Services\Machine\MachineService;
+use App\Models\Machine\Machine;
 use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
@ -15,27 +15,45 @@ class ProcessHeartbeat implements ShouldQueue
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $serialNo;
-    protected $data;
+    protected $payload;

    /**
     * Create a new job instance.
     */
-    public function __construct(string $serialNo, array $data)
+    public function __construct(string $serialNo, array $payload)
    {
        $this->serialNo = $serialNo;
-        $this->data = $data;
+        $this->payload = $payload;
    }

    /**
     * Execute the job.
     */
-    public function handle(MachineService $machineService): void
+    public function handle(): void
    {
-        try {
-            $machineService->updateHeartbeat($this->serialNo, $this->data);
-        } catch (\Exception $e) {
-            Log::error("Failed to process heartbeat for machine {$this->serialNo}: " . $e->getMessage());
-            throw $e;
+        $machine = Machine::withoutGlobalScopes()->where('serial_no', $this->serialNo)->first();
+
+        if (!$machine) {
+            Log::warning("MQTT Heartbeat: Machine not found", ['serial_no' => $this->serialNo]);
+            return;
        }
+
+        // 更新機台狀態
+        $updateData = [
+            'last_heartbeat_at' => now(),
+        ];
+
+        // 根據 payload 動態更新欄位
+        if (isset($this->payload['current_page'])) {
+            $updateData['current_page'] = $this->payload['current_page'];
+        }
+        if (isset($this->payload['temperature'])) {
+            $updateData['temperature'] = $this->payload['temperature'];
+        }
+        if (isset($this->payload['firmware_version'])) {
+            $updateData['firmware_version'] = $this->payload['firmware_version'];
+        }
+
+        $machine->update($updateData);
    }
 }
--- a/app/Jobs/Machine/ProcessMachineError.php
+++ b/app/Jobs/Machine/ProcessMachineError.php
@ -3,38 +3,53 @@
 namespace App\Jobs\Machine;

 use App\Models\Machine\Machine;
-use App\Services\Machine\MachineService;
 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 ProcessMachineError implements ShouldQueue
 {
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $serialNo;
-    protected $data;
+    protected $payload;

    /**
     * Create a new job instance.
     */
-    public function __construct(string $serialNo, array $data)
+    public function __construct(string $serialNo, array $payload)
    {
        $this->serialNo = $serialNo;
-        $this->data = $data;
+        $this->payload = $payload;
    }

    /**
     * Execute the job.
     */
-    public function handle(MachineService $service): void
+    public function handle(): void
    {
-        $machine = Machine::where('serial_no', $this->serialNo)->first();
+        $machine = Machine::withoutGlobalScopes()->where('serial_no', $this->serialNo)->first();

-        if ($machine) {
-            $service->recordErrorLog($machine, $this->data);
+        if (!$machine) {
+            Log::warning("MQTT Error Report: Machine not found", ['serial_no' => $this->serialNo]);
+            return;
        }
+
+        $message = $this->payload['message'] ?? 'Unknown error';
+        $errorCode = $this->payload['error_code'] ?? 'E000';
+        $level = $this->payload['level'] ?? 'error';
+
+        // 記錄到機台日誌
+        ProcessStateLog::dispatch(
+            $machine->id,
+            $machine->company_id,
+            "[{$errorCode}] {$message}",
+            $level,
+            $this->payload,
+            'error'
+        );
    }
 }
--- a/app/Jobs/Machine/ProcessTransaction.php
+++ b/app/Jobs/Machine/ProcessTransaction.php
@ -0,0 +1,57 @@
+<?php
+
+namespace App\Jobs\Machine;
+
+use App\Models\Machine\Machine;
+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 ProcessTransaction implements ShouldQueue
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+
+    protected $serialNo;
+    protected $payload;
+
+    /**
+     * Create a new job instance.
+     */
+    public function __construct(string $serialNo, array $payload)
+    {
+        $this->serialNo = $serialNo;
+        $this->payload = $payload;
+    }
+
+    /**
+     * Execute the job.
+     */
+    public function handle(): void
+    {
+        $machine = Machine::withoutGlobalScopes()->where('serial_no', $this->serialNo)->first();
+
+        if (!$machine) {
+            Log::warning("MQTT Transaction: Machine not found", ['serial_no' => $this->serialNo]);
+            return;
+        }
+
+        // TODO: 實作交易處理邏輯 (建立交易紀錄、扣庫存等)
+        // 目前先記錄 log 以供驗證
+        Log::info("MQTT Transaction received", [
+            'serial_no' => $this->serialNo,
+            'payload' => $this->payload
+        ]);
+
+        ProcessStateLog::dispatch(
+            $machine->id,
+            $machine->company_id,
+            "Transaction processed: " . ($this->payload['transaction_id'] ?? 'N/A'),
+            'info',
+            $this->payload,
+            'transaction'
+        );
+    }
+}
--- a/app/Models/Machine/Machine.php
+++ b/app/Models/Machine/Machine.php
@ -30,6 +30,13 @@ class Machine extends Model
                    ->where('machine_user.user_id', $user->id);
            });
        });
+
+        // 當 api_token 發生變動時，自動同步至 Redis (供 MQTT 認證使用)
+        static::saved(function ($machine) {
+            if ($machine->wasChanged('api_token') || $machine->wasRecentlyCreated) {
+                app(\App\Services\Machine\MachineService::class)->syncMqttAuth($machine);
+            }
+        });
    }

    protected $fillable = [
--- a/app/Services/Machine/MachineService.php
+++ b/app/Services/Machine/MachineService.php
@ -105,6 +105,32 @@ class MachineService
        });
    }

+    /**
+     * Sync machine API token to Redis for MQTT authentication.
+     * 
+     * @param Machine $machine
+     */
+    public function syncMqttAuth(Machine $machine): void
+    {
+        if (empty($machine->api_token)) {
+            return;
+        }
+
+        // MQTT 連線認證：Username = serial_no, Password = hash(api_token)
+        // 遵循 framework.md 4.5 規範
+        $redisKey = "machine_auth:{$machine->serial_no}";
+        
+        // 這裡採用 SHA256 雜湊，與 EMQX 設定對應
+        $hashedToken = hash('sha256', $machine->api_token);
+        
+        \Illuminate\Support\Facades\Redis::hSet($redisKey, 'password', $hashedToken);
+        
+        \Illuminate\Support\Facades\Log::info("MQTT Auth synced to Redis", [
+            'serial_no' => $machine->serial_no,
+            'key' => $redisKey
+        ]);
+    }
+
    /**
     * Sync machine slots based on replenishment report.
     * 
--- a/app/Services/Machine/MqttCommandService.php
+++ b/app/Services/Machine/MqttCommandService.php
@ -0,0 +1,35 @@
+<?php
+
+namespace App\Services\Machine;
+
+use Illuminate\Support\Facades\Redis;
+use Illuminate\Support\Str;
+
+class MqttCommandService
+{
+    /**
+     * 發送指令至機台
+     *
+     * @param string $serialNo 機台序號
+     * @param string $command 指令名稱 (例如 dispense, reboot)
+     * @param array $payload 指令參數
+     * @return string Message ID
+     */
+    public function sendCommand(string $serialNo, string $command, array $payload = []): string
+    {
+        $messageId = 'MSG_' . Str::random(16);
+        $queueKey = config('mqtt.outgoing_queue', 'mqtt_outgoing_commands');
+
+        $data = [
+            'target' => $serialNo,
+            'command' => $command,
+            'payload' => $payload,
+            'message_id' => $messageId,
+            'timestamp' => now()->toIso8601String(),
+        ];
+
+        Redis::rpush($queueKey, json_encode($data));
+
+        return $messageId;
+    }
+}
--- a/compose.yaml
+++ b/compose.yaml
@ -91,6 +91,42 @@ services:
                - ping
            retries: 3
            timeout: 5s
+    emqx:
+        image: 'emqx/emqx:5.10.3'
+        container_name: star-cloud-emqx
+        hostname: emqx
+        ports:
+            - '${MQTT_PORT:-1883}:1883'
+            - '${EMQX_DASHBOARD_PORT:-18083}:18083'
+            - '8083:8083'
+        environment:
+            TZ: 'Asia/Taipei'
+            EMQX_DASHBOARD__BOOTSTRAP_USER_PASS: 'Star82779061'
+        volumes:
+            - 'sail-emqx:/opt/emqx/data'
+        networks:
+            - sail
+        depends_on:
+            - redis
+    mqtt-gateway:
+        build:
+            context: ./mqtt-gateway
+            dockerfile: Dockerfile
+        container_name: star-cloud-mqtt-gateway
+        hostname: mqtt-gateway
+        environment:
+            MQTT_BROKER_ADDR: 'tcp://emqx:1883'
+            MQTT_GATEWAY_CLIENT_ID: 'star-cloud-gateway'
+            MQTT_REDIS_ADDR: 'star-cloud-redis:6379'
+            MQTT_REDIS_PASSWORD: '${REDIS_PASSWORD:-}'
+            MQTT_INCOMING_QUEUE: 'mqtt_incoming_jobs'
+            MQTT_OUTGOING_QUEUE: 'mqtt_outgoing_commands'
+            TZ: 'Asia/Taipei'
+        networks:
+            - sail
+        depends_on:
+            - emqx
+            - redis
    # meilisearch:
    #     image: 'getmeili/meilisearch:latest'
    #     ports:
@ -133,5 +169,7 @@ volumes:
        driver: local
    sail-redis:
        driver: local
+    sail-emqx:
+        driver: local
    # sail-meilisearch:
    #     driver: local
--- a/config/database.php
+++ b/config/database.php
@ -136,6 +136,8 @@ return [
            'password' => env('REDIS_PASSWORD'),
            'port' => env('REDIS_PORT', '6379'),
            'database' => env('REDIS_DB', '0'),
+            'read_timeout' => -1,
+            'timeout' => env('REDIS_TIMEOUT', 10.0),
        ],

        'cache' => [
@ -145,6 +147,8 @@ return [
            'password' => env('REDIS_PASSWORD'),
            'port' => env('REDIS_PORT', '6379'),
            'database' => env('REDIS_CACHE_DB', '1'),
+            'read_timeout' => -1,
+            'timeout' => env('REDIS_TIMEOUT', 10.0),
        ],

    ],
--- a/config/mqtt.php
+++ b/config/mqtt.php
@ -0,0 +1,6 @@
+<?php
+
+return [
+    'incoming_queue' => env('MQTT_INCOMING_QUEUE', 'mqtt_incoming_jobs'),
+    'outgoing_queue' => env('MQTT_OUTGOING_QUEUE', 'mqtt_outgoing_commands'),
+];
--- a/mqtt-gateway/Dockerfile
+++ b/mqtt-gateway/Dockerfile
@ -0,0 +1,21 @@
+# Stage 1: Build
+FROM golang:1.23-alpine AS builder
+
+WORKDIR /app
+
+# 複製依賴檔案
+COPY go.mod ./
+# 註：如果沒有 go.sum，build 時會自動處理或報錯，我們先建立 basic
+RUN go mod download || true
+
+COPY . .
+RUN CGO_ENABLED=0 go build -o gateway .
+
+# Stage 2: Run
+FROM alpine:3.20
+
+WORKDIR /root/
+COPY --from=builder /app/gateway .
+
+# 運行
+CMD ["./gateway"]
--- a/mqtt-gateway/config/config.go
+++ b/mqtt-gateway/config/config.go
@ -0,0 +1,32 @@
+package config
+
+import (
+	"os"
+)
+
+type Config struct {
+	MQTTAddr         string
+	MQTTClientID     string
+	RedisAddr        string
+	RedisPassword    string
+	IncomingQueueKey string
+	OutgoingQueueKey string
+}
+
+func LoadConfig() *Config {
+	return &Config{
+		MQTTAddr:         getEnv("MQTT_BROKER_ADDR", "tcp://emqx:1883"),
+		MQTTClientID:     getEnv("MQTT_GATEWAY_CLIENT_ID", "star-cloud-gateway"),
+		RedisAddr:        getEnv("MQTT_REDIS_ADDR", "star-cloud-redis:6379"),
+		RedisPassword:    getEnv("MQTT_REDIS_PASSWORD", ""),
+		IncomingQueueKey: getEnv("MQTT_INCOMING_QUEUE", "mqtt_incoming_jobs"),
+		OutgoingQueueKey: getEnv("MQTT_OUTGOING_QUEUE", "mqtt_outgoing_commands"),
+	}
+}
+
+func getEnv(key, fallback string) string {
+	if value, ok := os.LookupEnv(key); ok {
+		return value
+	}
+	return fallback
+}
--- a/mqtt-gateway/go.mod
+++ b/mqtt-gateway/go.mod
@ -0,0 +1,16 @@
+module star-cloud-gateway
+
+go 1.23
+
+require (
+	github.com/eclipse/paho.mqtt.golang v1.5.0
+	github.com/redis/go-redis/v9 v9.7.0
+)
+
+require (
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
+	golang.org/x/net v0.27.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
+)
--- a/mqtt-gateway/go.sum
+++ b/mqtt-gateway/go.sum
@ -0,0 +1,18 @@
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
+github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
+github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
+golang.org/x/net v0.27.0 h1:5K3Njcw06/l2y9vpGCSdcxWOYHOUk3dVNGDXN+FvAys=
+golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE=
+golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
+golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
--- a/mqtt-gateway/internal/bridge/redis_consumer.go
+++ b/mqtt-gateway/internal/bridge/redis_consumer.go
@ -0,0 +1,62 @@
+package bridge
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/redis/go-redis/v9"
+)
+
+type RedisConsumer struct {
+	rdb        *redis.Client
+	mqttClient mqtt.Client
+	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}
+}
+
+func (c *RedisConsumer) Start(ctx context.Context) {
+	log.Printf("Redis Consumer started, listening on queue: %s", c.queueKey)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			// 使用 BLPOP 阻塞式讀取，逾時設為 0 表示永久等待
+			result, err := c.rdb.BLPop(ctx, 0, c.queueKey).Result()
+			if err != nil {
+				if err != context.Canceled {
+					log.Printf("Redis BLPop error: %v", err)
+				}
+				continue
+			}
+
+			// result[0] 是 key，result[1] 是 value
+			var cmd CommandPayload
+			err = json.Unmarshal([]byte(result[1]), &cmd)
+			if err != nil {
+				log.Printf("Failed to unmarshal command: %v", err)
+				continue
+			}
+
+			topic := "machine/" + cmd.Target + "/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)
+		}
+	}
+}
--- a/mqtt-gateway/internal/bridge/redis_pusher.go
+++ b/mqtt-gateway/internal/bridge/redis_pusher.go
@ -0,0 +1,41 @@
+package bridge
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+type RedisPusher struct {
+	client *redis.Client
+	key    string
+}
+
+type BridgePayload struct {
+	Type       string      `json:"type"`
+	SerialNo   string      `json:"serial_no"`
+	Payload    interface{} `json:"payload"`
+	ReceivedAt string      `json:"received_at"`
+}
+
+func NewRedisPusher(client *redis.Client, key string) *RedisPusher {
+	return &RedisPusher{client: client, key: key}
+}
+
+func (p *RedisPusher) Push(payload BridgePayload) error {
+	payload.ReceivedAt = time.Now().Format(time.RFC3339)
+	data, err := json.Marshal(payload)
+	if err != nil {
+		return err
+	}
+
+	ctx := context.Background()
+	err = p.client.RPush(ctx, p.key, data).Err()
+	if err != nil {
+		log.Printf("Failed to push to Redis: %v", err)
+	}
+	return err
+}
--- a/mqtt-gateway/internal/handler/mqtt_handler.go
+++ b/mqtt-gateway/internal/handler/mqtt_handler.go
@ -0,0 +1,48 @@
+package handler
+
+import (
+	"encoding/json"
+	"log"
+	"star-cloud-gateway/internal/bridge"
+	"strings"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+)
+
+type MqttHandler struct {
+	pusher *bridge.RedisPusher
+}
+
+func NewMqttHandler(pusher *bridge.RedisPusher) *MqttHandler {
+	return &MqttHandler{pusher: pusher}
+}
+
+func (h *MqttHandler) HandleMessage(client mqtt.Client, msg mqtt.Message) {
+	topic := msg.Topic()
+	payload := msg.Payload()
+
+	log.Printf("Received message on topic: %s", topic)
+
+	// 從 Topic 提取 SerialNo (架構: machine/{serial_no}/{type})
+	parts := strings.Split(topic, "/")
+	if len(parts) < 3 {
+		return
+	}
+	serialNo := parts[1]
+	msgType := parts[2]
+
+	var jsonPayload interface{}
+	err := json.Unmarshal(payload, &jsonPayload)
+	if err != nil {
+		log.Printf("Invalid JSON payload on topic %s: %v", topic, err)
+		return
+	}
+
+	bridgePayload := bridge.BridgePayload{
+		Type:     msgType,
+		SerialNo: serialNo,
+		Payload:  jsonPayload,
+	}
+
+	h.pusher.Push(bridgePayload)
+}
--- a/mqtt-gateway/main.go
+++ b/mqtt-gateway/main.go
@ -0,0 +1,74 @@
+package main
+
+import (
+	"context"
+	"log"
+	"os"
+	"os/signal"
+	"star-cloud-gateway/config"
+	"star-cloud-gateway/internal/bridge"
+	"star-cloud-gateway/internal/handler"
+	"syscall"
+
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"github.com/redis/go-redis/v9"
+)
+
+func main() {
+	cfg := config.LoadConfig()
+
+	// 1. 初始化 Redis
+	rdb := redis.NewClient(&redis.Options{
+		Addr:     cfg.RedisAddr,
+		Password: cfg.RedisPassword,
+		DB:       0,
+	})
+	if err := rdb.Ping(context.Background()).Err(); err != nil {
+		log.Fatalf("Failed to connect to Redis at %s: %v", cfg.RedisAddr, err)
+	}
+	log.Printf("Connected to Redis at %s", cfg.RedisAddr)
+
+	// 加入 Laravel 前綴，確保雙方桶子同一個
+	prefixedIncomingKey := "starcloud_database_" + cfg.IncomingQueueKey
+	pusher := bridge.NewRedisPusher(rdb, prefixedIncomingKey)
+	mqttHandler := handler.NewMqttHandler(pusher)
+
+	// 2. 初始化 MQTT
+	opts := mqtt.NewClientOptions()
+	opts.AddBroker(cfg.MQTTAddr)
+	opts.SetClientID(cfg.MQTTClientID)
+	opts.SetUsername("star-cloud-gateway")
+	opts.SetPassword("StarCloudSecret999") // 與 Redis 中的 hash 對應
+	opts.SetCleanSession(false)
+	opts.SetOnConnectHandler(func(c mqtt.Client) {
+		log.Printf("Connected to MQTT Broker as Gateway [star-cloud-gateway] at %s", cfg.MQTTAddr)
+		// 訂閱所有機台的上行 Topic
+		token := c.Subscribe("machine/+/+", 1, mqttHandler.HandleMessage)
+		token.Wait()
+		if token.Error() != nil {
+			log.Printf("Failed to subscribe to machine/+/+: %v", token.Error())
+		} else {
+			log.Println("Successfully subscribed to machine/+/+")
+		}
+	})
+
+	client := mqtt.NewClient(opts)
+	if token := client.Connect(); token.Wait() && token.Error() != nil {
+		log.Fatalf("Failed to connect to MQTT: %v", token.Error())
+	}
+
+	// 3. 啟動 Redis Consumer (下行指令)
+	consumer := bridge.NewRedisConsumer(rdb, client, cfg.OutgoingQueueKey)
+	ctx, cancel := context.WithCancel(context.Background())
+	go consumer.Start(ctx)
+
+	// 4. 阻塞並監聽信號
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+	<-sigChan
+
+	log.Println("Shutting down Gateway...")
+	cancel() // 停止 Consumer
+	client.Disconnect(250)
+	rdb.Close()
+}
--- a/simulate-heartbeat.sh
+++ b/simulate-heartbeat.sh
@ -0,0 +1,40 @@
+#!/bin/bash
+
+# =================================================================
+# Star Cloud MQTT Heartbeat Simulator
+# =================================================================
+# 用於模擬機台每 10 秒發送一次心跳包。
+# 依賴：Docker (使用 eclipse-mosquitto 鏡像進行發送)
+# =================================================================
+
+SERIAL_NO="SN00001"
+API_TOKEN="8QwOrmKX6sLmR5T1sChfvGjfc8eS4qaclOFd4ocItmeryii7SFqcC5uMpnzg"
+TOPIC="machine/${SERIAL_NO}/heartbeat"
+NETWORK="star-cloud_sail"
+BROKER_HOST="emqx"
+
+echo "🚀 開始模擬心跳發報 [${SERIAL_NO}]..."
+echo "📡 頻率：每 10 秒一次"
+echo "📝 Topic: ${TOPIC}"
+echo "-------------------------------------------------"
+
+while true; do
+    TIMESTAMP=$(date +"%Y-%m-%dT%H:%M:%S+08:00")
+    PAYLOAD="{\"current_page\":2,\"firmware_version\":\"1.0.6\",\"temperature\":$(awk "BEGIN {srand(); print 20+rand()*10}")}"
+    
+    echo "[${TIMESTAMP}] 正在發送心跳..."
+    
+    docker run --rm --network ${NETWORK} eclipse-mosquitto \
+        mosquitto_pub -h ${BROKER_HOST} -p 1883 \
+        -u "${SERIAL_NO}" -P "${API_TOKEN}" \
+        -t "${TOPIC}" -m "${PAYLOAD}"
+        
+    if [ $? -eq 0 ]; then
+        echo "✅ 發送成功"
+    else
+        echo "❌ 發送失敗，請檢查 Docker 網路或 EMQX 狀態"
+    fi
+    
+    echo "💤 等待 10 秒..."
+    sleep 10
+done