DriverVault/API Server/internal/api/chargingtasks_run.go

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)
	}
}