// Command moveprobe is the moves rehearsal probe (moves spec §14.4): a small
// app shaped like a customer app, deployed by the rehearsal runbook so that a
// move can be shown to carry every kind of state an instance has.
//
// It holds state in each place an instance can:
//
//   - the database: a background loop writes a row every 500 ms, a forked
//     worker writes one every second, and timer fires write one each;
//   - the files directory: instance.key, written at first boot and checked
//     against the database at every boot;
//   - the platform's instance key (CARLOS_INSTANCE_KEY), fingerprinted at
//     first boot and checked at every boot;
//   - its store binding (MOVEPROBE_*): an object written at boot and read
//     back on every readiness check;
//   - a sealed secret (MOVEPROBE_SECRET);
//   - timers: once-timers armed through the instance control socket, and a
//     declared schedule set by the operator with `carlos schedule set`.
//
// The exec backend starts it as `moveprobe --socket <path> --db <path>`
// (internal/activator/backend_exec.go, spawnChildFrom). The files directory
// is the database path without ".db" (the flat layout). "/" answers what
// "/ready" answers, because a move probes "/" when moves.ready_path is unset.
//
// MOVEPROBE_FAIL_READY_IN=<region>[,after-write], read at startup, makes the
// probe fail its readiness in the named region. With no suffix (before-write)
// it writes nothing at all: no database open, no file, no worker, no socket
// call. With ",after-write" it writes one background row, lets its worker
// write, and then fails readiness. A box with no CARLOS_REGION is region
// "home".
//
// The README beside this file has the build and deploy lines.
package main

import (
	"bufio"
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"crypto/subtle"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"io"
	"io/fs"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"

	_ "modernc.org/sqlite"
)

const (
	bgEvery     = 500 * time.Millisecond
	workerEvery = time.Second
	// workerGrace is how long the worker keeps writing after its parent has
	// gone. Long enough to outlive the main through a cut's stop and stop
	// evidence (seconds); short enough that an ordinary sleep, which stops
	// only the main, does not leave it writing for ever.
	workerGrace = 60 * time.Second
	// workerStale is how long the main waits for the worker's next status
	// line (one a second) before readiness fails.
	workerStale = 3 * time.Second
	storeWait   = 10 * time.Second
	// storeCacheFor is how long one readiness store check answers for:
	// anonymous /ready traffic costs at most one check per window.
	storeCacheFor = 5 * time.Second

	modeNormal      = "normal"
	modeBeforeWrite = "before-write"
	modeAfterWrite  = "after-write"
)

// workerGraceOverride, set at link time (-ldflags "-X
// main.workerGraceOverride=8s"), replaces workerGrace. Only the tests set
// it, to watch the worker leave after seconds rather than a minute; a build
// that does not set it keeps workerGrace.
var workerGraceOverride string

// orphanGrace is workerGrace, or the link-time override when one is set.
func orphanGrace() time.Duration {
	if d, err := time.ParseDuration(workerGraceOverride); err == nil && d > 0 {
		return d
	}
	return workerGrace
}

func main() {
	if len(os.Args) > 1 && os.Args[1] == "worker" {
		os.Exit(runWorker(os.Args[2:]))
	}
	os.Exit(run(os.Args[1:]))
}

func logf(format string, args ...any) {
	fmt.Fprintf(os.Stderr, "moveprobe: "+format+"\n", args...)
}

// region is CARLOS_REGION, or "home" when the box has no placement code and
// the agent leaves the variable out.
func region() (name, from string) {
	if r := strings.TrimSpace(os.Getenv("CARLOS_REGION")); r != "" {
		return r, "CARLOS_REGION"
	}
	return "home", "unset, treated as home"
}

// failMode reads MOVEPROBE_FAIL_READY_IN for this region. A suffix other
// than after-write is read as before-write, the mode that writes nothing.
func failMode(spec, here string) string {
	spec = strings.TrimSpace(spec)
	if spec == "" {
		return modeNormal
	}
	where, suffix, hasSuffix := strings.Cut(spec, ",")
	if strings.TrimSpace(where) != here {
		return modeNormal
	}
	if hasSuffix && strings.TrimSpace(suffix) == modeAfterWrite {
		return modeAfterWrite
	}
	if hasSuffix {
		logf("MOVEPROBE_FAIL_READY_IN suffix %q is not after-write; using before-write", suffix)
	}
	return modeBeforeWrite
}

func run(args []string) int {
	fl := flag.NewFlagSet("moveprobe", flag.ContinueOnError)
	sock := fl.String("socket", "", "unix socket to listen on")
	dbPath := fl.String("db", "", "the instance database")
	if err := fl.Parse(args); err != nil {
		return 2
	}
	if *sock == "" || *dbPath == "" {
		logf("--socket and --db are required")
		return 2
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	here, from := region()
	mode := failMode(os.Getenv("MOVEPROBE_FAIL_READY_IN"), here)
	if mode == modeBeforeWrite {
		logf("before-write mode in region %s: nothing is written", here)
		return serve(ctx, *sock, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
			http.Error(w, "before-write mode: this box writes nothing", http.StatusServiceUnavailable)
		}))
	}

	a, err := boot(ctx, *dbPath, here, from, mode)
	if err != nil {
		logf("%v", err)
		return 1
	}
	defer a.db.Close()

	loopCtx, stopLoop := context.WithCancel(ctx)
	defer stopLoop()
	if mode == modeAfterWrite {
		// The mode's whole promise is this row; without it the rollback
		// would have nothing to find, so the app does not start.
		if err := a.insertSeq("bg"); err != nil {
			logf("after-write: the bg row did not commit: %v", err)
			return 1
		}
	} else {
		go a.bgLoop(loopCtx)
	}
	a.forkWorker()
	return serve(ctx, *sock, a.routes())
}

// serve listens on sock until ctx ends, then shuts down.
func serve(ctx context.Context, sock string, h http.Handler) int {
	if fi, err := os.Lstat(sock); err == nil && fi.Mode()&fs.ModeSocket != 0 {
		os.Remove(sock)
	}
	ln, err := net.Listen("unix", sock)
	if err != nil {
		logf("listen: %v", err)
		return 1
	}
	srv := &http.Server{Handler: h, ReadHeaderTimeout: 10 * time.Second}
	errc := make(chan error, 1)
	go func() { errc <- srv.Serve(ln) }()
	select {
	case err := <-errc:
		logf("serve: %v", err)
		return 1
	case <-ctx.Done():
	}
	shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	srv.Shutdown(shutCtx)
	return 0
}

type app struct {
	db         *sql.DB
	dbPath     string
	filesDir   string
	host       string
	region     string
	regionFrom string
	mode       string
	bootID     string

	fileKey    string // meta instance_key; empty when the boot could not settle it
	fileKeyErr error  // set at boot when the file and the database disagree
	platformFP string // SHA-256 of CARLOS_INSTANCE_KEY
	platErr    error

	store    *s3.Client
	storeErr error
	bucket   string
	objKey   string
	objBody  string

	// The readiness store check: one in flight at a time, its result kept
	// for storeCacheFor.
	storeMu      sync.Mutex
	storeRunning chan struct{} // closed when the running check finishes
	storeAt      time.Time     // when the cached result was taken
	storeResult  error

	mu        sync.Mutex
	objPut    bool
	workerPID int
	// The worker reports every write over a pipe: how many committed and
	// how many failed since this start, how many of the latest failed in a
	// row, and the last error.
	workerCommitted int64
	workerFailed    int64
	workerInARow    int64
	workerLastErr   string
	workerLastAt    time.Time // the latest status line
	workerExit      string    // why the worker is gone; empty while it runs

	// control is the one client for the instance control socket, nil when
	// the agent delivered none.
	control *http.Client
}

func openDB(path string) (*sql.DB, error) {
	db, err := sql.Open("sqlite", "file:"+url.PathEscape(path)+"?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)")
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(1)
	return db, nil
}

func boot(ctx context.Context, dbPath, here, from, mode string) (*app, error) {
	id := make([]byte, 8)
	if _, err := rand.Read(id); err != nil {
		return nil, err
	}
	a := &app{
		dbPath:     dbPath,
		filesDir:   strings.TrimSuffix(dbPath, ".db"),
		host:       os.Getenv("CARLOS_HOST"),
		region:     here,
		regionFrom: from,
		mode:       mode,
		bootID:     hex.EncodeToString(id),
	}
	db, err := openDB(dbPath)
	if err != nil {
		return nil, fmt.Errorf("open %s: %w", dbPath, err)
	}
	a.db = db
	for _, q := range []string{
		`CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)`,
		`CREATE TABLE IF NOT EXISTS rows (
			id INTEGER PRIMARY KEY AUTOINCREMENT,
			kind TEXT NOT NULL,
			name TEXT NOT NULL DEFAULT '',
			seq INTEGER NOT NULL DEFAULT 0,
			due_at INTEGER NOT NULL DEFAULT 0,
			fired_at INTEGER NOT NULL,
			region TEXT NOT NULL)`,
		`CREATE INDEX IF NOT EXISTS rows_kind_seq ON rows (kind, seq)`,
	} {
		if _, err := db.Exec(q); err != nil {
			db.Close()
			return nil, fmt.Errorf("schema: %w", err)
		}
	}
	if err := os.MkdirAll(a.filesDir, 0o700); err != nil {
		db.Close()
		return nil, fmt.Errorf("files directory: %w", err)
	}
	if err := a.settleFileKey(); err != nil {
		db.Close()
		return nil, err
	}
	if err := a.settlePlatformKey(); err != nil {
		db.Close()
		return nil, err
	}
	a.setupStore()
	if sock := os.Getenv("CARLOS_CONTROL_SOCKET"); sock != "" {
		a.control = &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{
			DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
				var d net.Dialer
				return d.DialContext(ctx, "unix", sock)
			},
		}}
	}
	if a.storeErr == nil {
		putCtx, cancel := context.WithTimeout(ctx, storeWait)
		if err := a.putObject(putCtx); err != nil {
			logf("store: %v", err)
		}
		cancel()
	}
	return a, nil
}

func (a *app) meta(key string) (string, bool, error) {
	var v string
	err := a.db.QueryRow(`SELECT value FROM meta WHERE key = ?`, key).Scan(&v)
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	return v, err == nil, err
}

func (a *app) keyPath() string { return filepath.Join(a.filesDir, "instance.key") }

// settleFileKey writes instance.key at first boot and records it in meta;
// on a later boot it remembers meta's value for the readiness check. A file
// with no database key is a mismatch, never a reason to adopt the file.
func (a *app) settleFileKey() error {
	v, found, err := a.meta("instance_key")
	if err != nil {
		return fmt.Errorf("meta: %w", err)
	}
	if found {
		a.fileKey = v
		return nil
	}
	if _, err := os.Lstat(a.keyPath()); err == nil {
		a.fileKeyErr = errors.New("key mismatch: instance.key exists but the database has no key")
		return nil
	}
	raw := make([]byte, 32)
	if _, err := rand.Read(raw); err != nil {
		return err
	}
	key := hex.EncodeToString(raw)
	f, err := os.OpenFile(a.keyPath(), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
	if err != nil {
		return fmt.Errorf("instance.key: %w", err)
	}
	if _, err := f.WriteString(key + "\n"); err != nil {
		f.Close()
		return fmt.Errorf("instance.key: %w", err)
	}
	if err := f.Sync(); err != nil {
		f.Close()
		return fmt.Errorf("instance.key: %w", err)
	}
	if err := f.Close(); err != nil {
		return fmt.Errorf("instance.key: %w", err)
	}
	if _, err := a.db.Exec(`INSERT INTO meta (key, value) VALUES ('instance_key', ?)`, key); err != nil {
		return fmt.Errorf("meta: %w", err)
	}
	a.fileKey = key
	return nil
}

// settlePlatformKey records CARLOS_INSTANCE_KEY's fingerprint the first time
// it is seen and holds every later boot to it.
func (a *app) settlePlatformKey() error {
	k := os.Getenv("CARLOS_INSTANCE_KEY")
	if k == "" {
		a.platErr = errors.New("CARLOS_INSTANCE_KEY is not set")
		return nil
	}
	a.platformFP = fingerprint(k)
	v, found, err := a.meta("platform_key_sha256")
	if err != nil {
		return fmt.Errorf("meta: %w", err)
	}
	if !found {
		if _, err := a.db.Exec(`INSERT INTO meta (key, value) VALUES ('platform_key_sha256', ?)`, a.platformFP); err != nil {
			return fmt.Errorf("meta: %w", err)
		}
		return nil
	}
	if v != a.platformFP {
		a.platErr = errors.New("CARLOS_INSTANCE_KEY differs from the one seen at first boot")
	}
	return nil
}

func fingerprint(s string) string {
	sum := sha256.Sum256([]byte(s))
	return hex.EncodeToString(sum[:])
}

// fileKeyNow is instance.key as it is on disk now, trimmed.
func (a *app) fileKeyNow() (string, error) {
	b, err := os.ReadFile(a.keyPath())
	if err != nil {
		return "", err
	}
	return strings.TrimSpace(string(b)), nil
}

func (a *app) checkFileKey() error {
	if a.fileKeyErr != nil {
		return a.fileKeyErr
	}
	got, err := a.fileKeyNow()
	switch {
	case errors.Is(err, fs.ErrNotExist):
		return errors.New("key mismatch: instance.key is missing")
	case err != nil:
		return fmt.Errorf("key mismatch: instance.key: %v", err)
	case subtle.ConstantTimeCompare([]byte(got), []byte(a.fileKey)) != 1:
		return errors.New("key mismatch: instance.key differs from the database")
	}
	return nil
}

// ---- the store ----------------------------------------------------------------

type storeConfig struct {
	bucket, region, endpoint, accessKey, secretKey string
}

// storeFromEnv reads the MOVEPROBE binding. The endpoint is optional: the
// platform delivers it only when the deployment has one, and never on AWS.
func storeFromEnv() (storeConfig, error) {
	c := storeConfig{
		bucket:    os.Getenv("MOVEPROBE_BUCKET"),
		region:    os.Getenv("MOVEPROBE_REGION"),
		endpoint:  os.Getenv("MOVEPROBE_ENDPOINT"),
		accessKey: os.Getenv("MOVEPROBE_ACCESS_KEY"),
		secretKey: os.Getenv("MOVEPROBE_SECRET_KEY"),
	}
	var missing []string
	for _, v := range []struct{ name, val string }{
		{"MOVEPROBE_BUCKET", c.bucket},
		{"MOVEPROBE_REGION", c.region},
		{"MOVEPROBE_ACCESS_KEY", c.accessKey},
		{"MOVEPROBE_SECRET_KEY", c.secretKey},
	} {
		if v.val == "" {
			missing = append(missing, v.name)
		}
	}
	if len(missing) > 0 {
		return c, fmt.Errorf("store: %s not set", strings.Join(missing, ", "))
	}
	return c, nil
}

// newStoreClient builds the S3 client from the binding's keys alone. It
// never loads the default credential chain: the app shares the agent's uid
// on a flat box, so the chain could find the box's own role and read the
// object back without the binding. Path-style only with an endpoint, as
// objstore.S3ClientFor does.
func newStoreClient(c storeConfig) *s3.Client {
	return s3.New(s3.Options{
		Region: c.region,
		Credentials: aws.CredentialsProviderFunc(func(context.Context) (aws.Credentials, error) {
			return aws.Credentials{AccessKeyID: c.accessKey, SecretAccessKey: c.secretKey, Source: "MOVEPROBE"}, nil
		}),
		BaseEndpoint: func() *string {
			if c.endpoint == "" {
				return nil
			}
			return aws.String(c.endpoint)
		}(),
		UsePathStyle:               c.endpoint != "",
		RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
		ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
	})
}

func (a *app) setupStore() {
	c, err := storeFromEnv()
	if err != nil {
		a.storeErr = err
		return
	}
	a.store = newStoreClient(c)
	a.bucket = c.bucket
	a.objKey = "probe/" + a.bootID + ".txt"
	a.objBody = fmt.Sprintf("moveprobe boot %s host %s region %s at %s\n", a.bootID, a.host, a.region, time.Now().UTC().Format(time.RFC3339))
}

func (a *app) putObject(ctx context.Context) error {
	_, err := a.store.PutObject(ctx, &s3.PutObjectInput{
		Bucket: aws.String(a.bucket),
		Key:    aws.String(a.objKey),
		Body:   bytes.NewReader([]byte(a.objBody)),
	})
	if err != nil {
		return fmt.Errorf("put %s: %w", a.objKey, err)
	}
	a.mu.Lock()
	a.objPut = true
	a.mu.Unlock()
	return nil
}

func (a *app) getObject(ctx context.Context) (string, error) {
	out, err := a.store.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(a.bucket), Key: aws.String(a.objKey)})
	if err != nil {
		return "", fmt.Errorf("get %s: %w", a.objKey, err)
	}
	defer out.Body.Close()
	b, err := io.ReadAll(io.LimitReader(out.Body, 1<<16))
	if err != nil {
		return "", fmt.Errorf("get %s: %w", a.objKey, err)
	}
	return string(b), nil
}

// checkStore puts the boot's object if the boot could not, then reads it
// back.
func (a *app) checkStore(ctx context.Context) error {
	if a.storeErr != nil {
		return a.storeErr
	}
	ctx, cancel := context.WithTimeout(ctx, storeWait)
	defer cancel()
	a.mu.Lock()
	put := a.objPut
	a.mu.Unlock()
	if !put {
		if err := a.putObject(ctx); err != nil {
			return fmt.Errorf("store: %v", err)
		}
	}
	got, err := a.getObject(ctx)
	if err != nil {
		return fmt.Errorf("store: %v", err)
	}
	if got != a.objBody {
		return fmt.Errorf("store: %s reads back different bytes", a.objKey)
	}
	return nil
}

// storeVerdict is checkStore for readiness, coalesced: a result younger than
// storeCacheFor is reused, and a request that arrives while a check runs
// waits for that check instead of starting another. The check itself runs
// on its own deadline, so one caller giving up does not cut it short for the
// others.
func (a *app) storeVerdict(ctx context.Context) error {
	a.storeMu.Lock()
	if !a.storeAt.IsZero() && time.Since(a.storeAt) < storeCacheFor {
		err := a.storeResult
		a.storeMu.Unlock()
		return err
	}
	if running := a.storeRunning; running != nil {
		a.storeMu.Unlock()
		select {
		case <-running:
		case <-ctx.Done():
			return fmt.Errorf("store: %v", ctx.Err())
		}
		a.storeMu.Lock()
		err := a.storeResult
		a.storeMu.Unlock()
		return err
	}
	done := make(chan struct{})
	a.storeRunning = done
	a.storeMu.Unlock()

	err := a.checkStore(context.Background())

	a.storeMu.Lock()
	a.storeResult, a.storeAt, a.storeRunning = err, time.Now(), nil
	a.storeMu.Unlock()
	close(done)
	return err
}

// ---- rows ---------------------------------------------------------------------

// insertSeq writes the next row of a sequenced kind. The sequence is taken
// inside the insert, so two writers (a main and a worker from an earlier
// boot, or two generations) never take the same number.
func (a *app) insertSeq(kind string) error {
	return insertSeq(a.db, kind, a.region)
}

func insertSeq(db *sql.DB, kind, region string) error {
	_, err := db.Exec(`INSERT INTO rows (kind, seq, fired_at, region)
		SELECT ?, coalesce(max(seq), 0) + 1, ?, ? FROM rows WHERE kind = ?`,
		kind, time.Now().Unix(), region, kind)
	return err
}

func (a *app) bgLoop(ctx context.Context) {
	t := time.NewTicker(bgEvery)
	defer t.Stop()
	for {
		if err := a.insertSeq("bg"); err != nil {
			logf("bg: %v", err)
		}
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// ---- the worker ---------------------------------------------------------------

// forkWorker starts the worker in its own session, with its stdio on
// /dev/null, so it neither shares the main's signals nor holds the pipes the
// backend reads the main's output from. A pipe of the main's own (the
// worker's fd 3) carries the outcome of each of the worker's writes.
func (a *app) forkWorker() {
	status, statusW, err := os.Pipe()
	if err != nil {
		logf("worker: %v", err)
		return
	}
	cmd := exec.Command(os.Args[0], "worker", "--db", a.dbPath, "--parent", strconv.Itoa(os.Getpid()), "--status-fd", "3")
	cmd.SysProcAttr = detachAttr()
	cmd.ExtraFiles = []*os.File{statusW}
	err = cmd.Start()
	statusW.Close()
	if err != nil {
		status.Close()
		logf("worker: %v", err)
		a.mu.Lock()
		a.workerExit = "did not start: " + err.Error()
		a.mu.Unlock()
		return
	}
	a.mu.Lock()
	a.workerPID = cmd.Process.Pid
	a.mu.Unlock()
	go func() {
		err := cmd.Wait()
		how := "exit status 0"
		if err != nil {
			how = err.Error()
		}
		a.mu.Lock()
		a.workerExit = "exited (" + how + ")"
		a.mu.Unlock()
	}()
	go a.readWorkerStatus(status)
}

// readWorkerStatus reads the worker's lines: "ok", or "fail <error>".
func (a *app) readWorkerStatus(r *os.File) {
	defer r.Close()
	sc := bufio.NewScanner(r)
	for sc.Scan() {
		line := sc.Text()
		a.mu.Lock()
		a.workerLastAt = time.Now()
		if msg, failed := strings.CutPrefix(line, "fail "); failed {
			a.workerFailed++
			a.workerInARow++
			a.workerLastErr = msg
		} else {
			a.workerCommitted++
			a.workerInARow = 0
		}
		a.mu.Unlock()
	}
	// The pipe closes when the worker exits; Wait's own account, when it
	// comes, replaces this one.
	a.mu.Lock()
	if a.workerExit == "" {
		a.workerExit = "exited (its status pipe closed)"
	}
	a.mu.Unlock()
}

// checkWorker holds readiness to the worker: it is still running and has
// sent a status line within workerStale, at least one of its rows has
// committed since this start, and its latest write succeeded.
func (a *app) checkWorker() error {
	a.mu.Lock()
	defer a.mu.Unlock()
	switch {
	case a.workerExit != "":
		return errors.New("worker: " + a.workerExit)
	case !a.workerLastAt.IsZero() && time.Since(a.workerLastAt) > workerStale:
		return fmt.Errorf("worker: no status for %ds", int(time.Since(a.workerLastAt).Seconds()))
	case a.workerCommitted == 0 && a.workerInARow > 0:
		return fmt.Errorf("worker: no row committed since start; last error: %s", a.workerLastErr)
	case a.workerCommitted == 0:
		return errors.New("worker: no row committed since start")
	case a.workerInARow > 0:
		return fmt.Errorf("worker: the last write failed: %s", a.workerLastErr)
	}
	return nil
}

// runWorker writes a worker row every second. Once its parent has gone it
// keeps writing for workerGrace (orphanGrace) and then exits.
func runWorker(args []string) int {
	fl := flag.NewFlagSet("moveprobe worker", flag.ContinueOnError)
	dbPath := fl.String("db", "", "the instance database")
	parent := fl.Int("parent", 0, "the main's pid")
	statusFD := fl.Int("status-fd", 0, "a pipe to report each write on (0: none)")
	if err := fl.Parse(args); err != nil || *dbPath == "" || *parent <= 0 {
		return 2
	}
	here, _ := region()
	// Writes to the pipe fail once the main has gone; they are ignored.
	report := func(err error) {}
	if *statusFD > 0 {
		status := os.NewFile(uintptr(*statusFD), "status")
		defer status.Close()
		report = func(err error) {
			if err != nil {
				fmt.Fprintf(status, "fail %s\n", strings.ReplaceAll(err.Error(), "\n", " "))
			} else {
				fmt.Fprintln(status, "ok")
			}
		}
	}
	db, err := openDB(*dbPath)
	if err != nil {
		report(err)
		return 1
	}
	defer db.Close()
	grace := orphanGrace()
	var orphaned time.Time
	t := time.NewTicker(workerEvery)
	defer t.Stop()
	for {
		if orphaned.IsZero() && os.Getppid() != *parent {
			orphaned = time.Now()
		}
		if !orphaned.IsZero() && time.Since(orphaned) >= grace {
			return 0
		}
		report(insertSeq(db, "worker", here))
		<-t.C
	}
}

// ---- HTTP ---------------------------------------------------------------------

func (a *app) routes() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("/{$}", a.handleReady)
	mux.HandleFunc("/ready", a.handleReady)
	mux.HandleFunc("GET /report", a.bearer(os.Getenv("MOVEPROBE_SECRET"), a.handleReport))
	mux.HandleFunc("POST /arm-once", a.bearer(os.Getenv("MOVEPROBE_SECRET"), a.handleArmOnce))
	// The platform's runner signs each fire with the instance's admin token.
	mux.HandleFunc("POST /tick", a.bearer(os.Getenv("CARLOS_ADMIN_TOKEN"), a.fired("tick")))
	mux.HandleFunc("POST /once", a.bearer(os.Getenv("CARLOS_ADMIN_TOKEN"), a.fired("once")))
	return mux
}

// bearer admits a request carrying "Authorization: Bearer <want>". An empty
// want admits nothing.
func (a *app) bearer(want string, next http.HandlerFunc) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		got := sha256.Sum256([]byte(r.Header.Get("Authorization")))
		exp := sha256.Sum256([]byte("Bearer " + want))
		if want == "" || subtle.ConstantTimeCompare(got[:], exp[:]) != 1 {
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}
		next(w, r)
	}
}

// readiness is /ready's answer: every check that fails, or ready.
func (a *app) readiness(ctx context.Context) (int, string) {
	if a.mode == modeAfterWrite {
		return http.StatusServiceUnavailable, "after-write mode: not ready here"
	}
	var fails []string
	if err := a.checkFileKey(); err != nil {
		fails = append(fails, err.Error())
	}
	if a.platErr != nil {
		fails = append(fails, a.platErr.Error())
	}
	if os.Getenv("MOVEPROBE_SECRET") == "" {
		fails = append(fails, "MOVEPROBE_SECRET is not set")
	}
	if err := a.storeVerdict(ctx); err != nil {
		fails = append(fails, err.Error())
	}
	if err := a.checkWorker(); err != nil {
		fails = append(fails, err.Error())
	}
	if len(fails) > 0 {
		return http.StatusInternalServerError, strings.Join(fails, "; ")
	}
	return http.StatusOK, "ready"
}

func (a *app) handleReady(w http.ResponseWriter, r *http.Request) {
	code, msg := a.readiness(r.Context())
	if code != http.StatusOK {
		http.Error(w, msg, code)
		return
	}
	fmt.Fprintln(w, msg)
}

func (a *app) fired(kind string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		name := r.Header.Get("X-Carlos-Schedule")
		due, err := strconv.ParseInt(r.Header.Get("X-Carlos-Schedule-At"), 10, 64)
		if name == "" || err != nil {
			http.Error(w, "want X-Carlos-Schedule and X-Carlos-Schedule-At", http.StatusBadRequest)
			return
		}
		if _, err := a.db.Exec(`INSERT INTO rows (kind, name, due_at, fired_at, region) VALUES (?, ?, ?, ?, ?)`,
			kind, name, due, time.Now().Unix(), a.region); err != nil {
			http.Error(w, "database: "+err.Error(), http.StatusInternalServerError)
			return
		}
		fmt.Fprintln(w, "ok")
	}
}

// handleArmOnce registers once-<n> for /once, due in the given seconds,
// through the instance control socket.
func (a *app) handleArmOnce(w http.ResponseWriter, r *http.Request) {
	in, err := strconv.ParseInt(r.URL.Query().Get("in"), 10, 64)
	if err != nil || in < 0 || in > 400*24*3600 {
		http.Error(w, "in: want whole seconds, at most 400 days", http.StatusBadRequest)
		return
	}
	if a.control == nil {
		http.Error(w, "no control socket: CARLOS_CONTROL_SOCKET is not set", http.StatusServiceUnavailable)
		return
	}
	token := os.Getenv("CARLOS_ADMIN_TOKEN")
	if token == "" {
		http.Error(w, "CARLOS_ADMIN_TOKEN is not set", http.StatusServiceUnavailable)
		return
	}
	var n int64
	if err := a.db.QueryRow(`INSERT INTO meta (key, value) VALUES ('once_seq', '1')
		ON CONFLICT (key) DO UPDATE SET value = CAST(value AS INTEGER) + 1
		RETURNING CAST(value AS INTEGER)`).Scan(&n); err != nil {
		http.Error(w, "database: "+err.Error(), http.StatusInternalServerError)
		return
	}
	name := fmt.Sprintf("once-%d", n)
	at := time.Now().Add(time.Duration(in) * time.Second).UTC().Truncate(time.Second)
	body, _ := json.Marshal(map[string]string{"at": at.Format(time.RFC3339), "path": "/once"})

	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	req, err := http.NewRequestWithContext(ctx, http.MethodPut, "http://carlos/schedule/"+name, bytes.NewReader(body))
	if err != nil {
		http.Error(w, "control socket: "+err.Error(), http.StatusInternalServerError)
		return
	}
	req.Header.Set("Authorization", "Bearer "+token)
	req.Header.Set("Content-Type", "application/json")
	resp, err := a.control.Do(req)
	if err != nil {
		http.Error(w, "control socket: "+err.Error(), http.StatusBadGateway)
		return
	}
	// Drained before close, so the connection goes back for the next arm.
	defer func() {
		io.Copy(io.Discard, io.LimitReader(resp.Body, 1<<16))
		resp.Body.Close()
	}()
	reply, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
	if resp.StatusCode/100 != 2 {
		http.Error(w, fmt.Sprintf("control socket: %d %s", resp.StatusCode, strings.TrimSpace(string(reply))), http.StatusBadGateway)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]any{"name": name, "at": at.Format(time.RFC3339), "due_at": at.Unix()})
}

type rowCount struct {
	Kind   string `json:"kind"`
	Region string `json:"region"`
	Count  int64  `json:"count"`
	MinSeq int64  `json:"min_seq"`
	MaxSeq int64  `json:"max_seq"`
}

type firedCount struct {
	Kind  string `json:"kind"`
	Name  string `json:"name"`
	DueAt int64  `json:"due_at"`
	Count int64  `json:"count"`
}

type objectReport struct {
	Key   string `json:"key"`
	Body  string `json:"body"`
	Error string `json:"error,omitempty"`
}

type report struct {
	Host              string       `json:"host"`
	Region            string       `json:"region"`
	RegionFrom        string       `json:"region_from"`
	Mode              string       `json:"mode"`
	BootID            string       `json:"boot_id"`
	Rows              []rowCount   `json:"rows"`
	Fired             []firedCount `json:"fired"`
	FileKeySHA256     string       `json:"file_key_sha256"`
	PlatformKeySHA256 string       `json:"platform_key_sha256"`
	Object            objectReport `json:"object"`
	HasSecret         bool         `json:"has_secret"`
	WorkerPID         int          `json:"worker_pid"`
	WorkerCommitted   int64        `json:"worker_writes_committed"`
	WorkerFailed      int64        `json:"worker_writes_failed"`
	WorkerInARow      int64        `json:"worker_failed_in_a_row"`
	WorkerLastError   string       `json:"worker_last_error"`
	WorkerExit        string       `json:"worker_exit"`
	WorkerLastStatus  int64        `json:"worker_last_status_at"`
}

// counts reads the row counts and the timer fires. Any failed query, scan
// or read part way through is an error, never a shorter list.
func (a *app) counts(ctx context.Context) ([]rowCount, []firedCount, error) {
	rowsOut := []rowCount{}
	err := a.query(ctx, `SELECT kind, region, count(*), min(seq), max(seq) FROM rows GROUP BY kind, region ORDER BY kind, region`,
		func(rows *sql.Rows) error {
			var c rowCount
			if err := rows.Scan(&c.Kind, &c.Region, &c.Count, &c.MinSeq, &c.MaxSeq); err != nil {
				return err
			}
			rowsOut = append(rowsOut, c)
			return nil
		})
	if err != nil {
		return nil, nil, err
	}
	fired := []firedCount{}
	err = a.query(ctx, `SELECT kind, name, due_at, count(*) FROM rows WHERE kind IN ('tick', 'once') GROUP BY kind, name, due_at ORDER BY kind, name, due_at`,
		func(rows *sql.Rows) error {
			var c firedCount
			if err := rows.Scan(&c.Kind, &c.Name, &c.DueAt, &c.Count); err != nil {
				return err
			}
			fired = append(fired, c)
			return nil
		})
	if err != nil {
		return nil, nil, err
	}
	return rowsOut, fired, nil
}

func (a *app) query(ctx context.Context, q string, each func(*sql.Rows) error) error {
	rows, err := a.db.QueryContext(ctx, q)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		if err := each(rows); err != nil {
			return err
		}
	}
	return rows.Err()
}

func (a *app) handleReport(w http.ResponseWriter, r *http.Request) {
	rep := report{
		Host: a.host, Region: a.region, RegionFrom: a.regionFrom, Mode: a.mode, BootID: a.bootID,
		PlatformKeySHA256: a.platformFP,
		HasSecret:         os.Getenv("MOVEPROBE_SECRET") != "",
	}
	a.mu.Lock()
	rep.WorkerPID = a.workerPID
	rep.WorkerCommitted, rep.WorkerFailed = a.workerCommitted, a.workerFailed
	rep.WorkerInARow, rep.WorkerLastError = a.workerInARow, a.workerLastErr
	rep.WorkerExit = a.workerExit
	if !a.workerLastAt.IsZero() {
		rep.WorkerLastStatus = a.workerLastAt.Unix()
	}
	a.mu.Unlock()
	if k, err := a.fileKeyNow(); err == nil {
		rep.FileKeySHA256 = fingerprint(k)
	}

	var err error
	if rep.Rows, rep.Fired, err = a.counts(r.Context()); err != nil {
		http.Error(w, "report: "+err.Error(), http.StatusInternalServerError)
		return
	}

	rep.Object.Key = a.objKey
	if a.storeErr != nil {
		rep.Object.Error = a.storeErr.Error()
	} else {
		ctx, cancel := context.WithTimeout(r.Context(), storeWait)
		body, err := a.getObject(ctx)
		cancel()
		rep.Object.Body = body
		if err != nil {
			rep.Object.Error = err.Error()
		}
	}
	w.Header().Set("Content-Type", "application/json")
	enc := json.NewEncoder(w)
	enc.SetIndent("", "  ")
	enc.Encode(rep)
}