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 "drivervault/apiserver/internal/models" ) // 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 { due := dueSteps(rec, now) if len(due) == 0 { 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 } // Two steps of one task timed to the same minute contradict each other, // but nothing stops somebody writing them, so both are sent in the order // the flow holds them rather than one silently winning. for _, step := range due { results := s.fireChargingSteps(ctx, who, rec, []models.ChargingStep{step}) s.recordTaskRun(ctx, rec, step, results) log.Printf("scheduler: task %s (%s) step %s %s: %s", rec.ID, rec.Name, step.Time, step.Action, summarizeTaskRun(results)) } } } // dueSteps returns the steps of a task whose minute has come — usually one, and // none at all for the rest of the day. // // The times are 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 dueSteps(rec chargingTaskRecord, now time.Time) []models.ChargingStep { loc := time.Local if rec.Zone != "" { if l, err := time.LoadLocation(rec.Zone); err == nil { loc = l } } local := now.In(loc) if len(rec.Days) > 0 && !containsDay(rec.Days, int(local.Weekday())) { return nil } // 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. It is // per task rather than per step because two steps never share a minute in // any flow that means anything — and when they do, they are sent together. if last, err := time.Parse(time.RFC3339, rec.LastRun); err == nil { if !last.Before(local.Truncate(time.Minute)) { return nil } } var due []models.ChargingStep for _, step := range rec.Steps { h, m, ok := parseHHMM(step.Time) if !ok { continue // a step whose time is not a time of day fires nothing } if local.Hour() == h && local.Minute() == m { due = append(due, step) } } return due } func containsDay(days []int, day int) bool { for _, d := range days { if d == day { return true } } return false } // fireChargingSteps sends the given steps to each of the task's 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) fireChargingSteps(ctx context.Context, who *callerIdentity, rec chargingTaskRecord, steps []models.ChargingStep) []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 } for _, step := range steps { one := out status, err := s.sendStep(ctx, who, c.Serial, step) if err != nil { one.Error = err.Error() } else { one.Status = status } results = append(results, one) } } 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 } // sendStep sends one step'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) sendStep(ctx context.Context, who *callerIdentity, serial string, step models.ChargingStep) (string, error) { body := map[string]any{} if step.Action == "limit" { body["amps"] = step.Amps } raw, err := json.Marshal(body) if err != nil { return "", err } path := "/api/integrations/anker-solix/chargers/" + url.PathEscape(serial) + "/" + step.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", step.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, "; ")) } } // taskRunLine is the outcome of one firing as the list shows it: which step, and // how it went. A task holds several steps now, and "2 of 2 sent" says nothing // about which of them it was. // // The step is named by its time rather than its action, because everything this // server writes into last_result is in its own words — charger names, the // control gate's refusals — while the page has a translation for every action. // A clock time reads the same in all three languages. func taskRunLine(step models.ChargingStep, results []taskRunResult) string { return step.Time + " — " + summarizeTaskRun(results) } // 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, step models.ChargingStep, results []taskRunResult) { payload := map[string]any{ "last_run": time.Now().UTC().Format(time.RFC3339), "last_result": taskRunLine(step, 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) } }