// internal/models/middleware.go

type MiddlewareType string

const (
    MiddlewareMQTT MiddlewareType = "edgeOS(MQTT)"
    MiddlewareNATS MiddlewareType = "edgeOS(NATS)"
)

// MiddlewareConfig 前端提交的连接配置
type MiddlewareConfig struct {
    ID          string         `json:"id"`           // 唯一ID（后端生成）
    Name        string         `json:"name"`         // 连接名称，如 "生产环境MQTT"
    Type        MiddlewareType `json:"type"`         // edgeOS(MQTT) 或 edgeOS(NATS)
    Description string         `json:"description"`  // 备注
    MQTT        *MQTTConfig    `json:"mqtt,omitempty"`
    NATS        *NATSConfig    `json:"nats,omitempty"`
    Status      ConnStatus     `json:"status"`       // connected / disconnected / error
    CreatedAt   int64          `json:"created_at"`
    UpdatedAt   int64          `json:"updated_at"`
}

type MQTTConfig struct {
    Broker             string `json:"broker"`               // tcp://127.0.0.1:1883
    ClientID           string `json:"client_id"`
    Username           string `json:"username"`
    Password           string `json:"password"`
    QoS                byte   `json:"qos"`                  // 0/1/2
    CleanSession       bool   `json:"clean_session"`
    KeepAlive          int    `json:"keep_alive"`           // 秒
    ConnectTimeout     int    `json:"connect_timeout"`      // 秒
    AutoReconnect      bool   `json:"auto_reconnect"`
    MaxReconnectInterval int  `json:"max_reconnect_interval"` // 秒
    TLSEnabled         bool   `json:"tls_enabled"`
    CACert             string `json:"ca_cert,omitempty"`
    ClientCert         string `json:"client_cert,omitempty"`
    ClientKey          string `json:"client_key,omitempty"`
}

type NATSConfig struct {
    URL                string `json:"url"`                  // nats://127.0.0.1:4222
    ClientName         string `json:"client_name"`
    Username           string `json:"username"`
    Password           string `json:"password"`
    Token              string `json:"token,omitempty"`
    ConnectTimeout     int    `json:"connect_timeout"`
    ReconnectWait      int    `json:"reconnect_wait"`
    MaxReconnects      int    `json:"max_reconnects"`
    JetStreamEnabled   bool   `json:"jetstream_enabled"`
    TLSEnabled         bool   `json:"tls_enabled"`
}

type ConnStatus string

const (
    ConnStatusConnected    ConnStatus = "connected"
    ConnStatusDisconnected ConnStatus = "disconnected"
    ConnStatusError        ConnStatus = "error"
    ConnStatusConnecting   ConnStatus = "connecting"
)

// internal/middleware/manager.go

package middleware

import (
    "fmt"
    "sync"
    "time"

    "github.com/google/uuid"
    "github.com/sirupsen/logrus"
)

// ConnectionManager 管理所有消息总线
type ConnectionManager struct {
    mu          sync.RWMutex
    connections map[string]*Connection  // id -> Connection
    storage     MiddlewareStorage
    router      MessageRouter
    log         *logrus.Logger
}

// Connection 单个消息总线实例
type Connection struct {
    Config  *MiddlewareConfig
    Client  MiddlewareClient  // 接口，MQTT或NATS实现
    subs    []Subscription    // 已订阅列表
}

// MiddlewareClient 统一客户端接口
type MiddlewareClient interface {
    Connect() error
    Disconnect()
    Subscribe(topic string, handler MessageHandler) error
    Publish(topic string, msgType string, body interface{}) error
    IsConnected() bool
    Status() ConnStatus
}

// MessageHandler 消息处理函数
type MessageHandler func(topic string, msg *Message) error

func NewConnectionManager(storage MiddlewareStorage, router MessageRouter, log *logrus.Logger) *ConnectionManager {
    return &ConnectionManager{
        connections: make(map[string]*Connection),
        storage:     storage,
        router:      router,
        log:         log,
    }
}

// AddConnection 添加并连接中间件
func (m *ConnectionManager) AddConnection(cfg *MiddlewareConfig) (*MiddlewareConfig, error) {
    cfg.ID = uuid.NewString()
    cfg.Status = ConnStatusConnecting
    cfg.CreatedAt = time.Now().UnixMilli()
    cfg.UpdatedAt = cfg.CreatedAt

    var client MiddlewareClient
    var err error

    switch cfg.Type {
    case MiddlewareMQTT:
        client, err = NewMQTTClient(cfg.MQTT, m.log)
    case MiddlewareNATS:
        client, err = NewNATSClient(cfg.NATS, m.log)
    default:
        return nil, fmt.Errorf("unsupported middleware type: %s", cfg.Type)
    }
    if err != nil {
        cfg.Status = ConnStatusError
        return cfg, err
    }

    if err = client.Connect(); err != nil {
        cfg.Status = ConnStatusError
        _ = m.storage.Save(cfg)
        return cfg, fmt.Errorf("connect failed: %w", err)
    }

    cfg.Status = ConnStatusConnected
    conn := &Connection{Config: cfg, Client: client}

    // 连接成功后订阅全部功能主题
    if err = m.subscribeAll(conn); err != nil {
        m.log.Warnf("partial subscribe error: %v", err)
    }

    m.mu.Lock()
    m.connections[cfg.ID] = conn
    m.mu.Unlock()

    _ = m.storage.Save(cfg)
    return cfg, nil
}

// subscribeAll 订阅四个核心功能所需的所有主题
func (m *ConnectionManager) subscribeAll(conn *Connection) error {
    topics := buildTopicList(conn.Config.Type)
    for _, t := range topics {
        if err := conn.Client.Subscribe(t, m.router.Route); err != nil {
            return fmt.Errorf("subscribe %s failed: %w", t, err)
        }
        conn.subs = append(conn.subs, Subscription{Topic: t})
        m.log.Infof("[%s] subscribed: %s", conn.Config.Name, t)
    }
    return nil
}

// buildTopicList 根据协议类型返回需要订阅的主题列表
func buildTopicList(t MiddlewareType) []string {
    if t == MiddlewareMQTT {
        return []string{
            // 1. 节点注册（Stage 1：被动）
            "edgex/nodes/register",
            "edgex/nodes/unregister",
            // 1b. 节点发现（Stage 2：主动触发 EdgeX 重新注册）
            "edgex/cmd/nodes/register",
            // 2. 设备同步
            "edgex/devices/report",
            // 3. 点位同步(即物模型点位上报 可用将实时数据点位理解成物模型)
            "edgex/points/report",
            // 4. 实时数据（双向控制读侧）
            "edgex/data/#",
            // 响应
            "edgex/responses/#",
            // 心跳与状态
            "edgex/nodes/+/heartbeat",
            "edgex/nodes/+/status",
            "edgex/nodes/+/online",
            "edgex/nodes/+/offline",
            // 告警
            "edgex/events/alert",
            "edgex/events/error",
            "edgex/events/info",
        }
    }
    // NATS
    return []string{
        "edgex.nodes.register",
        "edgex.nodes.unregister",
        "edgex.cmd.nodes.register",
        "edgex.devices.report",
        "edgex.points.report",
        "edgex.data.>",
        "edgex.res.>",
        "edgex.nodes.heartbeat.>",
        "edgex.nodes.status.>",
        "edgex.events.alert",
        "edgex.events.error",
        "edgex.events.info",
    }
}

// internal/middleware/mqtt_client.go

package middleware

import (
    "encoding/json"
    "fmt"
    "time"

    mqtt "github.com/eclipse/paho.mqtt.golang"
    "github.com/sirupsen/logrus"
)

type mqttClient struct {
    cfg    *MQTTConfig
    client mqtt.Client
    log    *logrus.Logger
}

func NewMQTTClient(cfg *MQTTConfig, log *logrus.Logger) (MiddlewareClient, error) {
    opts := mqtt.NewClientOptions()
    opts.AddBroker(cfg.Broker)
    opts.SetClientID(cfg.ClientID)
    opts.SetUsername(cfg.Username)
    opts.SetPassword(cfg.Password)
    opts.SetCleanSession(cfg.CleanSession)
    opts.SetKeepAlive(time.Duration(cfg.KeepAlive) * time.Second)
    opts.SetConnectTimeout(time.Duration(cfg.ConnectTimeout) * time.Second)
    opts.SetAutoReconnect(cfg.AutoReconnect)

    opts.SetOnConnectHandler(func(c mqtt.Client) {
        log.Info("[MQTT] connected")
    })
    opts.SetConnectionLostHandler(func(c mqtt.Client, err error) {
        log.Warnf("[MQTT] connection lost: %v", err)
    })

    return &mqttClient{cfg: cfg, client: mqtt.NewClient(opts), log: log}, nil
}

func (c *mqttClient) Connect() error {
    if token := c.client.Connect(); token.Wait() && token.Error() != nil {
        return token.Error()
    }
    return nil
}

func (c *mqttClient) Disconnect() { c.client.Disconnect(250) }

func (c *mqttClient) IsConnected() bool { return c.client.IsConnected() }

func (c *mqttClient) Status() ConnStatus {
    if c.client.IsConnected() {
        return ConnStatusConnected
    }
    return ConnStatusDisconnected
}

func (c *mqttClient) Subscribe(topic string, handler MessageHandler) error {
    token := c.client.Subscribe(topic, c.cfg.QoS, func(_ mqtt.Client, msg mqtt.Message) {
        var m Message
        if err := json.Unmarshal(msg.Payload(), &m); err != nil {
            c.log.Errorf("[MQTT] unmarshal error on %s: %v", msg.Topic(), err)
            return
        }
        if err := handler(msg.Topic(), &m); err != nil {
            c.log.Errorf("[MQTT] handler error on %s: %v", msg.Topic(), err)
        }
    })
    token.Wait()
    return token.Error()
}

func (c *mqttClient) Publish(topic string, msgType string, body interface{}) error {
    msg := buildMessage(c.cfg.ClientID, msgType, body)
    payload, err := json.Marshal(msg)
    if err != nil {
        return err
    }
    token := c.client.Publish(topic, c.cfg.QoS, false, payload)
    token.Wait()
    return token.Error()
}

// internal/middleware/nats_client.go

package middleware

import (
    "encoding/json"
    "fmt"
    "time"

    "github.com/nats-io/nats.go"
    "github.com/sirupsen/logrus"
)

type natsClient struct {
    cfg *NATSConfig
    nc  *nats.Conn
    js  nats.JetStreamContext
    log *logrus.Logger
}

func NewNATSClient(cfg *NATSConfig, log *logrus.Logger) (MiddlewareClient, error) {
    return &natsClient{cfg: cfg, log: log}, nil
}

func (c *natsClient) Connect() error {
    opts := []nats.Option{
        nats.Name(c.cfg.ClientName),
        nats.UserInfo(c.cfg.Username, c.cfg.Password),
        nats.ReconnectWait(time.Duration(c.cfg.ReconnectWait) * time.Second),
        nats.MaxReconnects(c.cfg.MaxReconnects),
        nats.DisconnectErrHandler(func(_ *nats.Conn, err error) {
            c.log.Warnf("[NATS] disconnected: %v", err)
        }),
        nats.ReconnectHandler(func(_ *nats.Conn) {
            c.log.Info("[NATS] reconnected")
        }),
    }
    if c.cfg.Token != "" {
        opts = append(opts, nats.Token(c.cfg.Token))
    }

    nc, err := nats.Connect(c.cfg.URL, opts...)
    if err != nil {
        return err
    }
    c.nc = nc

    if c.cfg.JetStreamEnabled {
        if c.js, err = nc.JetStream(); err != nil {
            c.log.Warnf("[NATS] jetstream unavailable: %v", err)
        }
    }
    return nil
}

func (c *natsClient) Disconnect() {
    if c.nc != nil {
        c.nc.Drain()
    }
}

func (c *natsClient) IsConnected() bool { return c.nc != nil && c.nc.IsConnected() }

func (c *natsClient) Status() ConnStatus {
    if c.IsConnected() {
        return ConnStatusConnected
    }
    return ConnStatusDisconnected
}

func (c *natsClient) Subscribe(subject string, handler MessageHandler) error {
    _, err := c.nc.Subscribe(subject, func(msg *nats.Msg) {
        var m Message
        if err := json.Unmarshal(msg.Data, &m); err != nil {
            c.log.Errorf("[NATS] unmarshal error on %s: %v", msg.Subject, err)
            return
        }
        if err := handler(msg.Subject, &m); err != nil {
            c.log.Errorf("[NATS] handler error on %s: %v", msg.Subject, err)
        }
    })
    return err
}

func (c *natsClient) Publish(subject string, msgType string, body interface{}) error {
    msg := buildMessage(c.cfg.ClientName, msgType, body)
    payload, err := json.Marshal(msg)
    if err != nil {
        return err
    }
    return c.nc.Publish(subject, payload)
}

// Request 发起请求/响应（NATS 特有）
func (c *natsClient) Request(subject string, msgType string, body interface{}, timeout time.Duration) (*Message, error) {
    msg := buildMessage(c.cfg.ClientName, msgType, body)
    payload, err := json.Marshal(msg)
    if err != nil {
        return nil, err
    }
    resp, err := c.nc.Request(subject, payload, timeout)
    if err != nil {
        return nil, err
    }
    var m Message
    if err = json.Unmarshal(resp.Data, &m); err != nil {
        return nil, err
    }
    return &m, nil
}

// internal/handlers/node_handler.go

package handlers

import (
    "fmt"
    "time"

    "github.com/sirupsen/logrus"
)

type NodeHandler struct {
    nodeService NodeService
    publisher   Publisher
    log         *logrus.Logger
}

func (h *NodeHandler) HandleRegister(topic string, msg *Message) error {
    var body NodeRegisterBody
    if err := decodeBody(msg.Body, &body); err != nil {
        return fmt.Errorf("decode node_register body: %w", err)
    }

    // 1. 幂等检查：已注册则直接返回成功
    existing, _ := h.nodeService.GetByID(body.NodeID)
    if existing != nil {
        h.log.Infof("[NodeReg] node %s already registered, update info", body.NodeID)
        existing.Status = "online"
        existing.UpdatedAt = time.Now().UnixMilli()
        _ = h.nodeService.Update(existing)
        return h.replyRegisterSuccess(msg.Header, body.NodeID)
    }

    // 2. 持久化节点
    node := &Node{
        ID:           body.NodeID,
        Name:         body.NodeName,
        Model:        body.Model,
        Version:      body.Version,
        Capabilities: body.Capabilities,
        Protocol:     body.Protocol,
        Endpoint:     body.Endpoint,
        Metadata:     body.Metadata,
        Status:       "online",
        RegisteredAt: time.Now().UnixMilli(),
        UpdatedAt:    time.Now().UnixMilli(),
    }
    if err := h.nodeService.Save(node); err != nil {
        return fmt.Errorf("save node: %w", err)
    }
    h.log.Infof("[NodeReg] registered node: %s (%s)", body.NodeID, body.NodeName)

    // 3. 回复注册响应
    return h.replyRegisterSuccess(msg.Header, body.NodeID)
}

func (h *NodeHandler) replyRegisterSuccess(reqHeader MessageHeader, nodeID string) error {
    respBody := map[string]interface{}{
        "success":      true,
        "node_id":      nodeID,
        "access_token": generateToken(nodeID),
        "expires_in":   3600,
        "message":      "Node registered successfully",
    }
    respHeader := MessageHeader{
        MessageID:     generateUUID(),
        Timestamp:     time.Now().UnixMilli(),
        Source:        "edgeos-queen",
        Destination:   reqHeader.Source,
        MessageType:   "node_register_response",
        Version:       "1.0",
        CorrelationID: reqHeader.MessageID,
    }
    topic := fmt.Sprintf("edgex/responses/%s/%s", reqHeader.Source, reqHeader.MessageID)
    return h.publisher.Publish(topic, respHeader, respBody)
}

func (h *NodeHandler) HandleUnregister(topic string, msg *Message) error {
    var body NodeUnregisterBody
    if err := decodeBody(msg.Body, &body); err != nil {
        return fmt.Errorf("decode node_unregister body: %w", err)
    }
    return h.nodeService.MarkOffline(body.NodeID)
}

// internal/models/node.go

type Node struct {
    ID           string            `json:"id"`
    Name         string            `json:"name"`
    Model        string            `json:"model"`
    Version      string            `json:"version"`
    APIVersion   string            `json:"api_version"`
    Capabilities []string          `json:"capabilities"`
    Protocol     string            `json:"protocol"`
    Endpoint     NodeEndpoint      `json:"endpoint"`
    Metadata     map[string]string `json:"metadata"`
    Status       string            `json:"status"`  // online / offline / error
    RegisteredAt int64             `json:"registered_at"`
    UpdatedAt    int64             `json:"updated_at"`
    LastHeartbeat int64            `json:"last_heartbeat"`
}

type NodeEndpoint struct {
    Host string `json:"host"`
    Port int    `json:"port"`
}

// PublishNodeDiscovery 主动向第一个已连接中间件发布节点发现请求
func (m *Manager) PublishNodeDiscovery() error {
    msg := map[string]interface{}{
        "header": map[string]interface{}{
            "message_type": "discovery_request",
            "source":       "edgeos",
            "timestamp":    time.Now().UnixMilli(),
        },
        "body": map[string]interface{}{},
    }
    payload, _ := json.Marshal(msg)
    return m.publishToFirstClient("edgex/cmd/nodes/register", payload)
}

// PublishNodeDiscoveryTo 向指定中间件发布节点发现请求
func (m *Manager) PublishNodeDiscoveryTo(middlewareID string) error {
    m.mu.RLock()
    defer m.mu.RUnlock()
    entry, ok := m.clients[middlewareID]
    if !ok || entry.client == nil || !entry.client.IsConnected() {
        return fmt.Errorf("middleware %s not connected", middlewareID)
    }
    msg := map[string]interface{}{
        "header": map[string]interface{}{
            "message_type": "discovery_request",
            "source":       "edgeos",
            "timestamp":    time.Now().UnixMilli(),
        },
        "body": map[string]interface{}{},
    }
    payload, _ := json.Marshal(msg)
    return entry.publishFn("edgex/cmd/nodes/register", payload)
}

// NATS 客户端 PublishNodeDiscoveryTo
func (c *natsClient) PublishDiscoveryRequest() error {
    msg := map[string]interface{}{
        "header": map[string]interface{}{
            "message_type": "discovery_request",
            "source":       "edgeos",
            "timestamp":    time.Now().UnixMilli(),
        },
        "body": map[string]interface{}{},
    }
    payload, _ := json.Marshal(msg)
    return c.nc.Publish("edgex.cmd.nodes.register", payload)
}

// internal/handlers/device_handler.go

package handlers

type DeviceHandler struct {
    nodeService   NodeService
    deviceService DeviceService
    log           *logrus.Logger
}

func (h *DeviceHandler) HandleDeviceReport(topic string, msg *Message) error {
    var body DeviceReportBody
    if err := decodeBody(msg.Body, &body); err != nil {
        return fmt.Errorf("decode device_report: %w", err)
    }

    // 校验节点是否已注册
    node, err := h.nodeService.GetByID(body.NodeID)
    if err != nil || node == nil {
        return fmt.Errorf("node %s not registered", body.NodeID)
    }

    reportedIDs := make(map[string]struct{})
    for _, d := range body.Devices {
        reportedIDs[d.DeviceID] = struct{}{}

        existing, _ := h.deviceService.GetByID(body.NodeID, d.DeviceID)
        device := &Device{
            NodeID:         body.NodeID,
            DeviceID:       d.DeviceID,
            DeviceName:     d.DeviceName,
            DeviceProfile:  d.DeviceProfile,
            ServiceName:    d.ServiceName,
            Labels:         d.Labels,
            Description:    d.Description,
            AdminState:     d.AdminState,
            OperatingState: d.OperatingState,
            Properties:     d.Properties,
            UpdatedAt:      time.Now().UnixMilli(),
        }
        if existing == nil {
            device.CreatedAt = device.UpdatedAt
            if err = h.deviceService.Save(device); err != nil {
                h.log.Errorf("[DeviceSync] save device %s: %v", d.DeviceID, err)
                continue
            }
            h.log.Infof("[DeviceSync] new device: %s / %s", body.NodeID, d.DeviceID)
        } else {
            if err = h.deviceService.Update(device); err != nil {
                h.log.Errorf("[DeviceSync] update device %s: %v", d.DeviceID, err)
            }
        }
    }

    // 标记不在上报列表中的设备为离线
    allDevices, _ := h.deviceService.ListByNode(body.NodeID)
    for _, d := range allDevices {
        if _, ok := reportedIDs[d.DeviceID]; !ok {
            _ = h.deviceService.MarkOffline(body.NodeID, d.DeviceID)
        }
    }
    return nil
}

// TriggerDiscover 主动触发设备发现
func (h *DeviceHandler) TriggerDiscover(nodeID string, protocol string, publisher Publisher) error {
    body := map[string]interface{}{
        "protocol": protocol,
        "options": map[string]interface{}{
            "auto_register":    true,
            "sync_immediately": true,
        },
    }
    topic := fmt.Sprintf("edgex/cmd/%s/discover", nodeID)
    return publisher.Publish(topic, "discover_command", body)
}

// internal/models/device.go

type Device struct {
    NodeID         string                 `json:"node_id"`
    DeviceID       string                 `json:"device_id"`
    DeviceName     string                 `json:"device_name"`
    DeviceProfile  string                 `json:"device_profile"`
    ServiceName    string                 `json:"service_name"`
    Labels         []string               `json:"labels"`
    Description    string                 `json:"description"`
    AdminState     string                 `json:"admin_state"`
    OperatingState string                 `json:"operating_state"`
    Properties     map[string]interface{} `json:"properties"`
    Status         string                 `json:"status"`  // online / offline
    CreatedAt      int64                  `json:"created_at"`
    UpdatedAt      int64                  `json:"updated_at"`
}

// internal/handlers/point_handler.go

package handlers

import (
    "fmt"
    "strings"
    "time"

    "github.com/sirupsen/logrus"
)

type PointHandler struct {
    nodeService   NodeService
    deviceService DeviceService
    pointService  PointService
    broadcast     BroadcastFunc  // 向前端 WebSocket 推送
    log           *logrus.Logger
}

// HandlePointReport 处理点位元数据上报（edgex/points/report，可选）
// 保存点位的配置定义：类型、单位、地址、范围等，不覆盖 CurrentValue/Quality/LastUpdated。
func (h *PointHandler) HandlePointReport(topic string, msg *Message) error {
    var body PointReportBody
    if err := decodeBody(msg.Body, &body); err != nil {
        return fmt.Errorf("decode point_report: %w", err)
    }

    dev, err := h.deviceService.GetByID(body.NodeID, body.DeviceID)
    if err != nil || dev == nil {
        return fmt.Errorf("device %s/%s not found", body.NodeID, body.DeviceID)
    }

    now := time.Now().UnixMilli()
    for _, p := range body.Points {
        point := &Point{
            NodeID:       body.NodeID,
            DeviceID:     body.DeviceID,
            PointID:      p.PointID,
            PointName:    p.PointName,
            ResourceName: p.ResourceName,
            ValueType:    p.ValueType,
            AccessMode:   p.AccessMode,
            Unit:         p.Unit,
            Minimum:      p.Minimum,
            Maximum:      p.Maximum,
            Address:      p.Address,
            DataType:     p.DataType,
            Scale:        p.Scale,
            Offset:       p.Offset,
            UpdatedAt:    now,
        }
        existing, _ := h.pointService.GetByID(body.NodeID, body.DeviceID, p.PointID)
        if existing == nil {
            point.CreatedAt = now
            _ = h.pointService.Save(point)
        } else {
            // 只更新元数据字段，不覆盖 CurrentValue / Quality / LastUpdated
            _ = h.pointService.UpdateMeta(point)
        }
    }
    h.log.Infof("[PointMeta] synced %d point definitions for %s/%s",
        len(body.Points), body.NodeID, body.DeviceID)
    return nil
}

// HandleRealtimeData 处理实时数据上报（edgex/data/{node_id}/{device_id}）
//
// 两阶段 Merge 策略：
//   - 首次上报（设备无缓存）：body.points 为物模型全量数据，全量写入缓存，
//     同时为未经 point_report 注册的点位自动创建占位记录（point_id 作为 point_name）。
//   - 后续上报（设备已有缓存）：body.points 仅含变化点位，执行 Merge，
//     未出现在本次 payload 中的点位保持原值不变。
func (h *PointHandler) HandleRealtimeData(topic string, msg *Message) error {
    var body DataBody
    if err := decodeBody(msg.Body, &body); err != nil {
        return fmt.Errorf("decode data: %w", err)
    }

    // 兜底：从 topic 解析 nodeId / deviceId
    nodeID, deviceID := body.NodeID, body.DeviceID
    if nodeID == "" || deviceID == "" {
        parts := strings.Split(topic, "/")
        if len(parts) >= 4 {
            nodeID, deviceID = parts[2], parts[3]
        }
    }

    isFirstReport := !h.pointService.HasCache(nodeID, deviceID)

    now := time.Now().UnixMilli()
    for pointID, value := range body.Points {
        existing, _ := h.pointService.GetByID(nodeID, deviceID, pointID)
        if existing == nil {
            // 自动创建点位占位（元数据待 point_report 补全）
            placeholder := &Point{
                NodeID:       nodeID,
                DeviceID:     deviceID,
                PointID:      pointID,
                PointName:    pointID, // 暂用 point_id 作为名称
                CurrentValue: value,
                Quality:      body.Quality,
                LastUpdated:  body.Timestamp,
                CreatedAt:    now,
                UpdatedAt:    now,
            }
            _ = h.pointService.Save(placeholder)
        } else {
            // Merge：只更新当前值，保留元数据不变
            _ = h.pointService.UpdateValue(nodeID, deviceID, pointID, value, body.Quality, body.Timestamp)
        }
    }

    if isFirstReport {
        h.log.Infof("[PointSync] full snapshot: %d points for %s/%s",
            len(body.Points), nodeID, deviceID)
    } else {
        h.log.Debugf("[PointSync] delta merge: %d changed points for %s/%s",
            len(body.Points), nodeID, deviceID)
    }

    // 广播实时事件到前端
    // payload 只含本次上报的点位（差量），前端 store 同步执行 merge
    h.broadcast(RealtimeDataEvent{
        NodeID:         nodeID,
        DeviceID:       deviceID,
        Points:         body.Points,
        Timestamp:      body.Timestamp,
        Quality:        body.Quality,
        IsFullSnapshot: isFirstReport,
    })
    return nil
}

// internal/services/point_service.go

type PointService interface {
    // HasCache 检查设备是否已有点位缓存（用于判断全量 vs 差量 Merge）
    HasCache(nodeID, deviceID string) bool

    GetByID(nodeID, deviceID, pointID string) (*Point, error)
    Save(p *Point) error

    // UpdateMeta 仅更新点位元数据字段，不覆盖 CurrentValue/Quality/LastUpdated
    UpdateMeta(p *Point) error

    // UpdateValue 差量 Merge 的最小操作单元：更新当前值+质量+时间戳
    UpdateValue(nodeID, deviceID, pointID string, value interface{}, quality string, ts int64) error

    ListByDevice(nodeID, deviceID string) ([]*Point, error)
}

// internal/models/point.go

type Point struct {
    NodeID       string      `json:"node_id"`
    DeviceID     string      `json:"device_id"`
    PointID      string      `json:"point_id"`
    PointName    string      `json:"point_name"`         // 来自 point_report，缺省时等于 point_id
    ResourceName string      `json:"resource_name"`
    ValueType    string      `json:"value_type"`         // Float32 / Int16 / Bool / ...（元数据）
    AccessMode   string      `json:"access_mode"`        // R / W / RW（元数据，缺省时为空）
    Unit         string      `json:"unit"`
    Minimum      interface{} `json:"minimum,omitempty"`
    Maximum      interface{} `json:"maximum,omitempty"`
    Address      string      `json:"address"`
    DataType     string      `json:"data_type"`
    Scale        float64     `json:"scale"`
    Offset       float64     `json:"offset"`
    CurrentValue interface{} `json:"current_value"`      // 最新采集值（由实时数据驱动）
    Quality      string      `json:"quality"`            // Good / Bad / Uncertain
    LastUpdated  int64       `json:"last_updated"`       // 最新值对应的设备时间戳
    CreatedAt    int64       `json:"created_at"`
    UpdatedAt    int64       `json:"updated_at"`         // 元数据最后修改时间
}

// RealtimeDataEvent 向前端广播的实时数据事件
type RealtimeDataEvent struct {
    NodeID         string                 `json:"node_id"`
    DeviceID       string                 `json:"device_id"`
    Points         map[string]interface{} `json:"points"`           // 本次上报的点位（全量或差量）
    Timestamp      int64                  `json:"timestamp"`
    Quality        string                 `json:"quality"`
    IsFullSnapshot bool                   `json:"is_full_snapshot"` // true=全量，前端可用于初始化展示
}

// internal/services/control_service.go

package services

import (
    "fmt"
    "sync"
    "time"
)

type ControlService struct {
    publisher   Publisher
    pendingReqs sync.Map  // requestID -> chan *CommandResponse
    log         *logrus.Logger
}

// WritePoints 向 EdgeX 写入设备点位（异步等待响应）
func (s *ControlService) WritePoints(nodeID, deviceID string, points map[string]interface{}, timeout time.Duration) (*CommandResponse, error) {
    reqID := generateUUID()
    respCh := make(chan *CommandResponse, 1)
    s.pendingReqs.Store(reqID, respCh)
    defer s.pendingReqs.Delete(reqID)

    body := map[string]interface{}{
        "request_id": reqID,
        "device_id":  deviceID,
        "timestamp":  time.Now().UnixMilli(),
        "points":     points,
        "options": map[string]interface{}{
            "confirm":         true,
            "timeout_seconds": int(timeout.Seconds()),
        },
    }
    topic := fmt.Sprintf("edgex/cmd/%s/%s/write", nodeID, deviceID)
    if err := s.publisher.Publish(topic, "write_command", body); err != nil {
        return nil, fmt.Errorf("publish write_command: %w", err)
    }

    select {
    case resp := <-respCh:
        return resp, nil
    case <-time.After(timeout):
        return nil, fmt.Errorf("write command timeout after %v", timeout)
    }
}

// HandleCommandResponse 处理来自 EdgeX 的命令响应
func (s *ControlService) HandleCommandResponse(topic string, msg *Message) error {
    var body CommandResponseBody
    if err := decodeBody(msg.Body, &body); err != nil {
        return err
    }
    if ch, ok := s.pendingReqs.Load(body.RequestID); ok {
        ch.(chan *CommandResponse) <- &CommandResponse{
            RequestID: body.RequestID,
            Success:   body.Success,
            Message:   body.Message,
            Data:      body.Data,
        }
    }
    return nil
}

// CreateTask 创建采集任务
func (s *ControlService) CreateTask(nodeID string, task *TaskCreateRequest) error {
    body := map[string]interface{}{
        "task_id":   task.TaskID,
        "task_name": task.TaskName,
        "device_id": task.DeviceID,
        "schedule":  task.Schedule,
        "points":    task.Points,
        "options":   task.Options,
    }
    topic := fmt.Sprintf("edgex/cmd/%s/task/create", nodeID)
    return s.publisher.Publish(topic, "task_create", body)
}

// ControlTask 任务控制（pause/resume/stop）
func (s *ControlService) ControlTask(nodeID, taskID, action string) error {
    body := map[string]interface{}{
        "task_id": taskID,
        "action":  action,
    }
    topic := fmt.Sprintf("edgex/cmd/%s/task/%s/%s", nodeID, taskID, action)
    return s.publisher.Publish(topic, "task_control", body)
}

// internal/services/heartbeat_service.go

const HeartbeatTimeout = 3 * time.Minute

type HeartbeatService struct {
    nodeService NodeService
    lastSeen    sync.Map  // nodeID -> int64(UnixMilli)
    log         *logrus.Logger
}

func (s *HeartbeatService) HandleHeartbeat(_ string, msg *Message) error {
    var body HeartbeatBody
    if err := decodeBody(msg.Body, &body); err != nil {
        return err
    }
    s.lastSeen.Store(body.NodeID, time.Now().UnixMilli())
    return s.nodeService.UpdateHeartbeat(body.NodeID, body.Status, body.Metrics)
}

// StartWatchdog 定时检查心跳超时
func (s *HeartbeatService) StartWatchdog(ctx context.Context) {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            s.checkTimeouts()
        }
    }
}

func (s *HeartbeatService) checkTimeouts() {
    now := time.Now().UnixMilli()
    s.lastSeen.Range(func(key, value interface{}) bool {
        nodeID := key.(string)
        lastTS := value.(int64)
        if now-lastTS > HeartbeatTimeout.Milliseconds() {
            s.log.Warnf("[Heartbeat] node %s timeout, marking offline", nodeID)
            _ = s.nodeService.MarkOffline(nodeID)
        }
        return true
    })
}

// internal/handlers/event_handler.go

func (h *EventHandler) HandleAlert(_ string, msg *Message) error {
    var body AlertBody
    if err := decodeBody(msg.Body, &body); err != nil {
        return err
    }
    alert := &Alert{
        AlertID:   body.AlertID,
        NodeID:    body.NodeID,
        DeviceID:  body.DeviceID,
        Type:      body.AlertType,
        Severity:  body.Severity,
        Message:   body.Message,
        Timestamp: body.Timestamp,
        Details:   body.Details,
    }
    _ = h.alertService.Save(alert)
    // 高优先级告警实时推送前端
    if body.Severity == "critical" || body.Severity == "error" {
        h.broadcast(AlertEvent{Alert: alert})
    }
    return nil
}

// internal/utils/retry.go

type RetryConfig struct {
    MaxAttempts     int
    InitialInterval time.Duration
    MaxInterval     time.Duration
    Multiplier      float64
}

func RetryWithBackoff(cfg RetryConfig, fn func() error) error {
    interval := cfg.InitialInterval
    for i := 0; i < cfg.MaxAttempts; i++ {
        if err := fn(); err == nil {
            return nil
        } else if i == cfg.MaxAttempts-1 {
            return err
        }
        time.Sleep(interval)
        interval = time.Duration(float64(interval) * cfg.Multiplier)
        if interval > cfg.MaxInterval {
            interval = cfg.MaxInterval
        }
    }
    return nil
}

func (h *BaseHandler) ValidateToken(msg *Message) error {
    token := msg.Header.Token
    if token == "" {
        return errors.New("missing access_token")
    }
    return h.authService.ValidateNodeToken(msg.Header.Source, token)
}