feat(cloud): FTW Cloud C4: the box's notifications as signed events

.changeset/ftw-cloud-events.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+With FTW Cloud on and events in the grant, the box sends its notifications as signed events for the owners' phones; names never leave the house. A restart the box chooses, an update among them, tells FTW Cloud so it holds back "box out of reach". A new notification, "The car may not be ready", tells once per plug-in when the box knows the car's level and the plan cannot reach the goal by the ready time. A car that reports nothing itself gets "Car charged" in kWh once the box's estimate from a level someone set reaches the target. "Car plugged in" now waits 90 seconds and is sent only when the box needs the level or a goal. Each notification transport now has its own 10-second limit, so one that hangs no longer delays the others.

contract/push-catalogue.yaml

@@ -19,6 +19,12 @@ events:
   - kind: charging.session_complete
     title: Car charged
     body: "{kwh} kWh delivered — ready to go."
+  - kind: charging.goal_at_risk
+    # Only when the box knows the car's level, from the car or from someone
+    # who set it, and the plan cannot reach the goal by the ready time. An
+    # assumed level would make this a guess. Once per plug-in.
+    title: The car may not be ready
+    body: "It will not reach {target} % by {by} at this pace."
   - kind: charging.interrupted
     # Only after the hysteresis window the box applies: a charge that had
     # been running steadily and then stopped without finishing. A flapping

go/cmd/ftw/cloud_link.go

@@ -337,6 +337,20 @@ func startCloudLink(
 	return link
 }
 
+// cloudByeWait bounds how long shutdown waits for the bye to leave.
+const cloudByeWait = 2 * time.Second
+
+// cloudEventQueue turns a rendered notification into an event for the
+// uplink, which cuts the names out before it signs.
+func cloudEventQueue(link *cloudlink.Link) func(notifications.Message) bool {
+	return func(m notifications.Message) bool {
+		return link.QueueEvent(cloudlink.EventIn{
+			Kind: m.Kind, Priority: m.Priority, Args: m.Args,
+			Driver: m.Device, Loadpoint: m.LoadpointID, At: m.At,
+		})
+	}
+}
+
 // cloudForAPI hands the link to the API, or nothing: a typed nil in an
 // interface would pass the routes' nil check and panic on first use.
 func cloudForAPI(link *cloudlink.Link) api.CloudGrant {

go/cmd/ftw/main.go

@@ -68,6 +68,7 @@ import (
 	"github.com/srcfl/ftw/go/internal/selfupdate"
 	"github.com/srcfl/ftw/go/internal/state"
 	"github.com/srcfl/ftw/go/internal/telemetry"
+	"github.com/srcfl/ftw/go/protocol/cloudwire"
 )
 
 // Version gets injected at build time via -ldflags. Defaults to "dev" for
@@ -2657,6 +2658,20 @@ func main() {
 		notifSvc.AddPublisher(webPush)
 		deps.WebPush = webPush
 	}
+	// FTW Cloud carries the catalogue's events to the owners' phones,
+	// signed, when the grant includes them. The publisher only queues.
+	if cloudLink != nil {
+		notifSvc.AddPublisher(notifications.CloudPublisher{Queue: cloudEventQueue(cloudLink)})
+	}
+	notifSvc.SetChargingFacts(func(id string) (notifications.ChargingFacts, bool) {
+		lp, ok := lpMgr.State(id)
+		if !ok {
+			return notifications.ChargingFacts{}, false
+		}
+		return notifications.ChargingFacts{
+			Plugged: lp.PluggedIn, LevelKnown: lp.PluggedIn && lp.SoCSource != "assumed", HasGoal: lp.TargetSoC > 0,
+		}, true
+	})
 	if appUplink != nil {
 		deps.PushResync = appUplink.ResyncDeadman
 	}
@@ -2773,6 +2788,7 @@ func main() {
 		select {
 		case <-sigc:
 			slog.Info("shutting down")
+			cloudLink.Bye(cloudwire.ByeShutdown, cloudByeWait)
 			markNativeCleanExit(nativeRoot)
 			flushHistoryOnStop(st)
 			if err := st.RecordEvent("shutdown"); err != nil {
@@ -2785,6 +2801,9 @@ func main() {
 			} else {
 				slog.Info("restart requested via API — exiting cleanly so the supervisor brings us back")
 			}
+			// A restart the box chose, an update among them: FTW Cloud holds
+			// back "box out of reach" until the box is back.
+			cloudLink.Bye(cloudwire.ByeUpdate, cloudByeWait)
 			markNativeCleanExit(nativeRoot)
 			flushHistoryOnStop(st)
 			if err := st.RecordEvent("restart"); err != nil {
@@ -3195,6 +3214,15 @@ func main() {
 				}
 			}
 
+			// A car whose plan cannot make its goal: the loadpoint manager
+			// tells the household once per plug-in, from a plan that is
+			// neither being remade nor out of date.
+			if mpcSvc != nil {
+				if snap := mpcSvc.PlanSnapshot(); snap.Plan != nil && !snap.Pending && !snap.Outdated {
+					lpMgr.ObservePlanShortfall(snap.Plan.LoadpointShortfallWh, tickNow)
+				}
+			}
+
 			// First-vehicle-SoC replan trigger: as soon as any
 			// DerVehicle driver is online and reporting SoC, redo the
 			// plan once with measured-truth instead of the pluginSoC

go/cmd/sim-ocpp/main.go

@@ -121,6 +121,17 @@ func main() {
 			}
 			w.WriteHeader(204)
 		})
+		for path, stalled := range map[string]bool{"stall": true, "resume": false} {
+			mux.HandleFunc("POST /charger/{id}/"+path, func(w http.ResponseWriter, r *http.Request) {
+				sim := findSim(sims, r.PathValue("id"))
+				if sim == nil {
+					http.NotFound(w, r)
+					return
+				}
+				sim.Stall(stalled)
+				w.WriteHeader(204)
+			})
+		}
 		go func() {
 			slog.Info("control listening", "addr", *control)
 			if err := http.ListenAndServe(*control, mux); err != nil {

go/cmd/sim-ocpp/ocppcp/physics.go

@@ -19,6 +19,9 @@ type Physics struct {
 	DrawA    float64
 	EnergyWh float64
 	Plugged  bool
+	// Stalled is a charger that stops delivering while the car still wants
+	// energy, as a tripped relay or a cloud-side fault does.
+	Stalled bool
 }
 
 func newPhysics(m Model, tauS float64) Physics {
@@ -39,7 +42,7 @@ func (p Physics) PowerW() float64 {
 }
 
 func (p Physics) targetA() float64 {
-	if !p.Plugged || p.LimitA < minChargeAmps {
+	if !p.Plugged || p.Stalled || p.LimitA < minChargeAmps {
 		return 0
 	}
 	if p.LimitA > p.MaxAmps {

go/cmd/sim-ocpp/ocppcp/sim.go

@@ -169,6 +169,13 @@ func (s *Sim) Unplug() error {
 	return s.statusPlugged(false)
 }
 
+// Stall stops delivery with the car still asking, or lets it go on.
+func (s *Sim) Stall(stalled bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.physics.Stalled = stalled
+}
+
 // Tick advances physics. Tests call this with a positive dt and TauS=0 to settle.
 func (s *Sim) Tick(dt time.Duration) {
 	s.mu.Lock()

go/internal/cloudlink/conn.go

@@ -69,7 +69,7 @@ func (l *Link) once(ctx context.Context, snap grantSnapshot) (time.Duration, err
 		l.attempt = 0
 	}
 	l.mu.Unlock()
-	if errors.Is(err, errPurged) {
+	if errors.Is(err, errPurged) || errors.Is(err, errClosed) {
 		return 0, nil
 	}
 	return l.backoff(), err

go/internal/cloudlink/events.go

@@ -0,0 +1,196 @@
+package cloudlink
+
+import (
+	"encoding/base64"
+	"errors"
+	"time"
+
+	"github.com/srcfl/ftw/go/protocol"
+	"github.com/srcfl/ftw/go/protocol/cloudwire"
+)
+
+// Events and the last word.
+//
+// The box's notifications reach FTW Cloud as signed events, which the
+// service wraps unchanged in an alert for the owners' phones. QueueEvent only
+// queues; the writer signs and sends on its own goroutine, ahead of anything
+// else due on that tick. An event waits at most cloudwire.AlertMaxAgeMs for
+// the uplink: older, a phone would show it without its text and buttons, so
+// it is dropped instead. A reconnect sends the events still waiting and
+// replays none that left.
+
+// EventIn is one notification the box raised, before ids are cut.
+type EventIn struct {
+	Kind string
+	// Priority is the rule's, 1 to 5. It becomes the event's severity.
+	Priority int
+	// Args are the catalogue's placeholders. Names are never sent: the
+	// link drops the name placeholders and sends the device's opaque id.
+	Args map[string]string
+	// Driver and Loadpoint are the box's own names for what the event
+	// concerns, empty when it concerns neither.
+	Driver    string
+	Loadpoint string
+	At        time.Time
+}
+
+const (
+	// eventRing is how many events wait for the uplink. When it is full
+	// the oldest goes, and the drop is counted.
+	eventRing = 32
+	// eventIDBytes makes a 12-character event id.
+	eventIDBytes = 9
+)
+
+// nameArgs are catalogue placeholders that hold a name. The phone fills
+// them from the table it holds from the box.
+var nameArgs = map[string]bool{"name": true}
+
+// errClosed ends the uplink after its bye, as the box shuts down.
+var errClosed = errors.New("cloudlink: the box is shutting down")
+
+// QueueEvent keeps an event for the uplink to sign and send. It never
+// blocks. False means FTW Cloud will not carry it: off, being revoked, or a
+// grant that leaves events out.
+func (l *Link) QueueEvent(e EventIn) bool {
+	l.mu.Lock()
+	g := l.grant
+	if g == nil || g.Revoked || !g.has(protocol.ScopeEventsRead) || l.closing {
+		l.mu.Unlock()
+		return false
+	}
+	if e.At.IsZero() {
+		e.At = time.Now()
+	}
+	if len(l.events) >= eventRing {
+		l.events = l.events[1:]
+		l.eventsDropped++
+	}
+	l.events = append(l.events, e)
+	l.mu.Unlock()
+	l.signal()
+	return true
+}
+
+// sendEvents signs and sends what waits, oldest first.
+func (w *writer) sendEvents(snap grantSnapshot, now time.Time) error {
+	l := w.link
+	if !snap.allowed(cloudwire.MsgEvent) {
+		return nil
+	}
+	l.mu.Lock()
+	queued := l.events
+	l.events = nil
+	l.mu.Unlock()
+	for i, e := range queued {
+		if now.Sub(e.At) > cloudwire.AlertMaxAgeMs*time.Millisecond {
+			l.mu.Lock()
+			l.eventsDropped++
+			l.mu.Unlock()
+			continue
+		}
+		ev, err := l.projectEvent(snap, e)
+		if err != nil {
+			return err
+		}
+		if _, err := w.sendBody(snap, cloudwire.MsgEvent, &ev, true); err != nil {
+			// The one that failed may have left; the rest have not.
+			l.mu.Lock()
+			l.events = append(queued[i+1:len(queued):len(queued)], l.events...)
+			l.mu.Unlock()
+			return err
+		}
+	}
+	return nil
+}
+
+// projectEvent cuts opaque ids and drops names.
+func (l *Link) projectEvent(snap grantSnapshot, e EventIn) (cloudwire.Event, error) {
+	id, err := randomBytes(eventIDBytes)
+	if err != nil {
+		return cloudwire.Event{}, err
+	}
+	gk := snap.g.GrantKey
+	ev := cloudwire.Event{
+		EventID: base64.RawURLEncoding.EncodeToString(id), Kind: e.Kind,
+		Severity: severity(e.Priority), Args: map[string]string{},
+	}
+	for k, v := range e.Args {
+		if !nameArgs[k] {
+			ev.Args[k] = v
+		}
+	}
+	if e.Driver != "" {
+		ev.Src = cloudwire.SourceID(gk, e.Driver)
+	}
+	if e.Loadpoint != "" {
+		ev.LP = cloudwire.LoadpointID(gk, e.Loadpoint)
+	}
+	if l.clockSynced() {
+		ts := e.At.UnixMilli()
+		ev.Ts = &ts
+	}
+	return ev, nil
+}
+
+func severity(priority int) string {
+	switch {
+	case priority >= 5:
+		return cloudwire.SeverityAlarm
+	case priority == 4:
+		return cloudwire.SeverityWarning
+	default:
+		return cloudwire.SeverityInfo
+	}
+}
+
+// Bye tells the service why the box is going and stops the uplink for good.
+// why is cloudwire.ByeUpdate for a restart the box chose, such as an update,
+// which holds back box.unreachable for an hour; ByeShutdown otherwise. It
+// waits at most wait for the bye to leave, and returns at once when the
+// uplink is down.
+func (l *Link) Bye(why string, wait time.Duration) {
+	if l == nil {
+		return
+	}
+	l.mu.Lock()
+	if l.closing {
+		l.mu.Unlock()
+		return
+	}
+	l.closing, l.byeWhy = true, why
+	done := make(chan struct{})
+	l.byeDone = done
+	connected := l.connected
+	l.mu.Unlock()
+	l.signal()
+	if !connected {
+		return
+	}
+	select {
+	case <-done:
+	case <-time.After(wait):
+	}
+}
+
+// closeIfAsked sends the bye Bye asked for and ends the connection.
+func (w *writer) closeIfAsked(snap grantSnapshot) error {
+	l := w.link
+	l.mu.Lock()
+	closing, why, done := l.closing, l.byeWhy, l.byeDone
+	l.mu.Unlock()
+	if !closing {
+		return nil
+	}
+	err := w.bye(snap, why, false)
+	l.mu.Lock()
+	l.byeDone = nil
+	l.mu.Unlock()
+	if done != nil {
+		close(done)
+	}
+	if err != nil {
+		return err
+	}
+	return errClosed
+}