DriverVault/API Server/internal/ocpp/session.go

package ocpp

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"
)

// Control modes. These mirror the cascade field the operator picks in Settings
// (see internal/api/integrations_ankersolix.go).
const (
	ModeOff   = "off"   // monitoring only; no CSMS (charger keeps its normal backend)
	ModeOwn   = "own"   // DriverVault is the charger's Central System
	ModeProxy = "proxy" // DriverVault forwards to Anker's cloud and taps/injects
)

// injectedPrefix namespaces the message ids of CALLs DriverVault injects toward
// the charger, so that in proxy mode a reply to one of our calls is never
// confused with a reply destined for the upstream CSMS.
const injectedPrefix = "dv-"

// callTimeout bounds how long a control command waits for the charger to answer.
const callTimeout = 30 * time.Second

// Status is a point-in-time snapshot of one charger's OCPP state, safe to return
// to clients.
type Status struct {
	Serial          string    `json:"serial"`
	Mode            string    `json:"mode"`
	Connected       bool      `json:"connected"`
	Vendor          string    `json:"vendor,omitempty"`
	Model           string    `json:"model,omitempty"`
	Firmware        string    `json:"firmware,omitempty"`
	BootedAt        time.Time `json:"bootedAt,omitempty"`
	LastHeartbeat   time.Time `json:"lastHeartbeat,omitempty"`
	ConnectorStatus string    `json:"connectorStatus,omitempty"` // Available, Charging, Faulted, …
	ErrorCode       string    `json:"errorCode,omitempty"`
	MeterWh         int64     `json:"meterWh,omitempty"`
	TransactionID   int       `json:"transactionId,omitempty"`
	LastUpdated     time.Time `json:"lastUpdated,omitempty"`
}

// CallError is returned by Session.Call when the charger answers with a CALLERROR.
type CallError struct {
	Code        string
	Description string
}

func (e *CallError) Error() string {
	return fmt.Sprintf("ocpp charger rejected call: %s: %s", e.Code, e.Description)
}

// callHandler answers an inbound CALL from the charger (own mode only). It
// returns either a result payload, or a non-empty errCode/errDesc to send a
// CALLERROR.
type callHandler func(s *Session, action string, payload json.RawMessage) (result any, errCode, errDesc string)

// Session is one live charge-point connection. In own mode it also holds the
// local dispatch handler; in proxy mode it additionally holds the upstream
// connection and pumps frames between the two while tapping the stream.
type Session struct {
	serial string
	mode   string
	cp     *Conn // charge-point (downstream) connection
	up     *Conn // upstream CSMS connection (proxy mode only)

	handler callHandler
	onClose func(*Session)
	logf    func(string, ...any)

	mu      sync.Mutex
	pending map[string]chan Message // our injected call id -> result channel
	status  Status
	closed  bool
	done    chan struct{}

	nextTxn int // own mode: assigns transaction ids
}

// newSession builds a session. Call start once it is registered to begin its
// read loop(s). In proxy mode up must be non-nil; in own mode handler must be
// non-nil.
func newSession(serial, mode string, cp, up *Conn, handler callHandler, onClose func(*Session), logf func(string, ...any)) *Session {
	if logf == nil {
		logf = func(string, ...any) {}
	}
	return &Session{
		serial:  serial,
		mode:    mode,
		cp:      cp,
		up:      up,
		handler: handler,
		onClose: onClose,
		logf:    logf,
		pending: map[string]chan Message{},
		done:    make(chan struct{}),
		status:  Status{Serial: serial, Mode: mode, Connected: true},
	}
}

// start launches the read loop(s). Separated from newSession so the CSMS can
// register the session before any inbound frame (or an immediate disconnect) can
// fire the onClose deregister callback.
func (s *Session) start() {
	go s.cpLoop()
	if s.mode == ModeProxy && s.up != nil {
		go s.upLoop()
	}
}

// Serial returns the charger serial this session serves.
func (s *Session) Serial() string { return s.serial }

// Mode returns the control mode (own|proxy).
func (s *Session) Mode() string { return s.mode }

// Snapshot returns the current status.
func (s *Session) Snapshot() Status {
	s.mu.Lock()
	defer s.mu.Unlock()
	st := s.status
	st.Serial = s.serial
	st.Mode = s.mode
	return st
}

// Call injects a CALL toward the charger and waits for its reply. It is how every
// control command (commands.go) reaches the charger, in both own and proxy modes.
func (s *Session) Call(ctx context.Context, action string, payload any) (json.RawMessage, error) {
	ctx, cancel := context.WithTimeout(ctx, callTimeout)
	defer cancel()

	id := injectedPrefix + newMessageID()
	frame, err := EncodeCall(id, action, payload)
	if err != nil {
		return nil, err
	}
	ch := make(chan Message, 1)

	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return nil, ErrClosed
	}
	s.pending[id] = ch
	s.mu.Unlock()
	defer func() {
		s.mu.Lock()
		delete(s.pending, id)
		s.mu.Unlock()
	}()

	if err := s.cp.WriteMessage(frame); err != nil {
		return nil, err
	}
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	case <-s.done:
		return nil, ErrClosed
	case m := <-ch:
		if m.Type == MessageTypeCallError {
			return nil, &CallError{Code: m.ErrorCode, Description: m.ErrorDescription}
		}
		return m.Payload, nil
	}
}

// ---- read loops --------------------------------------------------------------

func (s *Session) cpLoop() {
	defer s.close(nil)
	for {
		data, err := s.cp.ReadMessage()
		if err != nil {
			s.close(err)
			return
		}
		msg, err := DecodeMessage(data)
		if err != nil {
			s.logf("ocpp: bad frame from charger %s: %v", s.serial, err)
			continue
		}
		s.tap(msg)

		if s.mode == ModeOwn {
			s.handleLocal(msg)
			continue
		}
		// Proxy: a reply to one of OUR injected calls is consumed locally and not
		// forwarded upstream; everything else is relayed to the upstream CSMS.
		if msg.Type != MessageTypeCall && s.isPending(msg.ID) {
			s.deliver(msg)
			continue
		}
		if s.up == nil || s.up.WriteMessage(data) != nil {
			s.close(fmt.Errorf("ocpp: upstream write failed for %s", s.serial))
			return
		}
	}
}

func (s *Session) upLoop() {
	for {
		data, err := s.up.ReadMessage()
		if err != nil {
			s.close(err)
			return
		}
		if msg, derr := DecodeMessage(data); derr == nil {
			s.tap(msg)
		}
		if err := s.cp.WriteMessage(data); err != nil {
			s.close(err)
			return
		}
	}
}

// handleLocal answers an inbound frame in own mode.
func (s *Session) handleLocal(msg Message) {
	switch msg.Type {
	case MessageTypeCall:
		result, code, desc := s.handler(s, msg.Action, msg.Payload)
		var (
			out []byte
			err error
		)
		if code != "" {
			out, err = EncodeCallError(msg.ID, code, desc, nil)
		} else {
			out, err = EncodeCallResult(msg.ID, result)
		}
		if err != nil {
			s.logf("ocpp: encode response for %s/%s: %v", s.serial, msg.Action, err)
			return
		}
		_ = s.cp.WriteMessage(out)
	case MessageTypeCallResult, MessageTypeCallError:
		s.deliver(msg)
	}
}

func (s *Session) isPending(id string) bool {
	s.mu.Lock()
	_, ok := s.pending[id]
	s.mu.Unlock()
	return ok
}

func (s *Session) deliver(msg Message) {
	s.mu.Lock()
	ch := s.pending[msg.ID]
	delete(s.pending, msg.ID)
	s.mu.Unlock()
	if ch != nil {
		ch <- msg
	}
}

// tap updates the status snapshot from charger-originated notifications, so both
// own and proxy modes keep a live view without special-casing the dispatch path.
func (s *Session) tap(msg Message) {
	if msg.Type != MessageTypeCall {
		return
	}
	switch msg.Action {
	case "BootNotification":
		var p struct {
			Vendor    string `json:"chargePointVendor"`
			Model     string `json:"chargePointModel"`
			Firmware  string `json:"firmwareVersion"`
			SerialCP  string `json:"chargePointSerialNumber"`
			SerialBox string `json:"chargeBoxSerialNumber"`
		}
		_ = json.Unmarshal(msg.Payload, &p)
		s.mutate(func(st *Status) {
			st.Vendor, st.Model, st.Firmware = p.Vendor, p.Model, p.Firmware
			st.BootedAt = time.Now()
		})
	case "Heartbeat":
		s.mutate(func(st *Status) { st.LastHeartbeat = time.Now() })
	case "StatusNotification":
		var p struct {
			Status    string `json:"status"`
			ErrorCode string `json:"errorCode"`
		}
		_ = json.Unmarshal(msg.Payload, &p)
		s.mutate(func(st *Status) {
			st.ConnectorStatus = p.Status
			if p.ErrorCode != "" && p.ErrorCode != "NoError" {
				st.ErrorCode = p.ErrorCode
			} else {
				st.ErrorCode = ""
			}
		})
	case "MeterValues":
		if wh, ok := meterWh(msg.Payload); ok {
			s.mutate(func(st *Status) { st.MeterWh = wh })
		}
	case "StopTransaction":
		s.mutate(func(st *Status) { st.TransactionID = 0 })
	}
}

func (s *Session) mutate(f func(*Status)) {
	s.mu.Lock()
	f(&s.status)
	s.status.LastUpdated = time.Now()
	s.mu.Unlock()
}

func (s *Session) close(err error) {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return
	}
	s.closed = true
	s.status.Connected = false
	close(s.done)
	s.mu.Unlock()

	_ = s.cp.Close()
	if s.up != nil {
		_ = s.up.Close()
	}
	if s.onClose != nil {
		s.onClose(s)
	}
	if err != nil && err != ErrClosed {
		s.logf("ocpp: session %s closed: %v", s.serial, err)
	}
}

// meterWh best-effort extracts an Energy.Active.Import.Register reading (in Wh)
// from a MeterValues payload.
func meterWh(payload json.RawMessage) (int64, bool) {
	var p struct {
		MeterValue []struct {
			SampledValue []struct {
				Value     string `json:"value"`
				Measurand string `json:"measurand"`
				Unit      string `json:"unit"`
			} `json:"sampledValue"`
		} `json:"meterValue"`
	}
	if err := json.Unmarshal(payload, &p); err != nil {
		return 0, false
	}
	for _, mv := range p.MeterValue {
		for _, sv := range mv.SampledValue {
			// Default measurand per spec is Energy.Active.Import.Register.
			if sv.Measurand != "" && sv.Measurand != "Energy.Active.Import.Register" {
				continue
			}
			var f float64
			if _, err := fmt.Sscanf(sv.Value, "%g", &f); err != nil {
				continue
			}
			if sv.Unit == "kWh" {
				f *= 1000
			}
			return int64(f), true
		}
	}
	return 0, false
}