refactor: 重构 token 管理，新增 NATS adapter 和多示例

按照 TOKEN_REFACTOR.md / FIX_CONFIG.md 方案重构核心层： - miot: Init() 优先读 storage token，新增 RefreshAuthInfo() 和刷新回调 - xiaomi: Config 改用 AuthInfo 结构体（含 UUID/UID/DataDir/OnTokenRefreshed） - xiaomi: NewClient(ctx, Config) 新签名，新增 RefreshToken()，后台自动刷新 - xiaomi/devices: 设备控制接口重构 - 新增示例 12_ac_subscribe，其余示例适配新 Config
2026-06-30 12:16:49 +08:00
parent a3d94c4b9f
commit 8ffc9e4e44
33 changed files with 1019 additions and 593 deletions
@@ -97,6 +97,10 @@ type MIoTClient struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 	mu     sync.RWMutex
+	refreshMu sync.Mutex // token 刷新互斥锁
+
+	// === Callbacks ===
+	tokenRefreshCB func(accessToken, refreshToken string)
 }

 // NewMIoTClient creates a new MIoTClient.
@@ -129,15 +133,41 @@ func NewMIoTClient(entryID string, entryData map[string]interface{}, uid string,
 // Init initializes the MIoTClient.
 // Aligns with Python MIoTClient.init_async.
 func (c *MIoTClient) Init() error {
-	// 1. Load user config
+	// 1. Load user config from storage, fallback to entryData
+	accessToken, _ := c.entryData["access_token"].(string)
+	refreshToken, _ := c.entryData["refresh_token"].(string)
+
 	if c.storage != nil {
-		config := c.storage.LoadUserConfig(c.uid, c.cloudServer, nil)
-		if len(config) > 0 {
-			fmt.Printf("[MIoTClient] Loaded user config\n")
+		authInfo := c.storage.LoadUserConfig(c.uid, c.cloudServer, nil)
+		if t, ok := authInfo["access_token"].(string); ok && t != "" {
+			accessToken = t
+			// Also restore refresh_token and expires_ts from storage
+			if rt, ok := authInfo["refresh_token"].(string); ok && rt != "" {
+				refreshToken = rt
+				c.entryData["refresh_token"] = rt
+			}
+			if et, ok := authInfo["expires_ts"].(float64); ok {
+				c.entryData["expires_ts"] = et
+			}
+			fmt.Printf("[MIoTClient] Loaded token from storage\n")
+		} else if accessToken != "" {
+			// First run: write entryData token to storage
+			expiresTS := float64(time.Now().Unix()) + 3600*24*30 // default 30 days
+			if et, ok := c.entryData["expires_ts"].(float64); ok && et > 0 {
+				expiresTS = et
+			}
+			c.storage.UpdateUserConfig(c.uid, c.cloudServer, map[string]interface{}{
+				"access_token":  accessToken,
+				"refresh_token": refreshToken,
+				"expires_ts":    expiresTS,
+			}, true)
+			fmt.Printf("[MIoTClient] Saved initial token to storage\n")
 		}
-		_ = config
 	}

+	// Keep access_token in entryData for downstream use
+	c.entryData["access_token"] = accessToken
+
 	// 2. Load device cache
 	if err := c.loadCacheDevice(); err != nil {
 		fmt.Printf("[MIoTClient] Warning: failed to load device cache: %v\n", err)
@@ -161,25 +191,21 @@ func (c *MIoTClient) Init() error {
 		fmt.Printf("[MIoTClient] Cert client initialized\n")
 	}

-	// 4. Initialize HTTP client (from existing code path)
-	if c.http == nil && c.entryData != nil {
-		accessToken, _ := c.entryData["access_token"].(string)
-		if accessToken != "" {
-			httpClient, err := NewMIoTHttpClient(c.cloudServer, OAUTH2_CLIENT_ID, accessToken)
-			if err != nil {
-				fmt.Printf("[MIoTClient] Warning: failed to create HTTP client: %v\n", err)
-			} else {
-				c.http = httpClient
-				fmt.Printf("[MIoTClient] HTTP client created\n")
-			}
+	// 4. Initialize HTTP client (use token from storage if available)
+	if c.http == nil && accessToken != "" {
+		httpClient, err := NewMIoTHttpClient(c.cloudServer, OAUTH2_CLIENT_ID, accessToken)
+		if err != nil {
+			fmt.Printf("[MIoTClient] Warning: failed to create HTTP client: %v\n", err)
+		} else {
+			c.http = httpClient
+			fmt.Printf("[MIoTClient] HTTP client created\n")
 		}
 	}

-	// 3.3 Initialize MipsCloudClient
-	if c.mipsCloud == nil && c.entryData != nil {
-		accessToken, _ := c.entryData["access_token"].(string)
+	// 3.3 Initialize MipsCloudClient (use token from storage if available)
+	if c.mipsCloud == nil && accessToken != "" {
 		uuid, _ := c.entryData["uuid"].(string)
-		if accessToken != "" && uuid != "" {
+		if uuid != "" {
 			broker := buildCloudBroker(c.cloudServer)
 			mips := NewMipsCloudClient(broker, "ha."+uuid, OAUTH2_CLIENT_ID, accessToken)
 			key := c.uid + "-" + c.cloudServer
@@ -553,6 +579,11 @@ func (c *MIoTClient) UnsubDeviceState(did string) error {
 }

 // ============================================================================
+// SetTokenRefreshCallback sets a callback invoked when the background timer refreshes the token.
+func (c *MIoTClient) SetTokenRefreshCallback(cb func(accessToken, refreshToken string)) {
+	c.tokenRefreshCB = cb
+}
+
 // M2: RequestRefreshAuthInfo / RequestRefreshUserCert
 // ============================================================================

@@ -577,6 +608,9 @@ func (c *MIoTClient) RequestRefreshAuthInfo(delaySec int) {
 // refreshAuthInfo performs the actual token refresh logic.
 // Aligns with Python: refresh_oauth_info_async.
 func (c *MIoTClient) refreshAuthInfo() {
+	c.refreshMu.Lock()
+	defer c.refreshMu.Unlock()
+
 	if c.oauth == nil || c.storage == nil {
 		return
 	}
@@ -587,13 +621,13 @@ func (c *MIoTClient) refreshAuthInfo() {
 	refreshToken, _ := authInfo["refresh_token"].(string)
 	expiresTS, _ := authInfo["expires_ts"].(float64)

-	// Check if refresh is needed: expires_ts - now <= MIHOME_CERT_EXPIRE_MARGIN
+	// Check if refresh is needed: expires_ts - now <= MIHOME_REFRESH_THRESHOLD
 	now := float64(time.Now().UnixMilli()) / 1000
-	needRefresh := (expiresTS - now) <= float64(MIHOME_CERT_EXPIRE_MARGIN)
+	needRefresh := (expiresTS - now) <= float64(MIHOME_REFRESH_THRESHOLD)

 	if !needRefresh && accessToken != "" {
 		// Schedule next check
-		nextDelay := int(expiresTS - now - float64(MIHOME_CERT_EXPIRE_MARGIN))
+		nextDelay := int(expiresTS - now - float64(MIHOME_REFRESH_THRESHOLD))
 		if nextDelay < 60 {
 			nextDelay = 60
 		}
@@ -618,7 +652,11 @@ func (c *MIoTClient) refreshAuthInfo() {

 	newAccessToken, _ := result["access_token"].(string)
 	newRefreshToken, _ := result["refresh_token"].(string)
-	newExpiresTS, _ := result["expires_ts"].(float64)
+	var newExpiresTS float64
+	switch v := result["expires_ts"].(type) {
+	case float64: newExpiresTS = v
+	case int64:    newExpiresTS = float64(v)
+	}

 	if newAccessToken == "" {
 		fmt.Printf("[MIoTClient] Token refresh returned empty access_token\n")
@@ -645,14 +683,82 @@ func (c *MIoTClient) refreshAuthInfo() {

 	fmt.Printf("[MIoTClient] Token refreshed successfully\n")

+	// Notify callback (xiaomi layer uses this to persist to DB)
+	if c.tokenRefreshCB != nil {
+		c.tokenRefreshCB(newAccessToken, newRefreshToken)
+	}
+
 	// Schedule next refresh
-	nextDelay := int(newExpiresTS - now - float64(MIHOME_CERT_EXPIRE_MARGIN))
+	nextDelay := int(newExpiresTS - now - float64(MIHOME_REFRESH_THRESHOLD))
 	if nextDelay < 60 {
 		nextDelay = 60
 	}
 	c.RequestRefreshAuthInfo(nextDelay)
 }

+// RefreshAuthInfo 手动触发 token 刷新并同步返回结果（对齐 Python refresh_oauth_info_async）。
+// 返回 map 包含 access_token / refresh_token / expires_ts。内部有互斥锁，可安全并发调用。
+func (c *MIoTClient) RefreshAuthInfo() (map[string]interface{}, error) {
+	c.refreshMu.Lock()
+	defer c.refreshMu.Unlock()
+	if c.oauth == nil || c.storage == nil {
+		return nil, fmt.Errorf("miot: oauth or storage not initialized")
+	}
+
+	authInfo := c.storage.LoadUserConfig(c.uid, c.cloudServer, nil)
+	refreshToken, _ := authInfo["refresh_token"].(string)
+	accessToken, _ := authInfo["access_token"].(string)
+	expiresTS, _ := authInfo["expires_ts"].(float64)
+
+	// 未临近过期且 token 有效 → 直接返回，避免重复刷新浪费 refresh_token
+	now := float64(time.Now().UnixMilli()) / 1000
+	if accessToken != "" && (expiresTS-now) > float64(MIHOME_REFRESH_THRESHOLD) {
+		return map[string]interface{}{
+			"access_token":  accessToken,
+			"refresh_token": refreshToken,
+			"expires_ts":    expiresTS,
+		}, nil
+	}
+
+	if refreshToken == "" {
+		refreshToken, _ = c.entryData["refresh_token"].(string)
+	}
+	if refreshToken == "" {
+		return nil, fmt.Errorf("miot: no refresh token available")
+	}
+
+	result, err := c.oauth.RefreshAccessToken(refreshToken)
+	if err != nil {
+		return nil, fmt.Errorf("miot: refresh access token: %w", err)
+	}
+
+	// Save new auth info
+	c.entryData["access_token"] = result["access_token"]
+	c.entryData["refresh_token"] = result["refresh_token"]
+	if et, ok := result["expires_ts"]; ok {
+		c.entryData["expires_ts"] = et
+	}
+	c.storage.UpdateUserConfig(c.uid, c.cloudServer, map[string]interface{}{
+		"access_token":  result["access_token"],
+		"refresh_token": result["refresh_token"],
+		"expires_ts":    result["expires_ts"],
+	}, true)
+
+	// Update live clients
+	if c.http != nil {
+		if ac, ok := result["access_token"].(string); ok {
+			c.http.UpdateHTTPHeader(c.cloudServer, OAUTH2_CLIENT_ID, ac)
+		}
+	}
+	if c.mipsCloud != nil {
+		ac, _ := result["access_token"].(string)
+		c.mipsCloud.UpdateAccessToken(ac)
+	}
+
+	fmt.Printf("[MIoTClient] RefreshAuthInfo: token refreshed\n")
+	return result, nil
+}
+
 // RequestRefreshUserCert schedules a user certificate refresh after delaySec seconds.
 // Uses MIHOME_CERT_EXPIRE_MARGIN to determine when refresh is needed.
 // Aligns with Python: __request_refresh_user_cert(delay_sec).