package iot import ( "context" "errors" "fmt" "io" "log/slog" "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/bus" "git.yeij.top/AskaEth/Cyrene_v2/internal/config" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" ) // Manager 是 IoT 子系统的门面:依据配置选定 Provider,对外提供 // List/Control/Status,并把设备状态跃迁以 DeviceUpdate 事件的形式发布到进程内 bus。 type Manager struct { providers []Provider bus bus.Bus log *slog.Logger } // Option 用于定制由 New 构建的 Manager。 type Option func(*Manager) // WithBus 注入现成的 bus,而不让 Manager 自建一个。此设计与其余子系统保持一致: // iot.device_update 由此流经共享 bus,能到达 gateway hub / 后台触发订阅者, // 而不是滞留在一个 Manager 私有的隔离 bus 上。 func WithBus(b bus.Bus) Option { return func(m *Manager) { m.bus = b } } // New 依据 cfg 构建 Manager,并以 cfg.Provider 选定设备 Provider。 // // 支持的 Provider 取值:"" 或 "mock" 注册内置 mock Provider(八台模拟设备); // "none" 关闭该子系统(不注册任何 Provider,List 返回空切片,Control/Status 以 // ErrDeviceNotFound 失败)。其他取值返回 ErrUnknownProvider。 // // log 可为 nil,此时会替换为丢弃输出的 logger。Manager 自持一个全新且隔离的 bus, // 经 Bus 暴露给订阅方挂接。 func New(cfg config.IoTConfig, log *slog.Logger, opts ...Option) (*Manager, error) { m := &Manager{ bus: bus.New(), log: logger.WithSource(discardLogger(log), "iot/manager"), } for _, opt := range opts { opt(m) } switch strings.TrimSpace(cfg.Provider) { case "", "mock": mp, err := newMockProvider(log) if err != nil { return nil, err } m.providers = append(m.providers, mp) case "none": // 不注册任何 Provider:子系统处于关闭状态。 default: return nil, fmt.Errorf("iot: new: provider %q: %w", cfg.Provider, ErrUnknownProvider) } m.log.Info("iot manager initialized", "provider", cfg.Provider, "providers", len(m.providers)) return m, nil } // Register 把 p 注册为设备来源。注册的 Provider 若 Name 已存在,返回 ErrDuplicate; // p 为 nil 则返回 ErrInvalidArgument。 func (m *Manager) Register(p Provider) error { if p == nil { return fmt.Errorf("iot: register: %w", ErrInvalidArgument) } for _, existing := range m.providers { if existing.Name() == p.Name() { return fmt.Errorf("iot: register %q: %w", p.Name(), ErrDuplicate) } } m.providers = append(m.providers, p) m.log.Info("iot provider registered", "provider", p.Name()) return nil } // Bus 返回 Manager 的进程内事件总线。订阅方挂接 DeviceUpdateSubject // (或 "iot.>")以观察设备状态跃迁。 func (m *Manager) Bus() bus.Bus { return m.bus } // List 返回全部已注册 Provider 的所有设备。 func (m *Manager) List(ctx context.Context) ([]Device, error) { if err := ctx.Err(); err != nil { return nil, err } var out []Device for _, p := range m.providers { devices, err := p.List(ctx) if err != nil { return nil, fmt.Errorf("iot: list %q: %w", p.Name(), err) } out = append(out, devices...) } m.log.Debug("iot devices listed", "count", len(out)) return out, nil } // Control 把 action 施加到拥有该设备的 Provider 上,成功后向 bus 发布一条 // DeviceUpdate。若没有任何 Provider 拥有该设备,返回 ErrDeviceNotFound。 func (m *Manager) Control(ctx context.Context, deviceID, action string, params map[string]any) (*Device, error) { if err := ctx.Err(); err != nil { return nil, err } if len(m.providers) == 0 { return nil, fmt.Errorf("iot: control: %w", ErrProviderNotRegistered) } for _, p := range m.providers { d, err := p.Control(ctx, deviceID, action, params) if err != nil { if errors.Is(err, ErrDeviceNotFound) { continue } return nil, err } m.publishUpdate(ctx, p.Name(), *d, action, params) return d, nil } return nil, fmt.Errorf("iot: control %q: %w", deviceID, ErrDeviceNotFound) } // Status 返回由匹配 Provider 所持有设备的当前状态。若没有任何 Provider 拥有该 // 设备,返回 ErrDeviceNotFound。 func (m *Manager) Status(ctx context.Context, deviceID string) (*Device, error) { if err := ctx.Err(); err != nil { return nil, err } if len(m.providers) == 0 { return nil, fmt.Errorf("iot: status: %w", ErrProviderNotRegistered) } for _, p := range m.providers { d, err := p.Status(ctx, deviceID) if err != nil { if errors.Is(err, ErrDeviceNotFound) { continue } return nil, err } return d, nil } return nil, fmt.Errorf("iot: status %q: %w", deviceID, ErrDeviceNotFound) } // publishUpdate 把 DeviceUpdate 扇出到 bus。发布的载荷带上本次跃迁的显式严重度: // 命中安全规则(safetySeverity)的跃迁被标成安全域,其余一律 SeverityRoutine, // 因此常规设备变化不会被误升级。发布失败(例如 ctx 被取消)只记日志、 // 不作为错误返回:设备状态已经改变,不应因为某个观察者没收到事件就把本次控制判为 // 失败。 func (m *Manager) publishUpdate(ctx context.Context, provider string, d Device, action string, params map[string]any) { severity, safety := safetySeverity(d.Type, action) msg := fmt.Sprintf("设备 %s · %s(%s)", d.Name, action, provider) if safety { msg = fmt.Sprintf("%s · 安全域 %s", msg, severity.String()) } if err := m.bus.Publish(ctx, DeviceUpdateSubject, DeviceUpdate{ Provider: provider, Device: d, Action: action, Params: params, Message: msg, Severity: severity, }); err != nil { m.log.Error("iot publish device update", "device", d.ID, "action", action, "error", err) return } m.log.Debug("iot device update published", "device", d.ID, "action", action, "provider", provider, "severity", severity.String()) } // discardLogger 在 log 非 nil 时原样返回,否则返回一个不写任何输出的 logger, // 保证 New 始终有可用的 logger。 func discardLogger(log *slog.Logger) *slog.Logger { if log != nil { return log } return slog.New(slog.NewTextHandler(io.Discard, nil)) }