ha-n95-local-control/internal/robot/actor.go

package robot

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sync"
	"time"

	"git.i3omb.com/gronod/ha-n95-local-control/internal/ctl"
	"git.i3omb.com/gronod/ha-n95-local-control/internal/session"
)

// Command is one vacuum command submitted to the actor. This phase accepts
// the §11 names start, stop, return_to_base, clean_spot, locate and
// set_fan_speed; every other name is rejected without writing XMPP.
type Command struct {
	Name    string
	Args    map[string]string
	Payload []byte
	Reject  string
}

// Clock abstracts time so tests can advance the correlation deadline without
// sleeping.
type Clock interface {
	Now() time.Time
	After(d time.Duration) <-chan time.Time
}

type realClock struct{}

func (realClock) Now() time.Time                         { return time.Now() }
func (realClock) After(d time.Duration) <-chan time.Time { return time.After(d) }

var (
	errOffline        = errors.New("offline")
	errConnectionLost = errors.New("connection-lost")
)

type evReady struct{ e session.ReadyEvent }
type evAnnounce struct{ e session.ReadyEvent }
type evDown struct{ e session.DownEvent }
type evStanza struct{ e session.StanzaEvent }
type evSubmit struct {
	cmd   Command
	reply chan error
}

type evSetRepublish struct{ fn RepublishFunc }
type evSetSendCommand struct{ fn SendCommandFunc }

// evBarrier is a test synchronization point: handling it closes the channel
// once every event queued before it has been processed.
type evBarrier chan struct{}

// Actor serializes one robot's commands, ctl correlations and state mutation.
// It implements session.Observer; observer calls enqueue on a mailbox
// buffered to 64 and processed by a single goroutine. Stanza delivery may
// block: stanza order is the point of the actor (§14).
type Actor struct {
	ctx           context.Context
	jid           string
	serial        string
	controllerJID string
	republish     RepublishFunc
	trace         ctl.TraceSink
	diag          session.DiagnosticSink
	clock         Clock
	mailbox       chan any

	// Fields below are owned by the actor goroutine.
	snap          Snapshot
	corr          *ctl.Correlator
	gen           uint64
	send          func([]byte) error
	requestedFan  map[string]string
	sendCommandFn SendCommandFunc
}

// NewActor starts the actor goroutine. A nil clock uses real time.
func NewActor(ctx context.Context, jid, serial, controllerJID string, republish RepublishFunc, trace ctl.TraceSink, diag session.DiagnosticSink, clock Clock) *Actor {
	if clock == nil {
		clock = realClock{}
	}
	a := &Actor{
		ctx:           ctx,
		jid:           jid,
		serial:        serial,
		controllerJID: controllerJID,
		republish:     republish,
		trace:         trace,
		diag:          diag,
		clock:         clock,
		mailbox:       make(chan any, 64),
		snap:          NewSnapshot(),
		requestedFan:  map[string]string{},
	}
	a.corr = ctl.NewCorrelator(clock.Now)
	go a.run()
	return a
}

func (a *Actor) SessionReady(e session.ReadyEvent) {
	if e.JID == a.jid {
		a.enqueue(evReady{e})
	}
}

func (a *Actor) AnnounceOK(e session.ReadyEvent) {
	if e.JID == a.jid {
		a.enqueue(evAnnounce{e})
	}
}

func (a *Actor) SessionDown(e session.DownEvent) {
	if e.JID == a.jid {
		a.enqueue(evDown{e})
	}
}

func (a *Actor) Stanza(e session.StanzaEvent) {
	if e.JID == a.jid {
		a.enqueue(evStanza{e})
	}
}

func (a *Actor) enqueue(ev any) {
	select {
	case a.mailbox <- ev:
	case <-a.ctx.Done():
	}
}

// Submit only enqueues work; the actor goroutine remains the sole Snapshot
// mutator and the only RepublishFunc caller. Callers — including the phase 04
// MQTT callback — must never invoke RepublishFunc themselves, so a publish
// cannot deadlock the callback. Submit returns once the send attempt has run:
// nil on a successful write, or the exact string offline, connection-lost or
// rejected:<name>.
func (a *Actor) Submit(ctx context.Context, command Command) error {
	reply := make(chan error, 1)
	select {
	case a.mailbox <- evSubmit{cmd: command, reply: reply}:
	case <-ctx.Done():
		return ctx.Err()
	case <-a.ctx.Done():
		return errConnectionLost
	}
	select {
	case err := <-reply:
		return err
	case <-ctx.Done():
		return ctx.Err()
	case <-a.ctx.Done():
		return errConnectionLost
	}
}

func (a *Actor) run() {
	var expire <-chan time.Time
	arm := func() {
		wait := ctl.CommandTimeout
		if deadline, ok := a.corr.NextDeadline(); ok {
			if wait = deadline.Sub(a.clock.Now()); wait < 0 {
				wait = 0
			}
		}
		expire = a.clock.After(wait)
	}
	arm()
	for {
		select {
		case <-a.ctx.Done():
			return
		case ev := <-a.mailbox:
			a.handle(ev)
			arm()
		case now := <-expire:
			for _, tr := range a.corr.Expire(now) {
				a.handleTrace(tr)
			}
			arm()
		}
	}
}

func (a *Actor) handle(ev any) {
	switch e := ev.(type) {
	case evReady:
		a.gen = e.e.Generation
		a.send = e.e.Send
		a.serial = e.e.Serial
	case evAnnounce:
		a.onAnnounce(e.e)
	case evDown:
		a.onDown(e.e)
	case evStanza:
		a.onStanza(e.e)
	case evSubmit:
		a.onSubmit(e)
	case evSetRepublish:
		a.republish = e.fn
	case evSetSendCommand:
		a.sendCommandFn = e.fn
	case evBarrier:
		close(e)
	}
}

// onAnnounce runs the §8 fan-out for the current generation only. Results are
// not awaited between sends; correlations complete as stanzas arrive.
func (a *Actor) onAnnounce(e session.ReadyEvent) {
	if e.Generation != a.gen || a.send == nil {
		return
	}
	for _, item := range readySequence(a.clock.Now()) {
		_ = a.sendCommand(item.out, item.name, "")
	}
}

// onDown ends every outstanding correlation for the current generation as
// connection-lost, clears the generation, drains pending command submissions
// the same way, and leaves the snapshot in place. A down event for an older
// generation is ignored.
func (a *Actor) onDown(e session.DownEvent) {
	if e.Generation != a.gen {
		return
	}
	traces := a.corr.FailGeneration("connection-lost")
	a.gen = 0
	a.send = nil
	a.requestedFan = map[string]string{}
	for _, tr := range traces {
		a.handleTrace(tr)
	}
	for {
		select {
		case ev := <-a.mailbox:
			if s, ok := ev.(evSubmit); ok {
				a.failSubmit(s, errConnectionLost)
			} else {
				a.handle(ev)
			}
		default:
			return
		}
	}
}

func (a *Actor) onStanza(e session.StanzaEvent) {
	if e.Generation != a.gen {
		return
	}
	in, err := ctl.Parse(e.Stanza)
	if err != nil {
		a.diagnose(e, "malformed stanza: "+err.Error())
		return
	}
	switch in.Kind {
	case ctl.KindAck:
		tr, ok := a.corr.CompleteSID(in.SID)
		if !ok {
			a.diagnose(e, "iq result with no outstanding stanza id")
			return
		}
		a.handleTrace(tr)
	case ctl.KindResult:
		tr, ok := a.corr.CompleteCID(in.CID)
		if !ok {
			// Unknown or already completed cid: ignored, no state change.
			return
		}
		tr.Ret = in.Ret
		tr.Errno = in.Errno
		// Results carry no td; reuse the registered command name so Apply can
		// route payloads such as GetLifeSpan and SetCleanSpeed.
		in.TD = tr.Command
		fan := a.requestedFan[in.CID]
		delete(a.requestedFan, in.CID)
		a.emitTrace(tr)
		var applyErr error
		if in.Ret == "ok" {
			applyErr = Apply(&a.snap, in, fan)
		}
		a.snap.SetCommandError(commandErrorFor(tr))
		if applyErr != nil {
			a.diagnose(e, applyErr.Error())
		}
		a.publish()
	case ctl.KindPush, ctl.KindBattery:
		// Unsolicited pushes and bare battery are never acked (§6.3).
		before := a.snap.Clone()
		err := Apply(&a.snap, in, "")
		if err != nil {
			a.diagnose(e, err.Error())
		}
		if err == nil || !reflect.DeepEqual(before, a.snap) {
			a.publish()
		}
	default:
		a.diagnose(e, "unparsed stanza")
	}
}

func (a *Actor) onSubmit(s evSubmit) {
	if a.gen == 0 || a.send == nil {
		a.failSubmit(s, errOffline)
		return
	}
	if s.cmd.Reject != "" {
		a.failSubmit(s, errors.New("rejected:"+s.cmd.Reject))
		return
	}
	if s.cmd.Name == "send_command" {
		if a.sendCommandFn == nil {
			a.failSubmit(s, errors.New("rejected:unsupported"))
			return
		}
		cmd, err := a.sendCommandFn(a.ctx, a.serial, s.cmd.Payload)
		if err != nil {
			a.failSubmit(s, err)
			return
		}
		s.cmd = cmd
	}
	out, name, fan, ok := commandStanza(s.cmd, a.snap.Facts.Fan, a.clock.Now())
	if !ok {
		a.failSubmit(s, errors.New("rejected:"+s.cmd.Name))
		return
	}
	if err := a.sendCommand(out, name, fan); err != nil {
		s.reply <- errConnectionLost
		return
	}
	s.reply <- nil
}

func (a *Actor) failSubmit(s evSubmit, err error) {
	v := err.Error()
	a.snap.SetCommandError(&v)
	a.publish()
	s.reply <- err
}

// sendCommand writes one ctl envelope and registers its correlation. A failed
// write fails the command locally as connection-lost and does not register a
// waiter.
func (a *Actor) sendCommand(out ctl.Outbound, name, fan string) error {
	stanza, sid, cid, err := ctl.Envelope(a.controllerJID, a.jid, out)
	if err != nil {
		return err
	}
	if err := a.send(stanza); err != nil {
		phase := "result"
		if out.OmitCtlID {
			phase = "ack"
		}
		a.handleTrace(ctl.Trace{
			SID:     sid,
			CID:     cid,
			Command: name,
			Phase:   phase,
			Ret:     "connection-lost",
			At:      a.clock.Now(),
		})
		return err
	}
	a.corr.Register(sid, cid, name, !out.OmitCtlID)
	if fan != "" {
		a.requestedFan[cid] = fan
	}
	return nil
}

func (a *Actor) handleTrace(tr ctl.Trace) {
	a.emitTrace(tr)
	if tr.CID != "" && (tr.Ret == "timeout" || tr.Ret == "connection-lost") {
		delete(a.requestedFan, tr.CID)
	}
	a.snap.SetCommandError(commandErrorFor(tr))
	a.publish()
}

func (a *Actor) emitTrace(tr ctl.Trace) {
	if a.trace != nil {
		a.trace(a.serial, tr)
	}
}

func (a *Actor) publish() {
	if a.republish != nil {
		a.republish(a.ctx, a.snap.Clone())
	}
}

func (a *Actor) diagnose(e session.StanzaEvent, reason string) {
	if a.diag == nil {
		return
	}
	a.diag(session.Diagnostic{
		Serial:     a.serial,
		Direction:  session.DirectionIn,
		Kind:       "unparsed",
		Generation: e.Generation,
		XML:        e.Stanza,
		Reason:     reason,
	})
}

// commandErrorFor maps a trace to last_command_error: an ack or ret ok
// clears, ret fail stores fail[:errno], and failure reasons such as timeout
// and connection-lost are stored verbatim.
func commandErrorFor(tr ctl.Trace) *string {
	switch tr.Ret {
	case "", "ok":
		return nil
	case "fail":
		if tr.Errno != nil && *tr.Errno != "" {
			v := "fail:" + *tr.Errno
			return &v
		}
		v := "fail"
		return &v
	default:
		v := tr.Ret
		return &v
	}
}

// commandStanza maps a submitted command name to its §11 ctl body. It returns
// the correlator name — the ctl td — and, for SetCleanSpeed, the requested fan
// stored against the cid.
func commandStanza(cmd Command, fan string, now time.Time) (out ctl.Outbound, name, requestedFan string, ok bool) {
	switch cmd.Name {
	case "start":
		return cleanOutbound("auto", "s", fan), "Clean", "", true
	case "stop":
		return cleanOutbound("stop", "h", fan), "Clean", "", true
	case "clean_spot":
		return cleanOutbound("spot", "s", fan), "Clean", "", true
	case "return_to_base":
		return chargeOutbound("go"), "Charge", "", true
	case "locate":
		return playSoundOutbound(), "PlaySound", "", true
	case "set_fan_speed":
		speed := cmd.Args["speed"]
		if !validFan(speed) {
			return ctl.Outbound{}, "", "", false
		}
		return setCleanSpeedOutbound(speed), "SetCleanSpeed", speed, true
	case "clean":
		return cleanOutbound(cmd.Args["type"], "s", fan), "Clean", "", true
	case "move":
		return moveOutbound(cmd.Args["action"]), "Move", "", true
	case "cancel_return":
		return chargeOutbound("stopGo"), "Charge", "", true
	case "set_time":
		return setTimeOutbound(now), "SetTime", "", true
	case "get_status":
		return ctl.Outbound{}, "GetStatus", "", true
	case "get_lifespan":
		return ctl.Outbound{}, "GetLifeSpan", "", true
	case "add_sched":
		return ctl.AddSchedOutbound(cmd.Args["name"], cmd.Args["on"], cmd.Args["time"], cmd.Args["repeat"]), "AddSched", "", true
	case "mod_sched":
		return ctl.ModSchedOutbound(cmd.Args["name"], cmd.Args["on"], cmd.Args["time"], cmd.Args["repeat"]), "ModSched", "", true
	case "del_sched":
		return ctl.DelSchedOutbound(cmd.Args["name"]), "DelSched", "", true
	case "get_sched":
		return ctl.GetSchedOutbound(), "GetSched", "", true
	case "raw":
		out := ctl.Outbound{
			TD:    cmd.Args["td"],
			Inner: cmd.Payload,
			OmitCtlID: cmd.Args["id"] == "",
		}
		return out, cmd.Args["td"], "", true
	}
	return ctl.Outbound{}, "", "", false
}

func setTimeOutbound(now time.Time) ctl.Outbound {
	_, offset := now.Zone()
	hours := offset / 3600
	minutes := (offset % 3600) / 60
	return ctl.Outbound{
		TD:    "SetTime",
		Inner: []byte(fmt.Sprintf(`<time t="%d" tz="%d" tzm="%d"/>`, now.Unix(), hours, minutes)),
	}
}

// RepublishFactory builds a per-serial republish hook; the fleet invokes it
// when creating an actor on first SessionReady.
type RepublishFactory func(serial string) RepublishFunc

// SendCommandFunc executes a send_command payload; an error fails
// the command.
type SendCommandFunc func(ctx context.Context, serial string, payload []byte) (Command, error)

// Fleet is the session.Observer that owns one Actor per full bot JID. It
// creates the actor on the first SessionReady and forwards every later event
// for that JID to it.
type Fleet struct {
	ctx           context.Context
	controllerJID string
	republish     RepublishFunc
	trace         ctl.TraceSink
	diag          session.DiagnosticSink
	clock         Clock

	mu               sync.Mutex
	actors           map[string]*Actor
	actorsBySerial   map[string]*Actor
	republishFactory RepublishFactory
	sendCommandFn    SendCommandFunc
}

func NewFleet(ctx context.Context, controllerJID string, republish RepublishFunc, trace ctl.TraceSink, diag session.DiagnosticSink, clock Clock) *Fleet {
	return &Fleet{
		ctx:            ctx,
		controllerJID:  controllerJID,
		republish:      republish,
		trace:          trace,
		diag:           diag,
		clock:          clock,
		actors:         map[string]*Actor{},
		actorsBySerial: map[string]*Actor{},
	}
}

// Actor returns the actor for jid, if one has been created.
func (f *Fleet) Actor(jid string) (*Actor, bool) {
	f.mu.Lock()
	defer f.mu.Unlock()
	a, ok := f.actors[jid]
	return a, ok
}

// SetRepublishFactory installs the per-serial republish factory before or
// after events; existing actors receive it through their mailbox.
func (f *Fleet) SetRepublishFactory(factory RepublishFactory) {
	f.mu.Lock()
	f.republishFactory = factory
	list := make([]*Actor, 0, len(f.actorsBySerial))
	serials := make([]string, 0, len(f.actorsBySerial))
	for serial, a := range f.actorsBySerial {
		list = append(list, a)
		serials = append(serials, serial)
	}
	f.mu.Unlock()
	for i, a := range list {
		var rf RepublishFunc
		if factory != nil {
			rf = factory(serials[i])
		}
		a.enqueue(evSetRepublish{rf})
	}
}

// SetSendCommandFunc installs the send_command hook on the fleet and every
// existing actor.
func (f *Fleet) SetSendCommandFunc(fn SendCommandFunc) {
	f.mu.Lock()
	f.sendCommandFn = fn
	list := make([]*Actor, 0, len(f.actorsBySerial))
	for _, a := range f.actorsBySerial {
		list = append(list, a)
	}
	f.mu.Unlock()
	for _, a := range list {
		a.enqueue(evSetSendCommand{fn})
	}
}

// Submit routes a command to the actor for serial; an unknown serial is
// offline.
func (f *Fleet) Submit(ctx context.Context, serial string, command Command) error {
	f.mu.Lock()
	a := f.actorsBySerial[serial]
	f.mu.Unlock()
	if a == nil {
		return errOffline
	}
	return a.Submit(ctx, command)
}

func (f *Fleet) SessionReady(e session.ReadyEvent) {
	f.mu.Lock()
	a := f.actors[e.JID]
	if a == nil {
		republish := f.republish
		if f.republishFactory != nil {
			republish = f.republishFactory(e.Serial)
		}
		a = NewActor(f.ctx, e.JID, e.Serial, f.controllerJID, republish, f.trace, f.diag, f.clock)
		f.actors[e.JID] = a
		f.actorsBySerial[e.Serial] = a
		a.enqueue(evSetSendCommand{f.sendCommandFn})
	}
	f.mu.Unlock()
	a.SessionReady(e)
}

func (f *Fleet) AnnounceOK(e session.ReadyEvent) {
	if a, ok := f.Actor(e.JID); ok {
		a.AnnounceOK(e)
	}
}

func (f *Fleet) SessionDown(e session.DownEvent) {
	if a, ok := f.Actor(e.JID); ok {
		a.SessionDown(e)
	}
}

func (f *Fleet) Stanza(e session.StanzaEvent) {
	if a, ok := f.Actor(e.JID); ok {
		a.Stanza(e)
	}
}