One schedule for every charger, and a clock on the server to keep it
The charger's own cloud schedule is one window inside one box: charge between these hours, every day, and that is the whole vocabulary. A third tab on Charging holds a list instead — each line an action, a time, the days it repeats on and the chargers it acts on — and one list covers the whole account rather than each charger hiding its own. The clock is the server's. A schedule that only fires while a tab is open is a reminder, so a ticker sweeps every enabled task and fires whichever minute has come. It sends by handing a synthesised request to the same control endpoint the page's buttons use, so a scheduled command goes through the same cascade, ownership gate, rate limit and audit trail — what the owner cannot press by hand, the scheduler cannot send for them. A task names its chargers, or names none, which means all of them and keeps meaning that for a charger imported next year. Times are stored as a wall clock plus the zone they were written in, so 23:00 stays 23:00 wherever the server sits. One action per task: a charging window is the two tasks that open and close it, which is how it is read back, edited and switched off. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
tajniak81
co-authored by
Claude Opus 5
Claude Opus 5
parent
8e4c22cfcf
commit
5a4515978f
@@ -0,0 +1,398 @@
|
||||
package api
|
||||
|
||||
// The home-charger scheduler: the user's own list of charging tasks.
|
||||
//
|
||||
// The charger's own cloud schedule can say one thing — "charge between these
|
||||
// hours" — and it says it inside one charger. This is a list: as many tasks as
|
||||
// the owner likes, each with its own action, time, days and set of chargers, and
|
||||
// one list covering every charger on the account rather than a schedule hidden
|
||||
// inside each box.
|
||||
//
|
||||
// GET /api/charging-tasks — the caller's tasks, earliest first
|
||||
// POST /api/charging-tasks — write one
|
||||
// PATCH /api/charging-tasks/{id} — edit one (any subset of its fields)
|
||||
// DELETE /api/charging-tasks/{id} — forget one
|
||||
// POST /api/charging-tasks/{id}/run — fire it now, without waiting for its time
|
||||
//
|
||||
// A task belongs to the person, like the chargers it acts on, so there is no
|
||||
// sharing here: everyone sees their own list only. Firing one is the business of
|
||||
// chargingtasks_run.go, which is also what the ticker calls.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"drivervault/apiserver/internal/models"
|
||||
)
|
||||
|
||||
// The actions a task may carry. Each is a control command the charger already
|
||||
// understands; the scheduler adds no verbs of its own, so a task can only ask
|
||||
// for something the buttons on the Charging page can ask for by hand.
|
||||
var taskActions = map[string]bool{
|
||||
"start": true,
|
||||
"stop": true,
|
||||
"limit": true,
|
||||
"boost": true,
|
||||
}
|
||||
|
||||
// chargingTaskRecord is the PocketBase-facing shape of a task.
|
||||
type chargingTaskRecord struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Chargers []string `json:"chargers"`
|
||||
Action string `json:"action"`
|
||||
Amps float64 `json:"amps"`
|
||||
Time string `json:"time"`
|
||||
Zone string `json:"zone"`
|
||||
Days []int `json:"days"`
|
||||
Enabled bool `json:"enabled"`
|
||||
LastRun string `json:"last_run"`
|
||||
LastResult string `json:"last_result"`
|
||||
Owner string `json:"owner"`
|
||||
Created string `json:"created"`
|
||||
}
|
||||
|
||||
func (rec chargingTaskRecord) toModel() models.ChargingTask {
|
||||
// The two lists are never null on the wire: the page iterates them, and a
|
||||
// null would make "every charger" and "every day" read as an error there.
|
||||
chargers := rec.Chargers
|
||||
if chargers == nil {
|
||||
chargers = []string{}
|
||||
}
|
||||
days := rec.Days
|
||||
if days == nil {
|
||||
days = []int{}
|
||||
}
|
||||
return models.ChargingTask{
|
||||
ID: rec.ID,
|
||||
Name: rec.Name,
|
||||
Chargers: chargers,
|
||||
Action: rec.Action,
|
||||
Amps: rec.Amps,
|
||||
Time: rec.Time,
|
||||
Zone: rec.Zone,
|
||||
Days: days,
|
||||
Enabled: rec.Enabled,
|
||||
LastRun: rec.LastRun,
|
||||
LastResult: rec.LastResult,
|
||||
Created: rec.Created,
|
||||
}
|
||||
}
|
||||
|
||||
// chargingTaskBody is the write shape. Every field is a pointer so a PATCH can
|
||||
// name one of them without the rest being read as "clear these" — a task
|
||||
// switched off from the list must not lose its days on the way.
|
||||
type chargingTaskBody struct {
|
||||
Name *string `json:"name"`
|
||||
Chargers *[]string `json:"chargers"`
|
||||
Action *string `json:"action"`
|
||||
Amps *float64 `json:"amps"`
|
||||
Time *string `json:"time"`
|
||||
Zone *string `json:"zone"`
|
||||
Days *[]int `json:"days"`
|
||||
Enabled *bool `json:"enabled"`
|
||||
}
|
||||
|
||||
// parseHHMM splits a 24-hour clock time into its two numbers. "H:MM" is accepted
|
||||
// as well as "HH:MM"; anything else is not a time of day.
|
||||
func parseHHMM(raw string) (int, int, bool) {
|
||||
hh, mm, found := strings.Cut(strings.TrimSpace(raw), ":")
|
||||
if !found || len(hh) == 0 || len(hh) > 2 || len(mm) != 2 {
|
||||
return 0, 0, false
|
||||
}
|
||||
h, err := strconv.Atoi(hh)
|
||||
if err != nil {
|
||||
return 0, 0, false
|
||||
}
|
||||
m, err := strconv.Atoi(mm)
|
||||
if err != nil {
|
||||
return 0, 0, false
|
||||
}
|
||||
if h < 0 || h > 23 || m < 0 || m > 59 {
|
||||
return 0, 0, false
|
||||
}
|
||||
return h, m, true
|
||||
}
|
||||
|
||||
// normalizeTaskTime returns the padded 24-hour form of a time of day, or "" for
|
||||
// anything that is not one.
|
||||
func normalizeTaskTime(raw string) string {
|
||||
h, m, ok := parseHHMM(raw)
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
return fmt.Sprintf("%02d:%02d", h, m)
|
||||
}
|
||||
|
||||
// normalizeDays drops anything that is not a weekday number and sorts what is
|
||||
// left, so the stored list reads the same however the client sent it. Empty
|
||||
// stays empty, which means every day.
|
||||
func normalizeDays(days []int) []int {
|
||||
seen := [7]bool{}
|
||||
for _, d := range days {
|
||||
if d >= 0 && d <= 6 {
|
||||
seen[d] = true
|
||||
}
|
||||
}
|
||||
out := []int{}
|
||||
for d, ok := range seen {
|
||||
if ok {
|
||||
out = append(out, d)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// taskPayload turns a write body into the PocketBase fields it names, validating
|
||||
// as it goes. `full` demands the fields a task cannot exist without, so a POST is
|
||||
// checked as a whole and a PATCH only where it speaks.
|
||||
func taskPayload(body chargingTaskBody, full bool) (map[string]any, error) {
|
||||
payload := map[string]any{}
|
||||
|
||||
if body.Name != nil {
|
||||
name := strings.TrimSpace(*body.Name)
|
||||
if name == "" {
|
||||
return nil, fmt.Errorf("name is required")
|
||||
}
|
||||
payload["name"] = name
|
||||
} else if full {
|
||||
return nil, fmt.Errorf("name is required")
|
||||
}
|
||||
|
||||
if body.Action != nil {
|
||||
action := strings.TrimSpace(*body.Action)
|
||||
if !taskActions[action] {
|
||||
return nil, fmt.Errorf("unknown action: %s", action)
|
||||
}
|
||||
payload["action"] = action
|
||||
} else if full {
|
||||
return nil, fmt.Errorf("action is required")
|
||||
}
|
||||
|
||||
if body.Time != nil {
|
||||
at := normalizeTaskTime(*body.Time)
|
||||
if at == "" {
|
||||
return nil, fmt.Errorf("time must be a 24-hour clock time, e.g. 23:00")
|
||||
}
|
||||
payload["time"] = at
|
||||
} else if full {
|
||||
return nil, fmt.Errorf("time is required")
|
||||
}
|
||||
|
||||
// The ceiling only means anything to "limit", but it is stored whatever the
|
||||
// action is: switching a task to "limit" and back should not lose the amps
|
||||
// that were typed. 6 A is the charger's own floor — below it the box pauses
|
||||
// rather than charging slowly — and its ratings top out at 32 A.
|
||||
if body.Amps != nil {
|
||||
amps := *body.Amps
|
||||
if amps != 0 && (amps < 6 || amps > 32) {
|
||||
return nil, fmt.Errorf("the current limit must be between 6 and 32 A")
|
||||
}
|
||||
payload["amps"] = amps
|
||||
}
|
||||
|
||||
if body.Chargers != nil {
|
||||
ids := []string{}
|
||||
for _, id := range *body.Chargers {
|
||||
if id = strings.TrimSpace(id); id != "" {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
}
|
||||
payload["chargers"] = ids
|
||||
}
|
||||
if body.Days != nil {
|
||||
payload["days"] = normalizeDays(*body.Days)
|
||||
}
|
||||
if body.Zone != nil {
|
||||
payload["zone"] = strings.TrimSpace(*body.Zone)
|
||||
}
|
||||
if body.Enabled != nil {
|
||||
payload["enabled"] = *body.Enabled
|
||||
} else if full {
|
||||
// A task written without saying otherwise is on: nobody fills in a
|
||||
// schedule in order to leave it switched off.
|
||||
payload["enabled"] = true
|
||||
}
|
||||
return payload, nil
|
||||
}
|
||||
|
||||
// checkLimitAmps refuses a task that ends up a "limit" with nothing to limit to,
|
||||
// which would fire and ask the charger for 0 A. Both halves are read from what
|
||||
// the write leaves behind — the payload where it speaks, the stored record where
|
||||
// it does not — so a PATCH that names only one of them is checked against the
|
||||
// other rather than against nothing.
|
||||
func checkLimitAmps(payload map[string]any, stored chargingTaskRecord) error {
|
||||
action := stored.Action
|
||||
if v, ok := payload["action"].(string); ok {
|
||||
action = v
|
||||
}
|
||||
amps := stored.Amps
|
||||
if v, ok := payload["amps"].(float64); ok {
|
||||
amps = v
|
||||
}
|
||||
if action == "limit" && amps <= 0 {
|
||||
return fmt.Errorf("a limit task needs the amps to limit to")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// listChargingTasks returns the caller's own tasks, earliest in the day first —
|
||||
// the order the day runs in, which is the order a schedule is read in.
|
||||
func (s *Server) listChargingTasks(w http.ResponseWriter, r *http.Request) {
|
||||
me := s.currentUserID(r)
|
||||
if me == "" {
|
||||
writeError(w, http.StatusUnauthorized, "not authenticated")
|
||||
return
|
||||
}
|
||||
query := func(sort string) url.Values {
|
||||
q := url.Values{
|
||||
"filter": {fmt.Sprintf("owner='%s'", me)},
|
||||
"perPage": {"200"},
|
||||
}
|
||||
if sort != "" {
|
||||
q.Set("sort", sort)
|
||||
}
|
||||
return q
|
||||
}
|
||||
res, err := s.pb.List(r.Context(), colChargingTasks, query("time"))
|
||||
if err != nil {
|
||||
// Same fallback the charger list makes: an order is a nicety, the list is
|
||||
// not, so a collection that predates the sort field still answers.
|
||||
res, err = s.pb.List(r.Context(), colChargingTasks, query(""))
|
||||
}
|
||||
if err != nil {
|
||||
writePBError(w, err)
|
||||
return
|
||||
}
|
||||
var recs []chargingTaskRecord
|
||||
if err := json.Unmarshal(res.Items, &recs); err != nil {
|
||||
writeError(w, http.StatusInternalServerError, err.Error())
|
||||
return
|
||||
}
|
||||
out := make([]models.ChargingTask, 0, len(recs))
|
||||
for _, rec := range recs {
|
||||
out = append(out, rec.toModel())
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"tasks": out})
|
||||
}
|
||||
|
||||
// ownedChargingTask fetches one task and checks it is the caller's. It writes the
|
||||
// error response itself and returns ok=false when it is not.
|
||||
func (s *Server) ownedChargingTask(w http.ResponseWriter, r *http.Request) (chargingTaskRecord, bool) {
|
||||
var rec chargingTaskRecord
|
||||
id := strings.TrimSpace(r.PathValue("id"))
|
||||
if id == "" {
|
||||
writeError(w, http.StatusBadRequest, "task id is required")
|
||||
return rec, false
|
||||
}
|
||||
if err := s.pb.GetOne(r.Context(), colChargingTasks, id, &rec); err != nil {
|
||||
writePBError(w, err)
|
||||
return rec, false
|
||||
}
|
||||
// Someone else's task is not found rather than forbidden, as a charger is:
|
||||
// whether a record exists is not this caller's business either.
|
||||
if rec.Owner != s.currentUserID(r) {
|
||||
writeError(w, http.StatusNotFound, "task not found")
|
||||
return rec, false
|
||||
}
|
||||
return rec, true
|
||||
}
|
||||
|
||||
func (s *Server) createChargingTask(w http.ResponseWriter, r *http.Request) {
|
||||
me := s.currentUserID(r)
|
||||
if me == "" {
|
||||
writeError(w, http.StatusUnauthorized, "not authenticated")
|
||||
return
|
||||
}
|
||||
var body chargingTaskBody
|
||||
if err := decodeJSON(r, &body); err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
payload, err := taskPayload(body, true)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
if err := checkLimitAmps(payload, chargingTaskRecord{}); err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
payload["owner"] = me
|
||||
var rec chargingTaskRecord
|
||||
if err := s.pb.Create(r.Context(), colChargingTasks, payload, &rec); err != nil {
|
||||
writePBError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusCreated, map[string]any{"task": rec.toModel()})
|
||||
}
|
||||
|
||||
func (s *Server) updateChargingTask(w http.ResponseWriter, r *http.Request) {
|
||||
rec, ok := s.ownedChargingTask(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var body chargingTaskBody
|
||||
if err := decodeJSON(r, &body); err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
payload, err := taskPayload(body, false)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
if err := checkLimitAmps(payload, rec); err != nil {
|
||||
writeError(w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
if len(payload) == 0 {
|
||||
writeJSON(w, http.StatusOK, map[string]any{"task": rec.toModel()})
|
||||
return
|
||||
}
|
||||
var updated chargingTaskRecord
|
||||
if err := s.pb.Update(r.Context(), colChargingTasks, rec.ID, payload, &updated); err != nil {
|
||||
writePBError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"task": updated.toModel()})
|
||||
}
|
||||
|
||||
func (s *Server) deleteChargingTask(w http.ResponseWriter, r *http.Request) {
|
||||
rec, ok := s.ownedChargingTask(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if err := s.pb.Delete(r.Context(), colChargingTasks, rec.ID); err != nil {
|
||||
writePBError(w, err)
|
||||
return
|
||||
}
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
|
||||
// runChargingTaskNow fires a task on demand — the "Run now" button beside it. It
|
||||
// takes the same path the ticker does, so what comes back is exactly what the
|
||||
// task will do at its own time, errors included. A switched-off task still runs
|
||||
// from here: the switch says whether the clock fires it, not whether the button
|
||||
// does.
|
||||
func (s *Server) runChargingTaskNow(w http.ResponseWriter, r *http.Request) {
|
||||
rec, ok := s.ownedChargingTask(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
who := caller(r)
|
||||
if who == nil {
|
||||
writeError(w, http.StatusUnauthorized, "not authenticated")
|
||||
return
|
||||
}
|
||||
results := s.fireChargingTask(r.Context(), who, rec)
|
||||
s.recordTaskRun(r.Context(), rec, results)
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"results": results,
|
||||
"summary": summarizeTaskRun(results),
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,347 @@
|
||||
package api
|
||||
|
||||
// Firing the scheduler's tasks: the ticker that watches the clock, and the one
|
||||
// path a task takes whichever set it off.
|
||||
//
|
||||
// The scheduler is only worth having if it fires with nobody looking, so the
|
||||
// clock is watched here rather than in the browser — a schedule that needs the
|
||||
// page open is a reminder, not a schedule. The ticker sweeps every enabled task
|
||||
// on the server, of every user, and fires whichever ones' minute has come.
|
||||
//
|
||||
// A task acts by sending exactly the control command the Charging page's own
|
||||
// buttons send: the request is built here and handed to the same handler, so it
|
||||
// goes through the same cascade, the same ownership gate, the same rate limit
|
||||
// and the same audit trail. Nothing about a scheduled command is privileged —
|
||||
// what the owner cannot press by hand, the scheduler cannot send for them.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
_ "time/tzdata" // tasks are timed in the user's own zone; hosts without a zone database are common on Windows
|
||||
)
|
||||
|
||||
// How often the clock is looked at. Tasks are timed to the minute, so a sweep
|
||||
// twice a minute is enough to land in every minute without the drift a
|
||||
// once-a-minute ticker accumulates.
|
||||
const taskTick = 30 * time.Second
|
||||
|
||||
// How long one sweep may take. Generous rather than tight: the cloud transport
|
||||
// alone allows 45 seconds for a single command, and a task can name several
|
||||
// chargers. A sweep that overran would not overlap the next one — the ticker
|
||||
// drops a tick nobody is waiting on — but it would abandon chargers halfway
|
||||
// down a task's list, which is the worst of both.
|
||||
const taskSweepTimeout = 5 * time.Minute
|
||||
|
||||
// taskRunResult is what happened to one charger in one firing.
|
||||
type taskRunResult struct {
|
||||
Charger string `json:"charger"` // home charger record id
|
||||
Name string `json:"name"`
|
||||
Serial string `json:"serial,omitempty"`
|
||||
Status string `json:"status,omitempty"` // the charger's own answer ("Accepted", …)
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// StartScheduler starts the ticker that fires due charging tasks. It is safe to
|
||||
// call before PocketBase is reachable: a sweep that cannot read the tasks simply
|
||||
// finds none and the next one tries again.
|
||||
//
|
||||
// One server runs one scheduler. Two API Servers pointed at the same database
|
||||
// would both sweep and both fire; the deployment is a single server (see the
|
||||
// README), and the last_run guard below narrows the window rather than closing
|
||||
// it.
|
||||
func (s *Server) StartScheduler() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
s.schedulerStop = cancel
|
||||
go func() {
|
||||
t := time.NewTicker(taskTick)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
s.sweepChargingTasks(ctx)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// sweepChargingTasks fires every task whose minute has come.
|
||||
func (s *Server) sweepChargingTasks(ctx context.Context) {
|
||||
if !s.pb.Configured() {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(ctx, taskSweepTimeout)
|
||||
defer cancel()
|
||||
|
||||
res, err := s.pb.List(ctx, colChargingTasks, url.Values{
|
||||
"filter": {"enabled=true"},
|
||||
"perPage": {"500"},
|
||||
})
|
||||
if err != nil {
|
||||
// A database that is not answering is not an error worth a line every
|
||||
// thirty seconds; the tasks are still there when it comes back.
|
||||
return
|
||||
}
|
||||
var recs []chargingTaskRecord
|
||||
if err := json.Unmarshal(res.Items, &recs); err != nil {
|
||||
return
|
||||
}
|
||||
now := time.Now()
|
||||
for _, rec := range recs {
|
||||
if !taskIsDue(rec, now) {
|
||||
continue
|
||||
}
|
||||
who, _, err := s.callerForUser(ctx, rec.Owner)
|
||||
if err != nil {
|
||||
log.Printf("scheduler: task %s (%s): owner unavailable: %v", rec.ID, rec.Name, err)
|
||||
continue
|
||||
}
|
||||
results := s.fireChargingTask(ctx, who, rec)
|
||||
s.recordTaskRun(ctx, rec, results)
|
||||
log.Printf("scheduler: task %s (%s) fired: %s", rec.ID, rec.Name, summarizeTaskRun(results))
|
||||
}
|
||||
}
|
||||
|
||||
// taskIsDue reports whether a task's minute has come and it has not already been
|
||||
// fired in it.
|
||||
//
|
||||
// The time is read in the task's own zone: the browser that wrote it said which,
|
||||
// and the server's clock is not the one the user set 23:00 by. A zone the host
|
||||
// has no database for falls back to the server's own rather than silently
|
||||
// shifting the schedule to UTC.
|
||||
//
|
||||
// A minute that passed while the server was down is not caught up afterwards. A
|
||||
// charging window that opened an hour ago is not a window anyone still wants
|
||||
// opened, and firing a backlog on boot would be the surprising half of the
|
||||
// choice.
|
||||
func taskIsDue(rec chargingTaskRecord, now time.Time) bool {
|
||||
h, m, ok := parseHHMM(rec.Time)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
loc := time.Local
|
||||
if rec.Zone != "" {
|
||||
if l, err := time.LoadLocation(rec.Zone); err == nil {
|
||||
loc = l
|
||||
}
|
||||
}
|
||||
local := now.In(loc)
|
||||
if local.Hour() != h || local.Minute() != m {
|
||||
return false
|
||||
}
|
||||
if len(rec.Days) > 0 && !containsDay(rec.Days, int(local.Weekday())) {
|
||||
return false
|
||||
}
|
||||
// The guard against firing twice: the sweep runs more often than once a
|
||||
// minute, so a task that has already run inside this minute is done.
|
||||
if last, err := time.Parse(time.RFC3339, rec.LastRun); err == nil {
|
||||
if !last.Before(local.Truncate(time.Minute)) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func containsDay(days []int, day int) bool {
|
||||
for _, d := range days {
|
||||
if d == day {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// fireChargingTask sends the task's command to each of its chargers and reports
|
||||
// what each one said. A task naming no chargers acts on every charger its owner
|
||||
// has — "all of them" is a standing wish, so a charger imported after the task
|
||||
// was written is covered by it too.
|
||||
func (s *Server) fireChargingTask(ctx context.Context, who *callerIdentity, rec chargingTaskRecord) []taskRunResult {
|
||||
chargers, err := s.ownerChargers(ctx, rec.Owner)
|
||||
if err != nil {
|
||||
return []taskRunResult{{Error: err.Error()}}
|
||||
}
|
||||
wanted := rec.Chargers
|
||||
results := []taskRunResult{}
|
||||
for _, c := range chargers {
|
||||
if len(wanted) > 0 && !containsID(wanted, c.ID) {
|
||||
continue
|
||||
}
|
||||
out := taskRunResult{Charger: c.ID, Name: c.Name, Serial: c.Serial}
|
||||
if c.Serial == "" {
|
||||
// Every transport addresses a charger by its serial, so one without
|
||||
// is not a charger anything can be sent to.
|
||||
out.Error = "this charger has no serial number, so no command can be addressed to it"
|
||||
results = append(results, out)
|
||||
continue
|
||||
}
|
||||
status, err := s.sendTaskAction(ctx, who, c.Serial, rec)
|
||||
if err != nil {
|
||||
out.Error = err.Error()
|
||||
} else {
|
||||
out.Status = status
|
||||
}
|
||||
results = append(results, out)
|
||||
}
|
||||
if len(results) == 0 {
|
||||
results = append(results, taskRunResult{Error: "this task names no charger that still exists"})
|
||||
}
|
||||
return results
|
||||
}
|
||||
|
||||
// ownerChargers reads one user's imported chargers. The scheduler reads them
|
||||
// itself rather than trusting the ids on the task: a charger removed from the
|
||||
// account should stop being acted on, whether or not the task was edited.
|
||||
func (s *Server) ownerChargers(ctx context.Context, owner string) ([]homeChargerRecord, error) {
|
||||
res, err := s.pb.List(ctx, colHomeChargers, url.Values{
|
||||
"filter": {fmt.Sprintf("owner='%s'", owner)},
|
||||
"perPage": {"200"},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var recs []homeChargerRecord
|
||||
if err := json.Unmarshal(res.Items, &recs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return recs, nil
|
||||
}
|
||||
|
||||
func containsID(ids []string, id string) bool {
|
||||
for _, x := range ids {
|
||||
if x == id {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// sendTaskAction sends one task's command to one charger, through the control
|
||||
// endpoint the page's own buttons use.
|
||||
//
|
||||
// The request is synthesised rather than the transports being called directly,
|
||||
// so a scheduled command cannot end up on a different footing from a pressed
|
||||
// one: the cascade, the ownership gate, the rate limit and the audit line are
|
||||
// the endpoint's, and there is no second copy of them here to drift.
|
||||
func (s *Server) sendTaskAction(ctx context.Context, who *callerIdentity, serial string, rec chargingTaskRecord) (string, error) {
|
||||
body := map[string]any{}
|
||||
if rec.Action == "limit" {
|
||||
body["amps"] = rec.Amps
|
||||
}
|
||||
raw, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
path := "/api/integrations/anker-solix/chargers/" + url.PathEscape(serial) + "/" + rec.Action
|
||||
req, err := http.NewRequestWithContext(
|
||||
context.WithValue(ctx, ctxCaller, who), http.MethodPost, path, bytes.NewReader(raw))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
// The endpoint reads its charger and its verb from the route, which nothing
|
||||
// matched here — this request never went through a mux.
|
||||
req.SetPathValue("sn", serial)
|
||||
req.SetPathValue("action", rec.Action)
|
||||
|
||||
rw := &captureWriter{header: http.Header{}}
|
||||
s.handleAnkerControlAction(rw, req)
|
||||
return rw.controlOutcome()
|
||||
}
|
||||
|
||||
// captureWriter is an http.ResponseWriter that keeps the response in memory,
|
||||
// so a handler can be called without a connection to write down.
|
||||
type captureWriter struct {
|
||||
status int
|
||||
header http.Header
|
||||
body bytes.Buffer
|
||||
}
|
||||
|
||||
func (c *captureWriter) Header() http.Header { return c.header }
|
||||
|
||||
func (c *captureWriter) WriteHeader(status int) {
|
||||
if c.status == 0 {
|
||||
c.status = status
|
||||
}
|
||||
}
|
||||
|
||||
func (c *captureWriter) Write(p []byte) (int, error) {
|
||||
if c.status == 0 {
|
||||
c.status = http.StatusOK
|
||||
}
|
||||
return c.body.Write(p)
|
||||
}
|
||||
|
||||
// controlOutcome reads a captured control response as the charger's answer or
|
||||
// the reason there wasn't one. The endpoint answers in one of two shapes — a
|
||||
// status on success, an error envelope otherwise — and both are read here rather
|
||||
// than the status code alone, because "why not" is the half worth keeping.
|
||||
func (c *captureWriter) controlOutcome() (string, error) {
|
||||
var out struct {
|
||||
Status string `json:"status"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
_ = json.Unmarshal(c.body.Bytes(), &out)
|
||||
if c.status >= 200 && c.status < 300 {
|
||||
if out.Status == "" {
|
||||
return "sent", nil
|
||||
}
|
||||
return out.Status, nil
|
||||
}
|
||||
if out.Error != "" {
|
||||
return "", fmt.Errorf("%s", out.Error)
|
||||
}
|
||||
return "", fmt.Errorf("the charger could not be reached (HTTP %d)", c.status)
|
||||
}
|
||||
|
||||
// summarizeTaskRun says how a firing went in one line — what the list shows
|
||||
// beside a task, and what goes in the log.
|
||||
func summarizeTaskRun(results []taskRunResult) string {
|
||||
ok, failed := 0, []string{}
|
||||
for _, r := range results {
|
||||
if r.Error == "" {
|
||||
ok++
|
||||
continue
|
||||
}
|
||||
name := r.Name
|
||||
if name == "" {
|
||||
name = r.Serial
|
||||
}
|
||||
if name != "" {
|
||||
failed = append(failed, name+": "+r.Error)
|
||||
} else {
|
||||
failed = append(failed, r.Error)
|
||||
}
|
||||
}
|
||||
sort.Strings(failed)
|
||||
switch {
|
||||
case len(failed) == 0:
|
||||
return fmt.Sprintf("%d of %d sent", ok, len(results))
|
||||
case ok == 0 && len(failed) == 1:
|
||||
return failed[0]
|
||||
default:
|
||||
return fmt.Sprintf("%d of %d sent — %s", ok, len(results), strings.Join(failed, "; "))
|
||||
}
|
||||
}
|
||||
|
||||
// recordTaskRun stamps a task with when it last fired and how it went. The stamp
|
||||
// is also the guard that keeps a task from firing twice inside its minute, so a
|
||||
// firing that could not be written down is worth a log line: without it the next
|
||||
// sweep would send the command again.
|
||||
func (s *Server) recordTaskRun(ctx context.Context, rec chargingTaskRecord, results []taskRunResult) {
|
||||
payload := map[string]any{
|
||||
"last_run": time.Now().UTC().Format(time.RFC3339),
|
||||
"last_result": summarizeTaskRun(results),
|
||||
}
|
||||
if err := s.pb.Update(ctx, colChargingTasks, rec.ID, payload, nil); err != nil {
|
||||
log.Printf("scheduler: task %s fired but its outcome could not be saved: %v", rec.ID, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,272 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The scheduler's clock is the part worth pinning down: it decides on its own,
|
||||
// with nobody watching, whether to send a car a command. A task that fires in
|
||||
// the wrong hour is worse than one that does not fire at all, so the zone, the
|
||||
// weekday and the not-twice-in-one-minute guard are each checked here.
|
||||
|
||||
func TestTaskIsDue(t *testing.T) {
|
||||
warsaw, err := time.LoadLocation("Europe/Warsaw")
|
||||
if err != nil {
|
||||
t.Fatalf("Europe/Warsaw: %v", err)
|
||||
}
|
||||
// A Wednesday, 23:00 in Warsaw — 22:00 UTC.
|
||||
at2300 := time.Date(2026, 9, 2, 23, 0, 30, 0, warsaw)
|
||||
|
||||
base := chargingTaskRecord{Time: "23:00", Zone: "Europe/Warsaw", Enabled: true}
|
||||
|
||||
if !taskIsDue(base, at2300) {
|
||||
t.Error("a task set for 23:00 is not due at 23:00 in its own zone")
|
||||
}
|
||||
// The same instant, handed over as UTC. The zone on the task is what the
|
||||
// time is read in, so where the server thinks it is must not matter.
|
||||
if !taskIsDue(base, at2300.UTC()) {
|
||||
t.Error("the task stopped being due when the same instant arrived as UTC")
|
||||
}
|
||||
if taskIsDue(base, at2300.Add(time.Minute)) {
|
||||
t.Error("a task fired a minute after its time")
|
||||
}
|
||||
if taskIsDue(base, at2300.Add(-time.Minute)) {
|
||||
t.Error("a task fired a minute before its time")
|
||||
}
|
||||
|
||||
// Weekdays. 2026-09-02 is a Wednesday (3).
|
||||
weeknights := base
|
||||
weeknights.Days = []int{1, 2, 3, 4, 5}
|
||||
if !taskIsDue(weeknights, at2300) {
|
||||
t.Error("a weeknight task is not due on a Wednesday")
|
||||
}
|
||||
weekends := base
|
||||
weekends.Days = []int{0, 6}
|
||||
if taskIsDue(weekends, at2300) {
|
||||
t.Error("a weekend task fired on a Wednesday")
|
||||
}
|
||||
|
||||
// Fired already inside this minute: the sweep runs more than once a minute,
|
||||
// and the second sweep must not send the command again.
|
||||
fired := base
|
||||
fired.LastRun = at2300.UTC().Format(time.RFC3339)
|
||||
if taskIsDue(fired, at2300.Add(20*time.Second)) {
|
||||
t.Error("a task fired twice inside its own minute")
|
||||
}
|
||||
// Yesterday's firing is not this minute's.
|
||||
yesterday := base
|
||||
yesterday.LastRun = at2300.Add(-24 * time.Hour).UTC().Format(time.RFC3339)
|
||||
if !taskIsDue(yesterday, at2300) {
|
||||
t.Error("yesterday's run stopped today's from firing")
|
||||
}
|
||||
|
||||
// A time that is not a time of day fires nothing rather than firing at
|
||||
// midnight, which is what a zero hour and minute would have meant.
|
||||
broken := base
|
||||
broken.Time = "later"
|
||||
if taskIsDue(broken, at2300) {
|
||||
t.Error("a task with an unreadable time fired")
|
||||
}
|
||||
}
|
||||
|
||||
// A zone the host has no database for falls back to the server's own clock
|
||||
// rather than silently shifting the schedule to UTC.
|
||||
func TestTaskIsDueUnknownZone(t *testing.T) {
|
||||
now := time.Now()
|
||||
rec := chargingTaskRecord{
|
||||
Time: now.Format("15:04"),
|
||||
Zone: "Mars/Olympus_Mons",
|
||||
}
|
||||
if !taskIsDue(rec, now) {
|
||||
t.Error("a task in an unknown zone did not fall back to the server's clock")
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseHHMM(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
in string
|
||||
h, m int
|
||||
wantOK bool
|
||||
}{
|
||||
{in: "23:00", h: 23, m: 0, wantOK: true},
|
||||
{in: "7:05", h: 7, m: 5, wantOK: true},
|
||||
{in: " 00:00 ", h: 0, m: 0, wantOK: true},
|
||||
{in: "24:00"},
|
||||
{in: "12:60"},
|
||||
{in: "12:5"}, // a half-typed minute is not a minute
|
||||
{in: "123:00"}, // nor is a three-digit hour
|
||||
{in: "12"},
|
||||
{in: ""},
|
||||
{in: "ab:cd"},
|
||||
} {
|
||||
h, m, ok := parseHHMM(tc.in)
|
||||
if ok != tc.wantOK {
|
||||
t.Errorf("parseHHMM(%q) ok = %v, want %v", tc.in, ok, tc.wantOK)
|
||||
continue
|
||||
}
|
||||
if ok && (h != tc.h || m != tc.m) {
|
||||
t.Errorf("parseHHMM(%q) = %d:%d, want %d:%d", tc.in, h, m, tc.h, tc.m)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeDays(t *testing.T) {
|
||||
got := normalizeDays([]int{5, 1, 5, 9, -1, 0})
|
||||
want := []int{0, 1, 5}
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("normalizeDays = %v, want %v", got, want)
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Fatalf("normalizeDays = %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
// Empty means every day and stays empty rather than becoming all seven —
|
||||
// the two read the same today, but only one keeps meaning "every day".
|
||||
if len(normalizeDays(nil)) != 0 {
|
||||
t.Error("normalizeDays(nil) invented days")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskPayloadValidation(t *testing.T) {
|
||||
str := func(s string) *string { return &s }
|
||||
num := func(f float64) *float64 { return &f }
|
||||
|
||||
// A whole task, as a create sends it.
|
||||
full := chargingTaskBody{
|
||||
Name: str(" Night rate "),
|
||||
Action: str("limit"),
|
||||
Time: str("7:05"),
|
||||
Amps: num(10),
|
||||
}
|
||||
payload, err := taskPayload(full, true)
|
||||
if err != nil {
|
||||
t.Fatalf("taskPayload: %v", err)
|
||||
}
|
||||
if payload["name"] != "Night rate" {
|
||||
t.Errorf("name = %v, want it trimmed", payload["name"])
|
||||
}
|
||||
if payload["time"] != "07:05" {
|
||||
t.Errorf("time = %v, want the padded 07:05", payload["time"])
|
||||
}
|
||||
if payload["enabled"] != true {
|
||||
t.Error("a new task was written switched off; nobody fills in a schedule to leave it off")
|
||||
}
|
||||
|
||||
// The fields a task cannot exist without.
|
||||
for _, missing := range []chargingTaskBody{
|
||||
{Action: str("start"), Time: str("23:00")},
|
||||
{Name: str("x"), Time: str("23:00")},
|
||||
{Name: str("x"), Action: str("start")},
|
||||
{Name: str(" "), Action: str("start"), Time: str("23:00")},
|
||||
{Name: str("x"), Action: str("melt"), Time: str("23:00")},
|
||||
{Name: str("x"), Action: str("start"), Time: str("half past")},
|
||||
} {
|
||||
if _, err := taskPayload(missing, true); err == nil {
|
||||
t.Errorf("taskPayload accepted an incomplete task: %+v", missing)
|
||||
}
|
||||
}
|
||||
|
||||
// The charger's own floor and ceiling.
|
||||
for _, amps := range []float64{1, 5, 33} {
|
||||
if _, err := taskPayload(chargingTaskBody{Amps: num(amps)}, false); err == nil {
|
||||
t.Errorf("taskPayload accepted %g A, outside what the charger will take", amps)
|
||||
}
|
||||
}
|
||||
|
||||
// A partial write says only what it names, so a task switched off from the
|
||||
// list keeps its days and its time.
|
||||
on := true
|
||||
partial, err := taskPayload(chargingTaskBody{Enabled: &on}, false)
|
||||
if err != nil {
|
||||
t.Fatalf("taskPayload(partial): %v", err)
|
||||
}
|
||||
if len(partial) != 1 {
|
||||
t.Errorf("a switch-only write touched %v, want only enabled", partial)
|
||||
}
|
||||
}
|
||||
|
||||
// A "limit" with no ceiling would fire and ask the charger for 0 A. Both halves
|
||||
// are read from whatever the write leaves behind, so a PATCH naming one of them
|
||||
// is checked against the stored other.
|
||||
func TestCheckLimitAmps(t *testing.T) {
|
||||
stored := chargingTaskRecord{Action: "start", Amps: 0}
|
||||
if err := checkLimitAmps(map[string]any{"action": "limit"}, stored); err == nil {
|
||||
t.Error("a task switched to limit with no amps was accepted")
|
||||
}
|
||||
if err := checkLimitAmps(map[string]any{"action": "limit", "amps": 16.0}, stored); err != nil {
|
||||
t.Errorf("a complete limit task was refused: %v", err)
|
||||
}
|
||||
// The amps are already on the record; the PATCH only changes the action.
|
||||
hasAmps := chargingTaskRecord{Action: "start", Amps: 16}
|
||||
if err := checkLimitAmps(map[string]any{"action": "limit"}, hasAmps); err != nil {
|
||||
t.Errorf("a limit task with stored amps was refused: %v", err)
|
||||
}
|
||||
// And the other way round: clearing the amps on a task that limits.
|
||||
limits := chargingTaskRecord{Action: "limit", Amps: 16}
|
||||
if err := checkLimitAmps(map[string]any{"amps": 0.0}, limits); err == nil {
|
||||
t.Error("the amps were cleared off a limit task")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSummarizeTaskRun(t *testing.T) {
|
||||
all := summarizeTaskRun([]taskRunResult{
|
||||
{Name: "Home-DK", Status: "Accepted"},
|
||||
{Name: "Home-PL", Status: "Accepted"},
|
||||
})
|
||||
if all != "2 of 2 sent" {
|
||||
t.Errorf("summary = %q, want \"2 of 2 sent\"", all)
|
||||
}
|
||||
|
||||
// One charger failing is the case the list has to be able to show: the
|
||||
// summary names which, because "1 of 2 sent" alone is not actionable.
|
||||
some := summarizeTaskRun([]taskRunResult{
|
||||
{Name: "Home-DK", Status: "Accepted"},
|
||||
{Name: "Home-PL", Error: "charger is not connected"},
|
||||
})
|
||||
if !strings.Contains(some, "Home-PL") || !strings.Contains(some, "not connected") {
|
||||
t.Errorf("summary = %q, want it to name the charger that failed and why", some)
|
||||
}
|
||||
|
||||
// A single failure is its own reason — no count to read past.
|
||||
only := summarizeTaskRun([]taskRunResult{{Name: "Home-PL", Error: "control mode is off"}})
|
||||
if only != "Home-PL: control mode is off" {
|
||||
t.Errorf("summary = %q, want the bare reason", only)
|
||||
}
|
||||
}
|
||||
|
||||
// The runner reads a control response the same way whichever transport answered
|
||||
// it: a status when the charger took the command, the endpoint's own words when
|
||||
// it did not.
|
||||
func TestCaptureWriterControlOutcome(t *testing.T) {
|
||||
ok := &captureWriter{header: nil}
|
||||
ok.WriteHeader(200)
|
||||
_, _ = ok.Write([]byte(`{"status":"Accepted"}`))
|
||||
if got, err := ok.controlOutcome(); err != nil || got != "Accepted" {
|
||||
t.Errorf("controlOutcome = %q, %v; want Accepted", got, err)
|
||||
}
|
||||
|
||||
// A 2xx with nothing to read still means the command went.
|
||||
bare := &captureWriter{}
|
||||
bare.WriteHeader(204)
|
||||
if got, err := bare.controlOutcome(); err != nil || got != "sent" {
|
||||
t.Errorf("controlOutcome = %q, %v; want sent", got, err)
|
||||
}
|
||||
|
||||
bad := &captureWriter{}
|
||||
bad.WriteHeader(409)
|
||||
_, _ = bad.Write([]byte(`{"error":"charger is not connected to the control backend"}`))
|
||||
_, err := bad.controlOutcome()
|
||||
if err == nil || !strings.Contains(err.Error(), "not connected") {
|
||||
t.Errorf("controlOutcome err = %v, want the endpoint's own reason", err)
|
||||
}
|
||||
|
||||
// An error status with no envelope to read still has to say something.
|
||||
mute := &captureWriter{}
|
||||
mute.WriteHeader(502)
|
||||
if _, err := mute.controlOutcome(); err == nil {
|
||||
t.Error("a 502 with an empty body was read as a success")
|
||||
}
|
||||
}
|
||||
@@ -159,6 +159,7 @@ const (
|
||||
colDocuments = "car_documents"
|
||||
colReminders = "reminders"
|
||||
colHomeChargers = "home_chargers"
|
||||
colChargingTasks = "charging_tasks"
|
||||
colControlAudit = "control_audit"
|
||||
colAppSettings = "app_settings"
|
||||
)
|
||||
@@ -182,6 +183,10 @@ type Server struct {
|
||||
ocpp *ocpp.CSMS
|
||||
control *controlIndex // token -> owning user/charger for the /ocpp endpoint
|
||||
ctlRL *rateLimiter // per user+charger control-command rate limit
|
||||
|
||||
// schedulerStop ends the ticker that fires the home-charger scheduler's
|
||||
// tasks, started by StartScheduler (see chargingtasks_run.go).
|
||||
schedulerStop context.CancelFunc
|
||||
}
|
||||
|
||||
// New constructs a Server around an already-built PocketBase client.
|
||||
@@ -285,6 +290,9 @@ func (s *Server) Stop(ctx context.Context) {
|
||||
if s.pluginsStop != nil {
|
||||
s.pluginsStop()
|
||||
}
|
||||
if s.schedulerStop != nil {
|
||||
s.schedulerStop()
|
||||
}
|
||||
s.ocpp.Shutdown(ctx)
|
||||
s.plugins.Shutdown(ctx)
|
||||
}
|
||||
@@ -488,6 +496,14 @@ func (s *Server) Handler() http.Handler {
|
||||
mux.HandleFunc("PATCH /api/home-chargers/{id}", s.updateHomeCharger)
|
||||
mux.HandleFunc("DELETE /api/home-chargers/{id}", s.deleteHomeCharger)
|
||||
|
||||
// The home-charger scheduler: one list of charging tasks per user, covering
|
||||
// every charger they own. See chargingtasks.go.
|
||||
mux.HandleFunc("GET /api/charging-tasks", s.listChargingTasks)
|
||||
mux.HandleFunc("POST /api/charging-tasks", s.createChargingTask)
|
||||
mux.HandleFunc("PATCH /api/charging-tasks/{id}", s.updateChargingTask)
|
||||
mux.HandleFunc("DELETE /api/charging-tasks/{id}", s.deleteChargingTask)
|
||||
mux.HandleFunc("POST /api/charging-tasks/{id}/run", s.runChargingTaskNow)
|
||||
|
||||
// Cars + sharing.
|
||||
mux.HandleFunc("GET /api/cars", s.listCars)
|
||||
mux.HandleFunc("POST /api/cars", s.createCar)
|
||||
|
||||
Reference in New Issue
Block a user