Cyrene_v2/internal/iot/manager.go

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。发布失败（例如 ctx 被取消）只记日志、
// 不作为错误返回：设备状态已经改变，不应因为某个观察者没收到事件就把本次控制判为
// 失败。
func (m *Manager) publishUpdate(ctx context.Context, provider string, d Device, action string, params map[string]any) {
	msg := fmt.Sprintf("设备 %s · %s（%s）", d.Name, action, provider)
	if err := m.bus.Publish(ctx, DeviceUpdateSubject, DeviceUpdate{
		Provider: provider,
		Device:   d,
		Action:   action,
		Params:   params,
		Message:  msg,
	}); 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)
}

// 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))
}