DriverVault/API Server/internal/plugins/builtin/ankersolix/cloudmqtt.go

package ankersolix

// Control over Anker's own cloud MQTT broker — the path for a charger the server
// cannot reach.
//
// The other two transports each need something a remote customer does not have.
// OCPP needs the charger to dial in to us, which means a public endpoint, a
// certificate, and a charger whose firmware accepts our CSMS. Modbus TCP needs
// the server to dial the charger, which means sharing a network with it. Most
// chargers sit behind a customer's router with neither. What they *do* have is
// the connection they already hold open to Anker: the mobile app controls them
// through it from anywhere, and it is the same broker the charger's own
// mqttStatus register reports as connected.
//
// So this transport joins that broker as the account, exactly as the app does:
//
//  1. app/devicemanage/get_user_mqtt_info hands out a client certificate and key
//     for the signed-in account, the broker's address, and the AWS root the
//     broker is verified against. The certificate is the credential; there is no
//     username or password on the MQTT connection itself.
//  2. Commands are published to cmd/{app}/{model}/{serial}/req and the charger's
//     own messages arrive on dt/{app}/{model}/{serial}/#.
//  3. Both directions carry a JSON envelope whose payload holds a base64 binary
//     frame — the device's own protocol, encoded in mqttframe.go.
//
// Two consequences shape the code:
//
//   - A connection is expensive (a TLS handshake with a fetched certificate) and
//     the plugin manager builds a throwaway instance per request, so the broker
//     connection lives in the account's shared session alongside the auth token,
//     for the same reason (see session.go). It closes itself after an idle spell.
//   - The charger does not publish its live state unless asked. A realtime
//     trigger turns the stream on for a bounded window, after which it stops
//     again — so a status read arms the trigger and waits for the next frame,
//     and a second read inside the window answers from what has since arrived.
//
// Unofficial, like the rest of the cloud half: this is the mobile app's private
// transport, and Anker may change it at any time.

import (
	"context"
	"crypto/rand"
	"crypto/tls"
	"crypto/x509"
	"encoding/base64"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"math/big"
	"net"
	"strings"
	"sync"
	"time"

	"drivervault/apiserver/internal/mqtt"
)

// epMqttInfo hands out the account's broker address and client certificate.
const epMqttInfo = "app/devicemanage/get_user_mqtt_info"

// defaultChargerModel is the product code used when the account inventory has
// not named one. The topics carry the model, and this connector is scoped to the
// V1 Smart EV Charger, so it is the only sensible fallback.
const defaultChargerModel = "A5191"

// Broker timings.
const (
	// mqttCredsTTL re-fetches the certificate periodically. It is issued per
	// account and long-lived, but refetching costs one cloud call an hour and
	// means a revoked certificate is not held forever.
	mqttCredsTTL = 1 * time.Hour

	// mqttIdle closes a broker connection nothing has used for this long. The
	// charger keeps publishing only while a trigger is live, so an idle
	// connection is genuinely idle.
	mqttIdle = 5 * time.Minute

	// mqttConnectTimeout bounds the TLS handshake and the wait for CONNACK.
	mqttConnectTimeout = 15 * time.Second

	// triggerWindow is how long the charger is asked to keep streaming telemetry,
	// and triggerRenew is how close to the end of that window a status read
	// re-arms it rather than racing the last frame.
	triggerWindow = 180 * time.Second
	triggerRenew  = 30 * time.Second

	// statusWait is how long a status read waits for a frame from the charger. A
	// triggered charger publishes every 3-5 seconds; this allows for the trigger
	// having to reach it first.
	statusWait = 12 * time.Second

	// commandWait is how long a command waits for the charger's confirmation
	// message. A command is fire-and-forget on the wire, so this only decides
	// whether we can say the charger answered — not whether it was sent.
	commandWait = 5 * time.Second

	// deviceCacheTTL is how long the account's charger list (serial to model) is
	// trusted before the cloud is asked again.
	deviceCacheTTL = 1 * time.Hour
)

// The mode values the charger's 0105 message accepts, and the names this
// connector takes for them — the same names the cloud plugin already uses for
// SolixEvChargerMode.
var mqttModeValues = map[string]uint8{
	modeStartCharge: 1,
	modeStopCharge:  2,
	modeSkipDelay:   3,
	modeBoostCharge: 4,
}

// mqttEncodingMode is the payload's encoding_type for the mode command. It is
// not encryption — the frame is plain either way — but the charger expects the
// field on this message, so it is sent with a seed like the app's.
const mqttEncodingMode = 2

// mqttCredentials is what get_user_mqtt_info returns: an address to dial and a
// certificate to dial it with.
type mqttCredentials struct {
	UserID        string `json:"user_id"`
	AppName       string `json:"app_name"`
	ThingName     string `json:"thing_name"`
	CertificateID string `json:"certificate_id"`
	CertificatePE string `json:"certificate_pem"`
	PrivateKey    string `json:"private_key"`
	EndpointAddr  string `json:"endpoint_addr"`
	RootCA        string `json:"aws_root_ca1_pem"`
}

// valid reports whether the credentials carry everything a connection needs.
func (c mqttCredentials) valid() bool {
	return strings.TrimSpace(c.EndpointAddr) != "" &&
		strings.TrimSpace(c.CertificatePE) != "" &&
		strings.TrimSpace(c.PrivateKey) != ""
}

// address is the broker's host:port. The endpoint is returned without a port;
// 8883 is the MQTT-over-TLS port the app uses.
func (c mqttCredentials) address() string {
	host := strings.TrimSpace(c.EndpointAddr)
	if _, _, err := net.SplitHostPort(host); err == nil {
		return host
	}
	return net.JoinHostPort(host, "8883")
}

// appName is the topic segment identifying the app the account belongs to.
func (c mqttCredentials) appName() string {
	if n := strings.TrimSpace(c.AppName); n != "" {
		return n
	}
	return "anker_power"
}

// ---- credentials and device lookup -------------------------------------------

// mqttCreds returns the account's broker credentials, fetching them at most once
// per mqttCredsTTL. They are held on the shared session rather than the plugin
// instance for the same reason the auth token is: the instance does not outlive
// the request.
func (p *Plugin) mqttCreds(ctx context.Context) (mqttCredentials, error) {
	s := p.sess
	if s == nil {
		return mqttCredentials{}, errors.New("anker-solix: plugin not initialised")
	}
	s.mqttMu.Lock()
	defer s.mqttMu.Unlock()

	if s.mqttCreds != nil && time.Since(s.mqttCredsAt) < mqttCredsTTL {
		return *s.mqttCreds, nil
	}
	body, err := p.apiRequest(ctx, epMqttInfo, map[string]any{})
	if err != nil {
		return mqttCredentials{}, err
	}
	var env struct {
		Data mqttCredentials `json:"data"`
	}
	if err := json.Unmarshal(body, &env); err != nil {
		return mqttCredentials{}, fmt.Errorf("anker-solix: decode MQTT info: %w", err)
	}
	if !env.Data.valid() {
		return mqttCredentials{}, errors.New("anker-solix: the cloud returned no MQTT certificate for this account; cloud control is not available on it")
	}
	s.mqttCreds, s.mqttCredsAt = &env.Data, time.Now()
	return env.Data, nil
}

// chargerModel returns the product code for a serial on this account, which the
// topics need. It doubles as the ownership check: a serial no view of the
// account reports is one this caller may not command, and saying so is better
// than publishing to a topic the broker will refuse anyway.
func (p *Plugin) chargerModel(ctx context.Context, sn string) (string, error) {
	s := p.sess
	if s == nil {
		return "", errors.New("anker-solix: plugin not initialised")
	}
	sn = strings.TrimSpace(sn)
	if sn == "" {
		return "", errors.New("anker-solix: a charger serial is required")
	}

	s.mqttMu.Lock()
	model, known := s.devices[sn]
	fresh := time.Since(s.devicesAt) < deviceCacheTTL
	s.mqttMu.Unlock()
	if known && fresh {
		return model, nil
	}

	// Unknown, or the list has gone stale: ask the cloud once and remember it.
	doc, err := p.chargerInventory(ctx)
	if err != nil {
		if known {
			return model, nil // the cloud is unreachable; the last list still stands
		}
		return "", err
	}
	found := map[string]string{}
	for _, c := range doc.Chargers {
		m := strings.ToUpper(strings.TrimSpace(c.Model))
		if m == "" {
			m = defaultChargerModel
		}
		found[c.SN] = m
	}
	s.mqttMu.Lock()
	s.devices, s.devicesAt = found, time.Now()
	s.mqttMu.Unlock()

	if m, ok := found[sn]; ok {
		return m, nil
	}
	return "", fmt.Errorf("anker-solix: charger %s is not on this Anker account", sn)
}

// ---- the broker connection ---------------------------------------------------

// mqttConn is one account's live broker connection, with the state it has
// collected from the chargers it is subscribed to.
type mqttConn struct {
	client  *mqtt.Client
	creds   mqttCredentials
	sessID  string
	started time.Time

	mu       sync.Mutex
	subs     map[string]bool         // topic filter -> subscribed
	devices  map[string]*deviceState // serial -> what it has told us
	lastUse  time.Time
	deadErr  error
	waiters  []chan struct{}
	shutdown chan struct{}
}

// deviceState is everything one charger has reported over this connection,
// merged across message types: telemetry overwrites telemetry, settings
// overwrite settings, and neither erases the other.
type deviceState struct {
	values         map[string]any
	telemetryAt    time.Time
	settingsAt     time.Time
	triggeredUntil time.Time
}

// mqttClient returns the account's broker connection, opening one if there is
// none or the last one died.
func (p *Plugin) mqttClient(ctx context.Context) (*mqttConn, error) {
	s := p.sess
	if s == nil {
		return nil, errors.New("anker-solix: plugin not initialised")
	}
	creds, err := p.mqttCreds(ctx)
	if err != nil {
		return nil, err
	}

	s.mqttMu.Lock()
	defer s.mqttMu.Unlock()
	if c := s.mqttConn; c != nil {
		if c.alive() && c.creds.CertificateID == creds.CertificateID {
			c.touch()
			return c, nil
		}
		c.close()
		s.mqttConn = nil
	}
	c, err := dialBroker(ctx, creds)
	if err != nil {
		return nil, err
	}
	s.mqttConn = c
	return c, nil
}

// dialBroker opens the mutually-authenticated connection. The account's
// certificate is the credential, and the broker is verified against the AWS root
// the same response supplied — the connection is to Anker's own broker, so
// neither side is trusted on the strength of the other.
func dialBroker(ctx context.Context, creds mqttCredentials) (*mqttConn, error) {
	cert, err := tls.X509KeyPair([]byte(creds.CertificatePE), []byte(creds.PrivateKey))
	if err != nil {
		return nil, fmt.Errorf("anker-solix: the cloud's MQTT certificate could not be loaded: %w", err)
	}
	roots := x509.NewCertPool()
	if ca := strings.TrimSpace(creds.RootCA); ca != "" {
		if !roots.AppendCertsFromPEM([]byte(ca)) {
			return nil, errors.New("anker-solix: the cloud's MQTT root certificate could not be parsed")
		}
	} else {
		// No root supplied: fall back to the system pool rather than skipping
		// verification, which would let anything answer for the broker.
		if roots, err = x509.SystemCertPool(); err != nil {
			return nil, fmt.Errorf("anker-solix: no root certificates to verify the MQTT broker: %w", err)
		}
	}
	host, _, splitErr := net.SplitHostPort(creds.address())
	if splitErr != nil {
		host = creds.EndpointAddr
	}

	client, err := mqtt.Connect(ctx, mqtt.Options{
		Address: creds.address(),
		TLS:     true,
		TLSConfig: &tls.Config{
			ServerName:   host,
			MinVersion:   tls.VersionTLS12,
			RootCAs:      roots,
			Certificates: []tls.Certificate{cert},
		},
		// The broker keys a session by client id and evicts the older holder, so
		// the app's own connection must not be displaced: the app uses
		// "{thing_name}_{5 digits}", and a different suffix is a different session.
		ClientID:       clientIDFor(creds),
		Keepalive:      60 * time.Second,
		ConnectTimeout: mqttConnectTimeout,
		Buffer:         256,
	})
	if err != nil {
		return nil, err
	}

	c := &mqttConn{
		client:   client,
		creds:    creds,
		sessID:   randomSessionID(),
		started:  time.Now(),
		subs:     map[string]bool{},
		devices:  map[string]*deviceState{},
		lastUse:  time.Now(),
		shutdown: make(chan struct{}),
	}
	go c.readLoop()
	go c.idleLoop()
	return c, nil
}

// clientIDFor builds an identifier no other holder of this account's certificate
// is using, so joining the broker never evicts the owner's mobile app.
func clientIDFor(creds mqttCredentials) string {
	thing := strings.TrimSpace(creds.ThingName)
	if thing == "" {
		thing = strings.TrimSpace(creds.UserID)
	}
	return fmt.Sprintf("%s_%05d", thing, randomBelow(100000))
}

// randomSessionID mimics the app's sess_id, a pair of four-digit groups.
func randomSessionID() string {
	return fmt.Sprintf("%04d-%04d", randomBelow(10000), randomBelow(10000))
}

// randomBelow returns a non-negative integer below n, falling back to a
// clock-derived value if the system source fails.
func randomBelow(n int64) int64 {
	v, err := rand.Int(rand.Reader, big.NewInt(n))
	if err != nil {
		return time.Now().UnixNano() % n
	}
	return v.Int64()
}

// alive reports whether the connection is still usable.
func (c *mqttConn) alive() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.deadErr == nil
}

// touch marks the connection as in use, so the idle sweep leaves it alone.
func (c *mqttConn) touch() {
	c.mu.Lock()
	c.lastUse = time.Now()
	c.mu.Unlock()
}

// close ends the connection and wakes anything waiting on a message.
func (c *mqttConn) close() {
	c.fail(mqtt.ErrClosed)
	_ = c.client.Close()
}

// fail records why the connection ended and releases every waiter.
func (c *mqttConn) fail(err error) {
	c.mu.Lock()
	if c.deadErr == nil {
		c.deadErr = err
		close(c.shutdown)
	}
	c.wakeLocked()
	c.mu.Unlock()
}

// wakeLocked releases everything waiting for a device message. Caller holds mu.
func (c *mqttConn) wakeLocked() {
	for _, ch := range c.waiters {
		close(ch)
	}
	c.waiters = nil
}

// readLoop owns the inbound stream for the life of the connection.
func (c *mqttConn) readLoop() {
	for msg := range c.client.Messages() {
		c.ingest(msg)
	}
	err := c.client.Err()
	if err == nil {
		err = mqtt.ErrClosed
	}
	c.fail(err)
}

// idleLoop closes a connection nothing has used for mqttIdle. The account keeps
// no state on the broker between commands, so dropping the socket costs only the
// next handshake.
func (c *mqttConn) idleLoop() {
	t := time.NewTicker(mqttIdle / 2)
	defer t.Stop()
	for {
		select {
		case <-c.shutdown:
			return
		case <-t.C:
			c.mu.Lock()
			idle := time.Since(c.lastUse)
			c.mu.Unlock()
			if idle >= mqttIdle {
				c.close()
				return
			}
		}
	}
}

// ingest decodes one inbound message and folds it into the sending charger's
// state. Anything it cannot read is dropped: these frames come from a cloud
// connection, and a malformed one must not be recorded as a reading.
func (c *mqttConn) ingest(msg mqtt.Message) {
	sn, data, ok := parseEnvelope(msg)
	if !ok {
		return
	}
	msgType, values, err := decodeFrame(data)
	if err != nil || len(values) == 0 {
		return
	}

	c.mu.Lock()
	defer c.mu.Unlock()
	st := c.devices[sn]
	if st == nil {
		st = &deviceState{values: map[string]any{}}
		c.devices[sn] = st
	}
	for k, v := range values {
		st.values[k] = v
	}
	now := time.Now()
	// Only a message this package can name counts as a report. An unnamed one
	// still leaves its fields behind, but it must not stamp settingsAt: that
	// timestamp is what a control command waits on to say the charger
	// acknowledged, and a frame we cannot read is not an acknowledgement.
	switch {
	case msgType == msgEVTelemetry:
		st.telemetryAt = now
	case evMessages[msgType] != nil:
		st.settingsAt = now
	}
	c.wakeLocked()
}

// parseEnvelope pulls the sending serial and the binary frame out of one MQTT
// message. The payload is a JSON string inside a JSON object, and the frame is
// base64 inside that — the shape the app both sends and receives.
func parseEnvelope(msg mqtt.Message) (string, []byte, bool) {
	var env struct {
		Head struct {
			DeviceSN string `json:"device_sn"`
		} `json:"head"`
		Payload string `json:"payload"`
	}
	if err := json.Unmarshal(msg.Payload, &env); err != nil {
		return "", nil, false
	}
	var inner struct {
		SN   string `json:"sn"`
		SN2  string `json:"device_sn"`
		Data string `json:"data"`
	}
	if err := json.Unmarshal([]byte(env.Payload), &inner); err != nil {
		return "", nil, false
	}
	sn := firstNonEmpty(inner.SN, inner.SN2, env.Head.DeviceSN, serialFromTopic(msg.Topic))
	if sn == "" || inner.Data == "" {
		return "", nil, false
	}
	data, err := base64.StdEncoding.DecodeString(inner.Data)
	if err != nil {
		return "", nil, false
	}
	return sn, data, true
}

// serialFromTopic reads the serial out of dt/{app}/{model}/{serial}/… , which is
// where it is when the payload does not repeat it.
func serialFromTopic(topic string) string {
	parts := strings.Split(topic, "/")
	if len(parts) < 4 {
		return ""
	}
	return parts[3]
}

func firstNonEmpty(vals ...string) string {
	for _, v := range vals {
		if s := strings.TrimSpace(v); s != "" {
			return s
		}
	}
	return ""
}

// ---- topics, subscribing and publishing --------------------------------------

// dataTopic is the filter carrying everything one charger publishes.
func dataTopic(creds mqttCredentials, model, sn string) string {
	return fmt.Sprintf("dt/%s/%s/%s/#", creds.appName(), model, sn)
}

// commandTopic is where one charger's commands are published.
func commandTopic(creds mqttCredentials, model, sn string) string {
	return fmt.Sprintf("cmd/%s/%s/%s/req", creds.appName(), model, sn)
}

// listen subscribes to a charger's data topic, once per connection.
func (c *mqttConn) listen(ctx context.Context, model, sn string) error {
	topic := dataTopic(c.creds, model, sn)
	c.mu.Lock()
	already := c.subs[topic]
	c.mu.Unlock()
	if already {
		return nil
	}
	if err := c.client.Subscribe(ctx, topic); err != nil {
		return fmt.Errorf("anker-solix: cannot listen to charger %s over the cloud: %w", sn, err)
	}
	c.mu.Lock()
	c.subs[topic] = true
	c.mu.Unlock()
	return nil
}

// publishFrame wraps a device frame in the app's envelope and publishes it to
// the charger's command topic.
func (c *mqttConn) publishFrame(ctx context.Context, model, sn string, frame []byte, encoding int) error {
	now := time.Now()
	seed := any(1)
	inner := map[string]any{
		"device_sn":  sn,
		"account_id": c.creds.UserID,
		"data":       base64.StdEncoding.EncodeToString(frame),
	}
	if encoding != 0 {
		inner["encoding_type"] = encoding
		seed = randomSeed()
	}
	payload, err := json.Marshal(inner)
	if err != nil {
		return err
	}
	envelope, err := json.Marshal(map[string]any{
		"head": map[string]any{
			"version":   "1.0.0.1",
			"client_id": fmt.Sprintf("android-%s-%s-%s", c.creds.appName(), c.creds.UserID, c.creds.CertificateID),
			"sess_id":   c.sessID,
			"msg_seq":   1,
			"seed":      seed,
			"timestamp": now.Unix(),
			// cmd_status 2 and cmd 17 are what the app sends on a control message;
			// the charger ignores neither, and a different pair goes unanswered.
			"cmd_status": 2,
			"cmd":        17,
			"sign_code":  1,
			"device_pn":  model,
			"device_sn":  sn,
		},
		"payload": string(payload),
	})
	if err != nil {
		return err
	}
	c.touch()
	return c.client.Publish(ctx, commandTopic(c.creds, model, sn), envelope)
}

// randomSeed is the 16-byte seed the app puts in the header of an encoded
// message.
func randomSeed() string {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		binary.LittleEndian.PutUint64(b[:8], uint64(time.Now().UnixNano()))
	}
	return encodeHex(b[:])
}

// ---- waiting for the charger to answer ----------------------------------------

// waitFor blocks until a charger's state satisfies ready, or the deadline
// passes. It reports whether ready was met; a connection that dies while waiting
// ends the wait with the reason.
func (c *mqttConn) waitFor(ctx context.Context, sn string, ready func(*deviceState) bool, timeout time.Duration) (bool, error) {
	deadline := time.After(timeout)
	for {
		c.mu.Lock()
		if err := c.deadErr; err != nil {
			c.mu.Unlock()
			return false, fmt.Errorf("anker-solix: the cloud connection dropped: %w", err)
		}
		if st := c.devices[sn]; st != nil && ready(st) {
			c.mu.Unlock()
			return true, nil
		}
		ch := make(chan struct{})
		c.waiters = append(c.waiters, ch)
		c.mu.Unlock()

		select {
		case <-ch:
		case <-deadline:
			return false, nil
		case <-ctx.Done():
			return false, ctx.Err()
		}
	}
}

// snapshotOf copies a charger's collected state out from under the lock.
func (c *mqttConn) snapshotOf(sn string) (map[string]any, time.Time, time.Time, time.Time) {
	c.mu.Lock()
	defer c.mu.Unlock()
	st := c.devices[sn]
	if st == nil {
		return nil, time.Time{}, time.Time{}, time.Time{}
	}
	out := make(map[string]any, len(st.values))
	for k, v := range st.values {
		out[k] = v
	}
	return out, st.telemetryAt, st.settingsAt, st.triggeredUntil
}

// noteTrigger records how long the charger has been asked to keep streaming.
func (c *mqttConn) noteTrigger(sn string, until time.Time) {
	c.mu.Lock()
	defer c.mu.Unlock()
	st := c.devices[sn]
	if st == nil {
		st = &deviceState{values: map[string]any{}}
		c.devices[sn] = st
	}
	st.triggeredUntil = until
}

// ---- commands ----------------------------------------------------------------

// mqttTrigger asks a charger to publish live telemetry for a while. Without it
// the charger is silent, so every status read arms one.
func (p *Plugin) mqttTrigger(ctx context.Context, c *mqttConn, model, sn string, window time.Duration) error {
	frame, err := encodeFrame(msgRealtimeTrigger, []cmdField{
		rawField(0xa1, 0x22),
		uintField(0xa2, 1),
		varField(0xa3, uint32(window/time.Second)),
		timestampField(time.Now()),
	})
	if err != nil {
		return err
	}
	if err := c.publishFrame(ctx, model, sn, frame, 0); err != nil {
		return err
	}
	c.noteTrigger(sn, time.Now().Add(window))
	return nil
}

// mqttSetMode sends the start / stop / skip-delay / boost command.
func (p *Plugin) mqttSetMode(ctx context.Context, c *mqttConn, model, sn, mode string) error {
	v, ok := mqttModeValues[mode]
	if !ok {
		return fmt.Errorf("anker-solix: %q is not one of %s, %s, %s or %s",
			mode, modeStartCharge, modeStopCharge, modeSkipDelay, modeBoostCharge)
	}
	frame, err := encodeFrame(msgEVMode, []cmdField{
		rawField(0xa1, 0x22),
		uintField(0xa2, v),
		timestampField(time.Now()),
	})
	if err != nil {
		return err
	}
	return c.publishFrame(ctx, model, sn, frame, mqttEncodingMode)
}

// mqttSetMaxCurrent sets the charging current ceiling, in amps. The limit is
// checked by the same rule the Modbus path uses, because the rule is the
// charger's: the transport differs, the charger does not.
func (p *Plugin) mqttSetMaxCurrent(ctx context.Context, c *mqttConn, model, sn string, amps float64) error {
	if err := checkMaxCurrent(amps); err != nil {
		return err
	}
	// It is the same wire field the "maxCurrentA" setting writes, so it is built
	// from the same table rather than encoded a second time here (mqttsettings.go).
	frame, err := encodeSettingCmd(settingCmds[settingIndex["maxCurrentA"]],
		map[string]any{"maxCurrentA": amps}, nil)
	if err != nil {
		return err
	}
	return c.publishFrame(ctx, model, sn, frame, 0)
}