---
name: MQTT 即時通訊與 Topic 規範
description: 定義 Star Cloud 與終端機台 (IoT) 之間的 MQTT 全域通訊拓撲、主題 (Topic) 結構、資料載體 (Payload) 格式與安全性認證機制。
---

# MQTT 即時通訊與 Topic 規範 (MQTT Protocol Specs)

本文件定義機台與 Star Cloud 之間的高併發即時通訊標準。MQTT 主要用於處理高頻率且對即時性要求高的訊息（如心跳、遠端指令），與 HTTP REST API 形成雙軌互補架構。

## 1. 連線基礎設定 (Connection Basics)

### 1.1 Broker 資訊
- **協議版本**：MQTT v3.1.1 (相容性最高)
- **Keep-Alive**：建議設定為 30 ~ 60 秒。

**各連線通道埠號：**

| 通道 | 埠號 | 用途 | 使用者 |
| :--- | :--- | :--- | :--- |
| TCP | 1883 | Docker 內部 / 區網測試 | Go Gateway、開發除錯 |
| SSL/TLS | 8883 | 加密 TCP (正式環境) | Go Gateway (選配) |
| WebSocket | 8083 | EMQX 原生 WS 監聽 | NPM 反向代理目標 |
| **WSS (公網)** | **443** | App 端透過 NPM 反代連線 | **Android / iOS App** |

### 1.2 身份認證與識別 (Authentication & Identity)
機台連線時必須提供以下憑據：
- **Username**：機台編號 (`serial_no`)，例如 `SN00001`。
- **Password**：機台 `api_token` 的 SHA256 雜湊值，由 B014 API 取得。
- **Client ID**：格式建議為 **`SC_{serial_no}_{random_suffix}`**。
    - **解析邏輯**：系統接收連線事件時，會自動移除 `SC_` 前綴，並移除第一個底線後的所有隨機字串，還原出純淨的 `serial_no`。
    - **範例**：`SC_M0408_pndk` ➜ 解析為 `M0408`。

**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 連線狀態即時同步 ($SYS Events)
系統透過訂閱 EMQX 的系統主題來實現「毫秒級」的連線狀態更新，不需等待心跳包：
- **訂閱主題**：
    - `$SYS/brokers/+/clients/+/connected` ➜ 標記為 `online`
    - `$SYS/brokers/+/clients/+/disconnected` ➜ 標記為 `offline`
- **處理流程**：Go Gateway 接收 ➜ 解析 Client ID 為 Serial No ➜ 推送 `type: status` 至 Redis ➜ Laravel `ProcessStatus` Job 更新資料庫與 `last_heartbeat_at`。

### 1.4 Go Gateway 認證
- **Username**：`star-cloud-gateway` (固定值)。
- **Password**：與 Redis Hash `machine_auth:star-cloud-gateway` 中的 `password` 欄位對應。
- Go Gateway 的認證資料須透過 Laravel 側寫入 Redis。

---

## 1-A. App 端 WSS 連線規範 (WebSocket Secure)

App (Android) 透過 **WSS** 穿越公網連線 Broker。路徑：`App → Cloudflare → NPM (port 8083 反代) → EMQX`。

### 連線參數

| 參數 | 值 |
| :--- | :--- |
| **URL** | `wss://mqtt.setdomains.work:443/mqtt` |
| **Username** | `{machineID}` (即 `serial_no`) |
| **Password** | `{api_token}` 原始值（B014 取得，**不做 SHA256**） |
| **Client ID** | `SC_{machineID}_{random}` |
| **Clean Session** | `true` |

> [!CAUTION]
> App 傳入原始 Token，EMQX 端的 `password_hash_algorithm` 須設為 `sha256`，由 Broker 自動雜湊比對 Redis 中的值。

### Android 範例 (HiveMQ)

```java
Mqtt3Client client = MqttClient.builder()
    .useMqttVersion3()
    .identifier("SC_" + machineID + "_" + randomSuffix)
    .serverHost("mqtt.setdomains.work")
    .serverPort(443)
    .sslWithDefaultConfig()
    .webSocketConfig().serverPath("/mqtt").applyWebSocketConfig()
    .simpleAuth().username(machineID).password(apiToken.getBytes()).applySimpleAuth()
    .willPublish()
        .topic("machine/" + machineID + "/heartbeat")
        .payload("{\"status\":\"offline\"}".getBytes())
        .applyWillPublish()
    .buildBlocking();
```

### Fallback 策略
- **指數退避重連**：1s → 2s → 4s → ... → 最大 60s。
- WSS 斷線超過 **3 分鐘** → 退回 HTTP 輪詢 (B010)，背景每 60 秒嘗試重連 WSS。
- WSS 恢復 → 立即切回 MQTT，停止 HTTP 輪詢。

---

## 2. 主題架構 (Topic Topology)

我們採用目錄式的層級結構，方便未來進行萬台設備的管理與 ACL 權限切分。

| 主題名稱 (Topic) | 方向 | QoS | 用途說明 |
| :--- | :--- | :--- | :--- |
| `machine/{serial_no}/heartbeat` | 設備 ➜ 雲端 | 0 | 每 1 分鐘上報狀態大包裝（溫度、韌體版本等）。 |
| `machine/{serial_no}/error` | 設備 ➜ 雲端 | 1 | 發生硬體故障、卡貨或門未關時立即上報。 |
| `machine/{serial_no}/transaction` | 設備 ➜ 雲端 | 1 | 交易生命週期上報 (B600 建立 / B601 發票 / B602 出貨結果)，透過 `action` 欄位區分。 |
| `machine/{serial_no}/command` | 雲端 ➜ 設備 | 1 | 雲端下發的即時指令（出貨、更新、重啟）。 |
| `machine/{serial_no}/command/ack` | 設備 ➜ 雲端 | 1 | 機台回覆指令執行結果（B055 出貨回報）。 |

---

## 3. 資料載體規範 (Payload Definitions)

所有 Payload 統一採用 **JSON** 格式，字母一律為 **snake_case**。

### 3.1 心跳上報 (Heartbeat) - `machine/{serial_no}/heartbeat`
這是機台最主要的狀態上報包，整合了關鍵的機台資訊。

```json
{
  "firmware_version": "2.1.6",
  "temperature": 20
}
```

> [!NOTE]
> 為了降低傳輸開銷，目前僅要求上報 **韌體版本** 與 **溫度** (整數型態)。其餘狀態如連線/斷線已透過 LWT 與 $SYS 事件處理。
> 雲端對於溫度的日誌記錄規則：**只要有變動 (current !== last) 即寫入日誌**，並維持每 4 小時一次的保底記錄。

### 3.2 異常與錯誤上報 (B013)
當機台發生硬體故障（如卡貨、通訊中斷）時，發布至此 Topic。

- **Topic**: `machine/{serial_no}/error`
- **Payload (JSON)**:
  ```json
  {
    "tid": 12,
    "error_code": "0403"
  }
  ```
#### B013 硬體代碼對照表 (由 MachineService 自動翻譯)

| 代碼 | 英文 Key (i18n) | 級別 | 範例繁中翻譯 |
| :--- | :--- | :--- | :--- |
| **0402** | Dispense successful | info | 出貨成功 |
| **0403** | Slot jammed | error | **貨道卡貨** (重大異常) |
| **0202** | Product empty | warning | 貨道缺貨 |
| **0412** | Elevator rise error | error | 昇降機上升異常 |
| **0415** | Pickup door error | error | 取貨門異常 |
| **5402** | Pickup door not closed | warning | **取貨門未關** (警告) |
| **5403** | Elevator failure | error | 昇降系統故障 |

*(註：`error_code` 應遵循此表定義，雲端將根據 `i18n` 自動轉換。`tid` 為貨道或任務 ID。)*

### 3.3 交易生命週期上報 (Transaction) - `machine/{serial_no}/transaction`

此 Topic 整合了交易生命週期的三個階段，透過 `action` 欄位區分：

#### B600 交易建立 (`action: create`)
```json
{
  "action": "create",
  "flow_id": "T202604140001",
  "payment_type": 1,
  "total_amount": 25.0,
  "discount_amount": 0.0,
  "pay_amount": 25.0,
  "items": [
    {
      "product_id": "5",
      "product_name": "Coca Cola",
      "price": 25.0,
      "quantity": 1
    }
  ]
}
```

#### B601 發票紀錄 (`action: invoice`)
```json
{
  "action": "invoice",
  "flow_id": "T202604140001",
  "invoice_no": "AB12345678",
  "invoice_date": "2026-04-22",
  "amount": 25.0,
  "random_no": "1234",
  "love_code": ""
}
```

#### B602 出貨結果 (`action: dispense`)
```json
{
  "action": "dispense",
  "flow_id": "T202604140001",
  "slot_no": "3",
  "product_id": "5",
  "amount": 25.0,
  "dispense_status": 1,
  "remaining_stock": 9,
  "machine_time": "2026-04-22 14:30:00"
}
```

> [!NOTE]
> - `serial_no` 不需放在 Payload 中，Go Gateway 自動從 Topic 路徑提取並注入。
> - 若 Payload 不含 `action` 欄位，系統預設視為 `create`（向後相容既有 B600 格式）。

### 3.4 雲端指令 (Downstream Commands) - `machine/{serial_no}/command`
這是雲端主動下發給機台的訊息，取代原本 B010 Response 的輪詢等待。

```json
{
  "command": "dispense",
  "command_id": "20",
  "payload": {
    "new_stock": 6,
    "old_stock": 7,
    "slot_no": "1"
  }
}
```
**常用指令集：**
- `reboot`: 遠端重啟機台系統。
- `reboot_card`: 遠端重啟讀卡機。
- `checkout`: 遠端結帳。
- `lock`: 鎖定頁鎖定。
- `unlock`: 鎖定頁解鎖。
- `change`: 遠端找零（夾帶 amount 參數）。
- `update_inventory`: 遠端即時修改庫存。
- `dispense`: 遠端出貨指令。

**`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 中發送，以降低連線開銷並簡化後端處理。

> [!IMPORTANT]
> **集中化原則**：除非極特殊的即時性需求，否則所有機台狀態（溫度、版本、門狀態、頁面碼）應統一包在 Heartbeat 訊息中。

---

## 4. 安全與 QOS 規範

1. **存取控制 (ACL)**：EMQX Broker 必須設定 ACL，禁止機台 A 訂閱機台 B 的 Topic。機台僅能訂閱與發布包含自身 `serial_no` 的路徑。
2. **QoS 策略**：
   - **QoS 0**：適用於高頻率心跳，即使掉一兩次包也不影響系統判斷。
   - **QoS 1**：適用於交易與指令，確保「至少送達一次」。App 端收到指令後應回覆回執。
3. **遺囑訊息 (Last Will and Testament)**：
   機台 Connect 時應設定 Last Will 於 `machine/{serial_no}/heartbeat`，Payload 為 `{"status": "offline"}`。當連線異常中斷時，雲端能立刻得知。
4. **傳輸加密**：
   - **App 端 (公網)**：強制使用 WSS (`wss://`)，TLS 由 NPM / Cloudflare 終止。
   - **Go Gateway (內網)**：預設走 TCP (`tcp://emqx:1883`)，區網環境可免 TLS。正式環境建議升級至 `ssl://emqx:8883`。

---

## 5. Redis 橋接注意事項

### 5.1 Key 前綴一致性
Laravel 的 Redis 驅動會自動在 Key 前加上前綴 (預設 `star_cloud_database_`)。Go Gateway 不使用 Laravel Redis 客戶端，因此**必須在程式碼中手動加上此前綴**：

```go
// main.go
prefixedIncomingKey := "star_cloud_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. 斷線保障機制

App 端的通訊策略為 **MQTT 唯一通道 + 本地暫存補發**，不維護 HTTP Fallback：

| 狀態 | 心跳 | 指令接收 | 交易/發票/出貨 |
| :--- | :--- | :--- | :--- |
| WSS 連線正常 | MQTT 發布 (1min) | MQTT 訂閱 | MQTT 發布 |
| WSS 斷線 | 暫停，等待重連 | 不可用 | 暫存至本地 SQLite 佇列 |
| WSS 恢復 | 恢復 MQTT 發布 | 恢復 MQTT 訂閱 | 自動逐筆補發暫存訊息 |

**本地暫存補發規則：**
- 重連後延遲 2 秒開始補發，每筆間隔 200ms
- 重試上限 10 次，超過標記為 failed
- 暫存超過 72 小時的訊息直接丟棄

> [!CAUTION]
> **已遷移至 MQTT 的端點 (B600/B601/B602/B010/B013) 不保留 HTTP 路由。** 斷線時依賴 App 本地暫存機制保證資料完整性。

---

## 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 (程式碼變更後必須執行) |