diff --git a/.agents/rules/framework.md b/.agents/rules/framework.md index f013c68..ed45278 100644 --- 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) * **命名規範**： diff --git a/.agents/skills/iot-communication/SKILL.md b/.agents/skills/iot-communication/SKILL.md index 4996396..715e846 100644 --- 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 直接進行資料庫寫入操作（針對機台發訊端點）。 diff --git a/.agents/skills/mqtt-communication-specs/SKILL.md b/.agents/skills/mqtt-communication-specs/SKILL.md index 1d6d7d8..a606f5a 100644 --- 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 (程式碼變更後必須執行) | diff --git a/README.md b/README.md index 7511fe3..8114e55 100644 --- 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 ` | | NPM 指令 | `./vendor/bin/sail npm ` | | 執行測試 | `./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 `。 - **高併發處理**: 核心日誌 (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 與部署 diff --git a/app/Console/Commands/ListenMqttQueue.php b/app/Console/Commands/ListenMqttQueue.php new file mode 100644 index 0000000..5f18eb5 --- /dev/null +++ b/app/Console/Commands/ListenMqttQueue.php @@ -0,0 +1,90 @@ +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; + } + } +} diff --git a/app/Console/Commands/MqttSyncAuth.php b/app/Console/Commands/MqttSyncAuth.php new file mode 100644 index 0000000..4f8910d --- /dev/null +++ b/app/Console/Commands/MqttSyncAuth.php @@ -0,0 +1,45 @@ +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."); + } +} diff --git a/app/Jobs/Machine/ProcessHeartbeat.php b/app/Jobs/Machine/ProcessHeartbeat.php index 92a68e8..dced146 100644 --- 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); } } diff --git a/app/Jobs/Machine/ProcessMachineError.php b/app/Jobs/Machine/ProcessMachineError.php index a5ed78c..1b13a29 100644 --- 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' + ); } } diff --git a/app/Jobs/Machine/ProcessTransaction.php b/app/Jobs/Machine/ProcessTransaction.php new file mode 100644 index 0000000..2870d0d --- /dev/null +++ b/app/Jobs/Machine/ProcessTransaction.php @@ -0,0 +1,57 @@ +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' + ); + } +} diff --git a/app/Models/Machine/Machine.php b/app/Models/Machine/Machine.php index 14638d4..478faa2 100644 --- 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 = [ diff --git a/app/Services/Machine/MachineService.php b/app/Services/Machine/MachineService.php index 17205ab..2cb3677 100644 --- 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. * diff --git a/app/Services/Machine/MqttCommandService.php b/app/Services/Machine/MqttCommandService.php new file mode 100644 index 0000000..9b2e2be --- /dev/null +++ b/app/Services/Machine/MqttCommandService.php @@ -0,0 +1,35 @@ + $serialNo, + 'command' => $command, + 'payload' => $payload, + 'message_id' => $messageId, + 'timestamp' => now()->toIso8601String(), + ]; + + Redis::rpush($queueKey, json_encode($data)); + + return $messageId; + } +} diff --git a/compose.yaml b/compose.yaml index 82e774f..91f9b07 100644 --- 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 diff --git a/config/database.php b/config/database.php index 84b127a..7afb09e 100644 --- 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), ], ], diff --git a/config/mqtt.php b/config/mqtt.php new file mode 100644 index 0000000..ba3782b --- /dev/null +++ b/config/mqtt.php @@ -0,0 +1,6 @@ + env('MQTT_INCOMING_QUEUE', 'mqtt_incoming_jobs'), + 'outgoing_queue' => env('MQTT_OUTGOING_QUEUE', 'mqtt_outgoing_commands'), +]; diff --git a/mqtt-gateway/Dockerfile b/mqtt-gateway/Dockerfile new file mode 100644 index 0000000..3806ae8 --- /dev/null +++ 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"] diff --git a/mqtt-gateway/config/config.go b/mqtt-gateway/config/config.go new file mode 100644 index 0000000..910664b --- /dev/null +++ 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 +} diff --git a/mqtt-gateway/go.mod b/mqtt-gateway/go.mod new file mode 100644 index 0000000..a25e873 --- /dev/null +++ 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 +) diff --git a/mqtt-gateway/go.sum b/mqtt-gateway/go.sum new file mode 100644 index 0000000..7890708 --- /dev/null +++ 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= diff --git a/mqtt-gateway/internal/bridge/redis_consumer.go b/mqtt-gateway/internal/bridge/redis_consumer.go new file mode 100644 index 0000000..0bf4e3a --- /dev/null +++ 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) + } + } +} diff --git a/mqtt-gateway/internal/bridge/redis_pusher.go b/mqtt-gateway/internal/bridge/redis_pusher.go new file mode 100644 index 0000000..3448046 --- /dev/null +++ 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 +} diff --git a/mqtt-gateway/internal/handler/mqtt_handler.go b/mqtt-gateway/internal/handler/mqtt_handler.go new file mode 100644 index 0000000..5c49289 --- /dev/null +++ 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) +} diff --git a/mqtt-gateway/main.go b/mqtt-gateway/main.go new file mode 100644 index 0000000..11e7cc1 --- /dev/null +++ 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() +} diff --git a/simulate-heartbeat.sh b/simulate-heartbeat.sh new file mode 100755 index 0000000..a84d36b --- /dev/null +++ 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