package host

// The wake story, end to end, in one process: an instance asleep → its timer
// comes due → the tick wakes it through the activator → the app's handler
// runs and sees the contract headers → the instance hibernates again once it
// is idle (spec §6.3, §7.1).
//
// The seam is activator.Backend — the same seam internal/activator's own
// tests use, with a fake that models an exec child faithfully enough for this
// story: the instance's unix socket exists ONLY while it is "running", so a
// delivery that did not wake it would find nothing to dial, and the handler
// running at all is proof the wake happened.

import (
	"context"
	"errors"
	"net"
	"net/http"
	"net/http/httptest"
	"net/http/httputil"
	"path/filepath"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/edge"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/schedule"
)

// --- the fake instance and its backend ---------------------------------------

// fakeInstanceProc is the spawned child: stopping it takes the socket away,
// which is what "hibernated" means to anything downstream.
type fakeInstanceProc struct {
	inst *fakeInstance
}

func (p *fakeInstanceProc) Stop(ctx context.Context) error { return p.inst.stop(ctx) }
func (p *fakeInstanceProc) Exited() bool                   { return !p.inst.running() }

// fakeInstance is an app listening on a unix socket, started and stopped by
// the backend below. It records every request it served.
type fakeInstance struct {
	socket  string
	handler http.Handler

	mu       sync.Mutex
	srv      *http.Server
	starts   int
	conns    int // accepted connections — the transport-reuse assertion
	requests []*http.Request
}

func (f *fakeInstance) running() bool {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.srv != nil
}

func (f *fakeInstance) start() error {
	ln, err := net.Listen("unix", f.socket)
	if err != nil {
		return err
	}
	srv := &http.Server{
		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			f.mu.Lock()
			f.requests = append(f.requests, r.Clone(context.Background()))
			f.mu.Unlock()
			f.handler.ServeHTTP(w, r)
		}),
		ConnState: func(_ net.Conn, st http.ConnState) {
			if st == http.StateNew {
				f.mu.Lock()
				f.conns++
				f.mu.Unlock()
			}
		},
	}
	f.mu.Lock()
	f.srv = srv
	f.starts++
	f.mu.Unlock()
	go srv.Serve(ln)
	return nil
}

func (f *fakeInstance) stop(ctx context.Context) error {
	f.mu.Lock()
	srv := f.srv
	f.srv = nil
	f.mu.Unlock()
	if srv == nil {
		return nil
	}
	return srv.Close()
}

func (f *fakeInstance) served() []*http.Request {
	f.mu.Lock()
	defer f.mu.Unlock()
	return append([]*http.Request(nil), f.requests...)
}

func (f *fakeInstance) startCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.starts
}

func (f *fakeInstance) connCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return f.conns
}

// fakeWakeBackend is activator.Backend over one fakeInstance: no S3, no
// processes, no replication — just the spawn/stop the wake path drives.
type fakeWakeBackend struct{ inst *fakeInstance }

type nopLease struct{}

func (nopLease) Replicate(context.Context, string, time.Duration) (activator.Rep, error) {
	return nopRep{}, nil
}
func (nopLease) Release(context.Context) error { return nil }

type nopRep struct{}

func (nopRep) Err() error                  { return nil }
func (nopRep) Close(context.Context) error { return nil }

func (b *fakeWakeBackend) Acquire(context.Context, registry.Route) (activator.Lease, error) {
	return nopLease{}, nil
}
func (b *fakeWakeBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
func (b *fakeWakeBackend) Seed(context.Context, registry.Route) error            { return nil }
func (b *fakeWakeBackend) Start(_ context.Context, _ registry.Route) (activator.Proc, string, error) {
	if err := b.inst.start(); err != nil {
		return nil, "", err
	}
	return &fakeInstanceProc{inst: b.inst}, "", nil
}
func (b *fakeWakeBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
func (b *fakeWakeBackend) HasSidecar(registry.Route) bool                         { return false }
func (b *fakeWakeBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) {
	return nil, nil
}

// StartWorker: this fixture drives the INSTANCE wake path, so nothing here
// ever asks for a worker. Refusing rather than returning a nil Proc is the
// safer stub — a nil Proc would be dereferenced by the caller and read as a
// crash inside the wake, which is a confusing way to learn a test wandered
// into the wrong door.
func (b *fakeWakeBackend) StartWorker(context.Context, registry.Route, map[string]string) (activator.Proc, string, error) {
	return nil, "", errors.New("fakeWakeBackend: no worker path in this fixture")
}

// wakeFixture builds a sleeping instance: a registry with one hibernate-
// flagged exec route, a fakeInstance that listens only while it is "running",
// and an activator over it with its sweeper started. dd is the agent's data
// dir — where scheduleToken mints CARLOS_ADMIN_TOKEN.
func wakeFixture(t *testing.T, host string, idle time.Duration, h http.HandlerFunc) (*registry.Registry, *fakeInstance, *activator.Service, string) {
	t.Helper()
	dd := t.TempDir()
	reg, err := registry.Open(filepath.Join(dd, "r.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	inst := &fakeInstance{socket: filepath.Join(dd, "app.sock"), handler: h}
	if err := reg.Upsert(registry.Route{
		Host: host, Kind: "instance", Account: "bab", App: "app", Status: "ready",
		Hibernate: true, Socket: inst.socket, DBPath: filepath.Join(dd, "app.db"),
	}); err != nil {
		t.Fatal(err)
	}
	svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: idle}, &fakeWakeBackend{inst: inst}, reg)
	svc.Start()
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		svc.Shutdown(ctx)
	})
	if svc.Owns(host) || inst.running() {
		t.Fatal("fixture: the instance is supposed to start asleep")
	}
	return reg, inst, svc, dd
}

// --- the story ---------------------------------------------------------------

// TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain is the branch
// review's "prove the wake story" test. It asserts, in order:
//
//  1. the instance is ASLEEP — nothing is listening on its socket;
//  2. a due timer's tick wakes it (the backend spawns exactly once);
//  3. the app's handler runs and sees the contract: the bearer token the
//     agent minted for THIS instance, and X-Carlos-Schedule-At carrying the
//     occurrence (spec §6.3's header table);
//  4. the in-flight hold is real — the handler is held open past the idle
//     window, over several sweeps, and the instance stays awake;
//  5. the delivery is recorded as a 2xx on the row; and
//  6. the instance hibernates once the tick has been idle for the window:
//     the activator drops the entry and the socket goes away again.
//
// What it does NOT prove: anything about the real ExecBackend — no process is
// forked, no database is restored, no lease is taken, and the "hibernate" of
// step 6 is the activator's stop path driving this fake's Stop, not a SIGTERM
// drain of a real child. Those belong to internal/activator's own tests; what
// this pins is that the runner's delivery leg drives that machinery correctly
// from a sleeping start.
func TestScheduledTickWakesASleepingInstanceAndItHibernatesAgain(t *testing.T) {
	t.Parallel()
	const host = "sleepy.bab.oncarlos.com"
	// The idle window is short so the test can watch a real hibernation.
	const idle = 400 * time.Millisecond

	// The handler PARKS until the test releases it, rather than sleeping a
	// fixed span. That is what takes the clock out of step 4: the instance
	// must stay awake for as long as the handler runs, so the test can look
	// at "still awake" whenever it likes, and a stalled machine only makes
	// the handler run longer — it cannot turn the assertion red.
	seen := make(chan *http.Request, 4)
	release := make(chan struct{})
	// If an assertion t.Fatalfs before the release below, the parked
	// handler goroutine must not outlive the test holding this channel:
	// close it on cleanup unless the success path already has. Reading a
	// closed channel succeeds immediately, so the select stays correct
	// after the normal close.
	t.Cleanup(func() {
		select {
		case <-release:
		default:
			close(release)
		}
	})
	reg, inst, svc, dd := wakeFixture(t, host, idle, func(w http.ResponseWriter, r *http.Request) {
		seen <- r.Clone(context.Background())
		<-release
		w.WriteHeader(http.StatusNoContent)
	})

	now := time.Now().UTC()
	occurrence := now.Unix() - 5
	r := &schedule.Runner{
		Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg),
		Now: func() time.Time { return now }, Timeout: 30 * time.Second,
		Box: "box1", Build: "test",
	}
	if err := reg.UpsertTimer(registry.Timer{
		Host: host, Name: "sync", Kind: schedule.KindSchedule, Spec: "every 6h0m0s",
		Path: "/jobs/sync", NextAt: occurrence, DueAt: occurrence,
	}); err != nil {
		t.Fatal(err)
	}

	// 2 + 3. The tick wakes it and the handler sees the contract.
	r.Tick(context.Background())
	var req *http.Request
	select {
	case req = <-seen:
	case <-time.After(30 * time.Second):
		t.Fatal("the handler never ran — the tick did not wake the sleeping instance")
	}
	tok, err := scheduleToken(dd, svc, reg)(host)
	if err != nil {
		t.Fatal(err)
	}
	if got := req.Header.Get("Authorization"); got != "Bearer "+tok {
		t.Fatalf("Authorization = %q, want the bearer of this instance's own minted token", got)
	}
	if got := req.Header.Get("X-Carlos-Schedule-At"); got != strconv.FormatInt(occurrence, 10) {
		t.Fatalf("X-Carlos-Schedule-At = %q, want the occurrence %d", got, occurrence)
	}
	if got := req.Header.Get("X-Carlos-Schedule"); got != "sync" {
		t.Fatalf("X-Carlos-Schedule = %q", got)
	}
	if got := req.Header.Get("X-Carlos-Schedule-Kind"); got != schedule.KindSchedule {
		t.Fatalf("X-Carlos-Schedule-Kind = %q, want %q", got, schedule.KindSchedule)
	}
	if req.Method != http.MethodPost || req.URL.Path != "/jobs/sync" {
		t.Fatalf("request = %s %s", req.Method, req.URL.Path)
	}
	if n := inst.startCount(); n != 1 {
		t.Fatalf("the instance was started %d times, want exactly 1 wake", n)
	}

	// 4. The in-flight hold. The handler is parked, so the instance has now
	// been "idle" (no activity stamp — that lands when the request finishes)
	// for longer than the whole idle window, and the sweeper has run several
	// times over it. It must still be awake: the activator's in-flight count
	// is what holds it open for the length of a handler (spec §6.3), and
	// without that hold a 30-minute job would be hibernated out from under
	// itself. No deadline pressure here — the handler waits for the test.
	time.Sleep(3 * idle)
	if !svc.Owns(host) || !inst.running() {
		t.Fatalf("the instance was hibernated with its tick still running (owns=%v running=%v) — the in-flight hold did not hold",
			svc.Owns(host), inst.running())
	}
	close(release)

	// 5. The row records the 204 the handler wrote once released, and
	// advances to the next occurrence.
	deadline := time.Now().Add(30 * time.Second)
	for {
		tm, _ := reg.GetTimer(host, "sync")
		if tm != nil && tm.LastStatus != "" {
			if tm.LastStatus != "204" {
				t.Fatalf("recorded %q, want 204 — the handler ran past the %s idle window and must not be cut short", tm.LastStatus, idle)
			}
			if tm.NextAt <= now.Unix() {
				t.Fatalf("row did not advance: %+v", tm)
			}
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the delivery was never recorded on the row")
		}
		time.Sleep(5 * time.Millisecond)
	}

	// 6. Idle again: the sweep hibernates it, and the socket goes with it.
	for {
		if !svc.Owns(host) && !inst.running() {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("the instance never hibernated after the tick (owns=%v running=%v)", svc.Owns(host), inst.running())
		}
		time.Sleep(5 * time.Millisecond)
	}
	// And it woke exactly once for the whole story — the tick did not
	// ping-pong the instance awake behind its own back.
	if n := inst.startCount(); n != 1 {
		t.Fatalf("the instance was started %d times over the story, want 1", n)
	}
	if reqs := inst.served(); len(reqs) != 1 {
		t.Fatalf("the instance served %d requests, want the one tick", len(reqs))
	}
}

// The delivery leg caches one proxy — and therefore one connection pool — per
// socket path, so a schedule that fires every minute stops dialling from
// scratch every time. Asserted where it is visible: the instance's accepted
// connection count.
func TestDeliveryReusesTheConnectionForASocket(t *testing.T) {
	const host = "warm.bab.oncarlos.com"
	dd := t.TempDir()
	inst := &fakeInstance{
		socket:  filepath.Join(dd, "app.sock"),
		handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }),
	}
	if err := inst.start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { inst.stop(context.Background()) })

	// svc nil: an instance that is not activator-managed is proxied straight,
	// which is the same delivery path minus the wake.
	d := &activatorDeliverer{}
	rt := registry.Route{Host: host, Kind: "instance", Socket: inst.socket}
	for i := 0; i < 3; i++ {
		req, err := schedule.BuildRequest(rt, registry.Timer{Name: "sync", Kind: schedule.KindSchedule, Path: "/jobs/sync", NextAt: 1, DueAt: 1}, "tok", "test")
		if err != nil {
			t.Fatal(err)
		}
		status, err := d.Deliver(context.Background(), rt, req)
		if status != http.StatusNoContent || err != nil {
			t.Fatalf("delivery %d = %d %v", i, status, err)
		}
	}
	if n := inst.connCount(); n != 1 {
		t.Fatalf("three deliveries opened %d connections, want 1 — the transport is not being reused", n)
	}
	if len(inst.served()) != 3 {
		t.Fatalf("the instance served %d ticks, want 3", len(inst.served()))
	}
}

// The arming switch (I6): a box with the activator armed and NO deploy store
// still gets a RUNNING runner. Its once-timers arrive on the control socket
// (spec §7.2) and nothing else on the box would ever fire them — the bug this
// pins is gating the runner on the store, which left those rows inert.
//
// Asserted by firing one: a once-row on a storeless box's registry is
// delivered to its instance and finishes.
func TestArmScheduleFiresOnceTimersWithNoDeployStore(t *testing.T) {
	t.Parallel()
	const host = "storeless.bab.oncarlos.com"
	served := make(chan struct{}, 1)
	reg, _, svc, dd := wakeFixture(t, host, 30*time.Second, func(w http.ResponseWriter, r *http.Request) {
		served <- struct{}{}
		w.WriteHeader(http.StatusNoContent)
	})

	sd := armSchedule(reg, nil, errors.New("no deploy store on this box"), svc, nil, dd)
	if sd == nil || sd.runner == nil {
		t.Fatal("a box with the activator armed and no store got no runner — its once-timers would never fire")
	}
	if sd.conv != nil || scheduleLegFor(sd) != nil {
		t.Fatal("no store, but a converger was built — it has nothing to read declarations from")
	}
	if sd.runner.Report != nil {
		t.Fatal("no store, but the runner reports — there is nowhere to write a status object")
	}

	now := time.Now()
	sd.runner.Now = func() time.Time { return now }
	if err := reg.UpsertTimer(registry.Timer{
		Host: host, Name: "remind", Kind: schedule.KindOnce, Spec: "at " + strconv.FormatInt(now.Unix()-1, 10),
		Path: "/jobs/remind", NextAt: now.Unix() - 1, DueAt: now.Unix() - 1,
	}); err != nil {
		t.Fatal(err)
	}
	sd.runner.Tick(context.Background())
	select {
	case <-served:
	case <-time.After(30 * time.Second):
		t.Fatal("the once-timer never reached the instance on a storeless box")
	}
	deadline := time.Now().Add(30 * time.Second)
	for {
		tm, _ := reg.GetTimer(host, "remind")
		if tm != nil && tm.LastStatus == "204" && tm.NextAt == 0 { // a finished once-row
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("the once-row never finished: %+v", tm)
		}
		time.Sleep(5 * time.Millisecond)
	}

	// The other two arms of the same switch.
	sd = armSchedule(reg, release.NewDirStore(t.TempDir()), nil, svc, nil, dd)
	if sd == nil || sd.conv == nil || sd.runner == nil || sd.runner.Report == nil {
		t.Fatalf("with a deploy store, want both halves and a reporter: %+v", sd)
	}
	if sd := armSchedule(reg, nil, errors.New("no deploy store"), nil, nil, dd); sd != nil {
		t.Fatal("no store and no activator: nothing can write a row or fire one, so nothing should arm")
	}
}

// The proxy cache is bounded: one entry per socket path, the same handler
// back for the same path, dropped when its delivery errors, and the whole map
// reset rather than grown past maxProxyCache.
func TestProxyCacheIsStableAndBounded(t *testing.T) {
	t.Parallel()
	d := &activatorDeliverer{}
	first := d.proxyFor("/run/carlos/a.sock")
	if again := d.proxyFor("/run/carlos/a.sock"); again != first {
		t.Fatal("the same socket got a second proxy — its connection pool is being thrown away")
	}
	if other := d.proxyFor("/run/carlos/b.sock"); other == first {
		t.Fatal("two sockets share one proxy — the transport would dial the wrong file")
	}
	d.dropProxy("/run/carlos/a.sock")
	if again := d.proxyFor("/run/carlos/a.sock"); again == first {
		t.Fatal("dropProxy kept the entry")
	}
	for i := 0; i < maxProxyCache+10; i++ {
		d.proxyFor("/run/carlos/" + strconv.Itoa(i) + ".sock")
	}
	d.mu.Lock()
	n := len(d.proxies)
	d.mu.Unlock()
	if n > maxProxyCache {
		t.Fatalf("cache holds %d entries, want at most maxProxyCache (%d)", n, maxProxyCache)
	}

	// A delivery that ERRORS retires its socket: the next tick dials fresh.
	// Reached here through the delivery's own deadline — deliberately, and
	// not through a missing socket, which is NOT an error at this seam: the
	// unix dialer retries a missing socket for a fixed 13s window
	// (edge.dialUnixWithRetry, for an instance mid-spawn) and ReverseProxy
	// then answers 502 itself, with no error for Deliver to see.
	rt := registry.Route{Host: "gone.bab.oncarlos.com", Kind: "instance", Socket: filepath.Join(t.TempDir(), "missing.sock")}
	req, err := schedule.BuildRequest(rt, registry.Timer{Name: "sync", Kind: schedule.KindSchedule, Path: "/jobs/sync", NextAt: 1, DueAt: 1}, "tok", "test")
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
	defer cancel()
	if status, err := d.Deliver(ctx, rt, req.WithContext(ctx)); err == nil {
		t.Fatalf("delivery that ran past its deadline = %d, want an error", status)
	}
	d.mu.Lock()
	_, kept := d.proxies[rt.Socket]
	d.mu.Unlock()
	if kept {
		t.Fatal("a failed delivery left its proxy (and its dead pool) in the cache")
	}
}

// The prune leg's eviction (schedule.SocketRetainer): a route that goes away
// stops receiving deliveries, so nothing it caches can error its own way out
// — the daily sweep's live-socket set is what reclaims it.
func TestRetainSocketsDropsRetiredRoutes(t *testing.T) {
	d := &activatorDeliverer{}
	keep := d.proxyFor("/run/carlos/keep.sock")
	d.proxyFor("/run/carlos/retired.sock")
	d.RetainSockets(map[string]bool{"/run/carlos/keep.sock": true})
	d.mu.Lock()
	_, retired := d.proxies["/run/carlos/retired.sock"]
	n := len(d.proxies)
	d.mu.Unlock()
	if retired || n != 1 {
		t.Fatalf("after the sweep the cache holds %d entries (retired kept: %v), want just the live one", n, retired)
	}
	if again := d.proxyFor("/run/carlos/keep.sock"); again != keep {
		t.Fatal("the sweep dropped a socket that is still routed here")
	}
	// An empty live set is a box with no exec routes at all: nothing to keep.
	d.RetainSockets(map[string]bool{})
	d.mu.Lock()
	n = len(d.proxies)
	d.mu.Unlock()
	if n != 0 {
		t.Fatalf("cache holds %d entries after an empty sweep, want 0", n)
	}
	// The runner must be able to find this method through the interface, or
	// the sweep silently never runs.
	if _, ok := any(d).(schedule.SocketRetainer); !ok {
		t.Fatal("activatorDeliverer no longer satisfies schedule.SocketRetainer — the prune leg's eviction is dead code")
	}
}

// closeIdle returns a dropped proxy's pooled connections immediately instead
// of waiting out the 90s IdleConnTimeout, and it can only do that through two
// type assertions on what edge.UnixProxy builds. Pin the shape: if edge ever
// returns something else, the eager close silently stops happening, and this
// test is what says so.
func TestUnixProxyKeepsTheShapeCloseIdleNeeds(t *testing.T) {
	rp, ok := edge.UnixProxy("/run/carlos/x.sock").(*httputil.ReverseProxy)
	if !ok {
		t.Fatal("edge.UnixProxy no longer returns a *httputil.ReverseProxy — closeIdle is now a no-op")
	}
	if _, ok := rp.Transport.(*http.Transport); !ok {
		t.Fatal("edge.UnixProxy's Transport is no longer an *http.Transport — closeIdle is now a no-op")
	}
}

// The bug, as a test: a schedule that fires more often than the idle window
// used to pin its instance awake forever, because each tick stamped activity
// and bought another full window. With the tick activity-invisible, the
// instance sleeps between ticks and wakes again for the next one.
//
// No timer rows are armed here, deliberately: this exercises the delivery
// leg alone. The sweep guard (which reads the timers table) would correctly
// hold a host awake for a tick due this soon, and that is Task 3's test.
func TestDeliveryDoesNotHoldTheInstanceAwake(t *testing.T) {
	t.Parallel()
	const host = "ticky.bab.oncarlos.com"
	const idle = 300 * time.Millisecond
	reg, inst, svc, _ := wakeFixture(t, host, idle, func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusNoContent)
	})
	rt, err := reg.Get(host)
	if err != nil {
		t.Fatalf("get route: %v", err)
	}
	d := &activatorDeliverer{svc: svc}

	// Tick twice per idle window for a few windows. Under the old accounting
	// the instance could never reach the sweep's threshold; now it does,
	// between ticks, and the next tick wakes it again.
	deadline := time.Now().Add(20 * time.Second)
	for inst.startCount() < 2 && time.Now().Before(deadline) {
		req := httptest.NewRequest(http.MethodPost, "http://carlos/jobs/sweep", nil)
		if _, err := d.Deliver(context.Background(), *rt, req); err != nil {
			t.Fatalf("Deliver: %v", err)
		}
		time.Sleep(idle / 2)
	}
	if n := inst.startCount(); n < 2 {
		t.Fatalf("the instance started %d time(s): it never hibernated between ticks, so the tick is still buying residency", n)
	}
}

// The app-facing half of the tick contract is untouched by any of this.
func TestDeliveryStillCarriesScheduleAtHeader(t *testing.T) {
	const host = "hdr.bab.oncarlos.com"
	seen := make(chan string, 1)
	reg, _, svc, _ := wakeFixture(t, host, time.Hour, func(w http.ResponseWriter, r *http.Request) {
		select {
		case seen <- r.Header.Get("X-Carlos-Schedule-At"):
		default:
		}
		w.WriteHeader(http.StatusNoContent)
	})
	rt, err := reg.Get(host)
	if err != nil {
		t.Fatalf("get route: %v", err)
	}
	d := &activatorDeliverer{svc: svc}

	req := httptest.NewRequest(http.MethodPost, "http://carlos/jobs/sweep", nil)
	req.Header.Set("X-Carlos-Schedule-At", "1788126300")
	status, err := d.Deliver(context.Background(), *rt, req)
	if err != nil {
		t.Fatalf("Deliver: %v", err)
	}
	if status != http.StatusNoContent {
		t.Fatalf("status = %d, want 204", status)
	}
	select {
	case got := <-seen:
		if got != "1788126300" {
			t.Errorf("handler saw X-Carlos-Schedule-At = %q, want it passed through unchanged", got)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("the handler never ran")
	}
}