package main

// deploywatch.go is `carlos deploy`'s watch once boxes report their rollouts
// (zero-downtime spec §9.1): completion is tied to the build being watched,
// instance by instance, box by box. A host its box reports on is watched
// through that report and never requested over HTTP, because a request would
// wake an asleep instance the spec counts as done once its box has installed
// the build. A host no box reports on is watched as it always was
// (checkTarget). With no report at all, the watch IS watchVersion.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/rolloutwords"
	"github.com/carlosframework/platform/internal/status"
)

const (
	watchIntro               = "watching %s roll out box by box: an awake instance counts once the new build has taken over, an asleep one once its box has installed it"
	watchWordSwapped         = "swapped"
	watchWordInstalled       = "installed"
	watchAsleepTail          = "starts on its next request"
	watchWordFailed          = "failed"
	watchWordHeld            = "held"
	watchHeldTail            = "stayed on %s because %s failed on this box"
	watchBox                 = "box %s: %s"
	watchCountSwapped        = "%d swapped"
	watchCountAsleep         = "%d asleep with it installed"
	watchCountFailed         = "%d failed"
	watchCountHeld           = "%d stayed on their last good build"
	watchFailedSummary       = "build %s did not reach every instance: %d failed and %d stayed on their last good build. Fix it and deploy a new build, or try this one again: carlos promote --app %s --retry %s %s"
	watchHoldDownSummary     = "an instance that swaps in hold mode is down, with nothing to fall back to: carlos rollback --app %s"
	watchTimeout             = "timed out waiting for build %s:"
	watchPendingNotInstalled = "its box has not installed this build yet"
	watchPendingOther        = "its box has %s installed, not this build"
	watchPendingQueued       = "queued"
	watchPendingRunning      = "starting beside the running build"
	watchPendingSettling     = "settling: making sure it stays up for 10 seconds"
	watchPendingWaiting      = "waiting for its box to start it"
	watchPendingDark         = "its box stopped reporting"
	watchPendingUnlisted     = "the console's instance list was cut short before it, so its state is unknown"
	watchBoxBusy             = "box %s is still working through this app's instances"
	watchListFailed          = "the console's instance list could not be read, so the rollout could not be followed"
	watchPendingUnread       = "the console could not read every box's report, so its state is unknown"
)

// watchList is one answer of the console's instance list, as the watch
// reads it. incomplete is the console's rollout_incomplete: some box's
// report could not be read, so a host with no report may yet have one.
type watchList struct {
	recs       []apiInstanceRecord
	truncated  bool
	incomplete bool
}

// listStatusError is a list the console answered with a status other than
// 200. The status decides what the watch makes of it (listAbsent).
type listStatusError struct {
	status int
	err    error
}

func (e *listStatusError) Error() string { return e.err.Error() }
func (e *listStatusError) Unwrap() error { return e.err }

// listAbsent reports whether a failed list is evidence enough to fall back
// to today's URL watch (Astra plan 6 R1 #4, R2 #4). A 4xx says the console
// has no list for this caller: a console from before the door, an app with
// no instances (404). A 403 counts too, on purpose: a Publish-only token can
// deploy (promote admits Publish) but cannot read the list (GET instances
// admits Instance and Deploy), so such a deploy gets exactly the watch it
// got before reports existed. Not a 401: the deploy has just authenticated
// with this token, so a 401 is a session that lapsed or a console that
// stumbled, and says nothing about the boxes. Not a 429 either. Those, a
// 5xx, a network error and an unreadable answer are retried until the
// deadline.
func listAbsent(err error) bool {
	var se *listStatusError
	return errors.As(err, &se) && se.status >= 400 && se.status < 500 &&
		se.status != http.StatusTooManyRequests && se.status != http.StatusUnauthorized
}

// watchInstanceList is the watch's read of the instance list. It is
// instanceList's request, kept apart because the watch needs what that
// method drops: the status of a refusal, and rollout_incomplete.
func (s *apiSession) watchInstanceList(ctx context.Context, a apiAcct, app string) (watchList, error) {
	status, body, err := s.getJSON(ctx, apiPath(a, app, "/instances"))
	if err != nil {
		return watchList{}, err
	}
	if status != http.StatusOK {
		return watchList{}, &listStatusError{status: status, err: apiErrorFromBody(status, body)}
	}
	var res struct {
		Instances         []apiInstanceRecord `json:"instances"`
		Truncated         bool                `json:"truncated"`
		RolloutIncomplete bool                `json:"rollout_incomplete"`
	}
	if err := json.Unmarshal(body, &res); err != nil {
		return watchList{}, fmt.Errorf("instances list: unreadable response: %w", err)
	}
	return watchList{recs: res.Instances, truncated: res.Truncated, incomplete: res.RolloutIncomplete}, nil
}

// deployWatchListEvery is how often the production watch reads the instance
// list (controller ruling R5). The list is rate limited per token (60 a
// minute, shared with the deploy's own calls), and each read costs the
// console a records list and a box join, so it keeps its own cadence; a host
// watched by URL is still polled every deployWatchPollEvery, as before. A
// failing first list is asked again on the same cadence until the deadline.
const (
	deployWatchListEvery = 5 * time.Second
	deployWatchPollEvery = 2 * time.Second
)

// rolloutState is one targeted instance's place in the rollout of the
// watched build (spec §9.1): pending, or one of four final states.
type rolloutState string

const (
	rolloutPending rolloutState = "pending"
	rolloutSwapped rolloutState = "swapped"
	rolloutAsleep  rolloutState = "asleep" // installed, awaiting first wake
	rolloutFailed  rolloutState = "failed"
	rolloutHeld    rolloutState = "held"
)

func (s rolloutState) final() bool {
	return s == rolloutSwapped || s == rolloutAsleep || s == rolloutFailed || s == rolloutHeld
}

// attemptOpen reports whether a box's attempt word is still in progress:
// queued, running or settling, or a word this binary does not know (a newer
// box), which is safer read as unfinished than as done. Failed and down are
// outcomes, not attempts in progress.
func attemptOpen(a string) bool {
	return a != "" && a != status.AttemptFailed && a != status.AttemptDown
}

// classifyRollout decides one instance's state for build ver from its box's
// report. In order:
//   - A box that cannot arm its scheduler (Unavailable) reports no per-host
//     state, so its awake=false means nothing: pending (R6).
//   - Until the box reports ver installed the instance is pending, however
//     idle its scheduler.
//   - A failure counts against ver only when the failed attempt was ver's,
//     and a hold-mode instance left down by ver is failed too (spec §9.1 "a
//     hold failure"). The console's join (Task 8) leaves the failed attempt
//     only on the host the build failed on, so this is checked before
//     held_by, which that host carries too.
//   - Held by ver, with no failure of its own: a sibling's failure record
//     holds it on its good build.
//   - A transient host (its process is starting or stopping) is pending: it
//     is neither asleep nor swapped, and its serving build says nothing
//     settled (Astra R3 section 5 item 10). A failure or a hold it already
//     carries still stands, as above.
//   - An attempt in progress is pending whether or not the instance is awake
//     (R8): a canary serving ver has not swapped until its settle closes, and
//     an asleep instance whose first wake is under observation is not asleep.
//   - Serving ver with nothing in progress: swapped. Asleep with nothing in
//     progress: installed, awaiting first wake.
func classifyRollout(r *apiRollout, ver string) rolloutState {
	switch {
	case r.Unavailable != "":
		return rolloutPending
	case r.Installed != ver:
		return rolloutPending
	case r.AttemptBuild == ver && (r.Attempt == status.AttemptFailed || r.Attempt == status.AttemptDown):
		return rolloutFailed
	case r.HeldBy == ver:
		return rolloutHeld
	case r.Transient:
		return rolloutPending
	case attemptOpen(r.Attempt):
		return rolloutPending
	case r.Serving == ver:
		return rolloutSwapped
	case !r.Awake && r.Serving == "" && r.Attempt == "":
		return rolloutAsleep
	}
	return rolloutPending
}

// follows reports whether the watch follows an instance through r: a report
// with per-host state, or one from a box that cannot arm its scheduler for
// any reason but a known decline: a refusing box's instances wait, and a
// code this binary does not know is a newer box's problem, so neither may be
// woken by a poll. A box that declined overlap runs today's stop-then-start
// paths, so its report is no report: the instance is watched by URL, like a
// flag-off box's (R6; fix round 1).
func follows(r *apiRollout) bool {
	return r != nil && !rolloutwords.UnavailableDeclines(r.Unavailable)
}

type watchMode int

const (
	modeUnknown watchMode = iota // not yet decided: a truncated list left it out, or an incomplete join had no report for it
	modeRollout                  // watched through its box's report, for the rest of the watch
	modeURL                      // watched by URL, as watchVersion does, until a report comes
)

type watchedHost struct {
	t    watchTarget
	mode watchMode
	// ro is the latest report the watch followed. It is kept when the box
	// goes dark, so the host still counts on its box.
	ro *apiRollout
	// dark is set while the latest complete list carries no report for a
	// host the watch follows (R1). It never sends the host to a URL poll.
	dark bool
	// unread is set while the host is undecided because the console's
	// rollout join was incomplete, not because the list was cut short.
	unread bool
	// absent is set while the latest answered list does not carry the host
	// at all: cut short, or a record the console could not read (Astra plan
	// 6 R3). An absent host is never polled by URL, even if it was being
	// polled before; it is cleared the next time the host is listed.
	absent bool
	state  rolloutState
	// fin is the report the host went final on. A final state survives
	// missing observations, and later reports that agree the build is
	// installed: what it printed and what it counts come from fin. A report
	// that the box has another build installed clears it (apply).
	fin     *apiRollout
	urlLive bool
	urlDiag string
	printed bool
}

// urlPolling is whether this sweep requests the host's URL: it is watched by
// URL, is not yet live, and the console still lists it.
func (h *watchedHost) urlPolling() bool {
	return h.mode == modeURL && !h.urlLive && !h.absent
}

func (h *watchedHost) done() bool {
	switch h.mode {
	case modeURL:
		return h.urlLive
	case modeRollout:
		return h.state.final()
	}
	return false
}

// rolloutWatch is one watch. list is the console's instance list, asked to
// answer by the watch's deadline; check is one URL poll (checkTarget in
// production); onURL runs once, the first time any host is watched by
// URL, so the probe-or-header note prints only when it is true. channel names
// the `promote --retry` a failure suggests. interval is the URL poll cadence
// and listEvery the list's (R5); a zero listEvery lists on every sweep.
type rolloutWatch struct {
	w                 io.Writer
	app, ver, channel string
	timeout, interval time.Duration
	listEvery         time.Duration
	list              func(deadline time.Time) (watchList, error)
	check             func(t watchTarget, n int64) (bool, string)
	onURL             func()
	diag              *watchDiagnostics
}

func (rw rolloutWatch) run(targets []watchTarget) error {
	// One deadline governs the whole watch (Astra plan 6 R1 #2): every list
	// request, every wait between lists, and what is left for watchVersion.
	deadline := time.Now().Add(rw.timeout)
	// A failed list (a 5xx, a rate limit, a console restart, the network)
	// says nothing about the boxes, so it is asked again on the list's
	// cadence until the deadline, and never read as "no reports" (Astra plan
	// 6 R1 #4: a URL poll would wake an asleep instance on a reporting box).
	l, err := rw.list(deadline)
	for err != nil && !listAbsent(err) {
		left := time.Until(deadline)
		if left <= 0 {
			return rw.listFailed(err)
		}
		time.Sleep(min(max(rw.listEvery, rw.interval), left))
		if time.Until(deadline) <= 0 {
			return rw.listFailed(err)
		}
		l, err = rw.list(deadline)
	}
	lastList := time.Now()
	// Today's watch, byte for byte (R10), only on affirmative evidence that
	// no target is on a reporting box: the console has no list for the app
	// (listAbsent), or every target is listed, the console read every box's
	// report, and none of the targets has one the watch follows (legacyOnly).
	if err != nil || legacyOnly(l, targets) {
		if rw.onURL != nil {
			rw.onURL()
		}
		return watchVersion(rw.w, targets, rw.ver, time.Until(deadline), rw.interval, rw.diag)
	}
	var hosts []*watchedHost
	seen := map[string]bool{}
	for _, t := range targets {
		if !seen[t.Host] {
			seen[t.Host] = true
			hosts = append(hosts, &watchedHost{t: t, state: rolloutPending})
		}
	}
	c := colorFor(rw.w)
	mutedf(rw.w, watchIntro, rw.ver)
	rw.apply(hosts, l)
	boxPrinted := map[string]bool{}
	announcedURL := false
	var n int64
	for {
		urlPending := false
		for _, h := range hosts {
			if h.urlPolling() {
				urlPending = true
			}
		}
		if urlPending && !announcedURL {
			announcedURL = true
			if rw.onURL != nil {
				rw.onURL()
			}
		}
		var wg sync.WaitGroup
		for _, h := range hosts {
			if !h.urlPolling() {
				continue
			}
			n++
			wg.Add(1)
			go func(h *watchedHost, n int64) {
				defer wg.Done()
				h.urlLive, h.urlDiag = rw.check(h.t, n)
			}(h, n)
		}
		wg.Wait()
		for _, h := range hosts {
			if h.done() && !h.printed {
				h.printed = true
				rw.printHost(c, h)
			}
		}
		boxes, labels := groupByBox(hosts)
		boxesDone := true
		for _, l := range labels {
			if !boxDone(boxes[l]) {
				boxesDone = false
				continue
			}
			if !boxPrinted[l] {
				boxPrinted[l] = true
				fmt.Fprintln(rw.w, boxSummary(l, boxes[l]))
			}
		}
		hostsDone := true
		for _, h := range hosts {
			hostsDone = hostsDone && h.done()
		}
		if hostsDone && boxesDone {
			return rw.result(hosts)
		}
		if time.Now().After(deadline) {
			return rw.timedOut(hosts, boxes, labels)
		}
		time.Sleep(rw.wait(hosts, lastList, deadline))
		if time.Since(lastList) >= rw.listEvery {
			l, err = rw.list(deadline)
			lastList = time.Now()
			// A list that failed (a rate limit, a console restart) changes
			// nothing: every host keeps its mode and state until a list
			// answers.
			if err == nil {
				rw.apply(hosts, l)
			}
		}
	}
}

// wait is how long the loop sleeps: the URL poll interval while any host is
// watched by URL, otherwise until the next list is due; never past the
// deadline, so the timeout report comes on time.
func (rw rolloutWatch) wait(hosts []*watchedHost, lastList, deadline time.Time) time.Duration {
	d := rw.interval
	urlPending := false
	for _, h := range hosts {
		urlPending = urlPending || h.urlPolling()
	}
	if !urlPending {
		d = max(d, time.Until(lastList.Add(rw.listEvery)))
	}
	if left := time.Until(deadline); left > 0 && left < d {
		d = left
	}
	return d
}

// legacyOnly is the evidence the fallback to watchVersion needs (Astra plan
// 6 R1 #4, #7): every target is on the list, the console's rollout join read
// every box, and no target has a report the watch follows. It is decided on
// the targets alone. A list cut short for the whole app neither causes nor
// blocks it, and a report on an instance this deploy does not target says
// nothing about the targets' boxes.
func legacyOnly(l watchList, targets []watchTarget) bool {
	if l.incomplete {
		return false
	}
	listed := make(map[string]*apiRollout, len(l.recs))
	for _, r := range l.recs {
		listed[r.Host] = r.Rollout
	}
	for _, t := range targets {
		ro, ok := listed[t.Host]
		if !ok || follows(ro) {
			return false
		}
	}
	return true
}

// listFailed is the timeout when no list answered before the deadline:
// nothing was followed and nothing was polled, so it names the list.
func (rw rolloutWatch) listFailed(err error) error {
	var b strings.Builder
	fmt.Fprintf(&b, watchTimeout, rw.ver)
	fmt.Fprintf(&b, "\n  %s: %v", watchListFailed, err)
	if rw.diag != nil && rw.diag.logsHint != "" {
		fmt.Fprintf(&b, "\nfull timeline: %s", rw.diag.logsHint)
	}
	return errors.New(b.String())
}

// apply folds one answered list into the hosts.
//
//   - A host with a report the watch follows is watched through it from now
//     on (R4), also when it was polled by URL, even live: the report is the
//     box's word about which build runs there, and a URL answer is not
//     (Astra plan 6 R1 #3). Its report is refreshed every list even once it
//     is final, because its box is done only on the box's latest scheduler
//     word.
//   - A final state persists only across MISSING observations: a host
//     absent from a truncated list, a dark box. An explicit report that the
//     host's box has another build installed takes a swapped, asleep or held
//     host back to pending, named by what is installed (pendingDetail): the
//     watched build no longer runs there, and Review Focus 4 forbids success
//     for a replaced build. A failure stays failed: the build did fail.
//   - A host missing from the list keeps its mode and state (R3), whether
//     the cap cut it off or the console could not read its record: nothing
//     says it uses the old path, and switching it to URL polling would wake
//     it if it is asleep. It is also no longer polled if it was (absent),
//     until it is listed again (Astra plan 6 R3).
//   - A host with no followed report on a list whose rollout join was
//     incomplete keeps its mode and state too: its box's report may be the
//     one the console could not read (Astra plan 6 R1 #4). An undecided host
//     stays undecided, and is named so in a timeout.
//   - A host the watch follows whose report is gone is dark (R1): pending if
//     it was not final, never polled.
//   - Any other host is watched by URL, exactly as watchVersion would.
func (rw rolloutWatch) apply(hosts []*watchedHost, l watchList) {
	listed := make(map[string]*apiRollout, len(l.recs))
	for i := range l.recs {
		listed[l.recs[i].Host] = l.recs[i].Rollout
	}
	for _, h := range hosts {
		ro, ok := listed[h.t.Host]
		h.absent = !ok
		switch {
		case follows(ro):
			if h.mode != modeRollout {
				// What a URL answer printed ("live") stands only if the
				// report agrees; otherwise the report's own word prints
				// once it is final.
				h.state, h.fin = rolloutPending, nil
			}
			h.mode, h.ro, h.dark, h.unread = modeRollout, ro, false, false
			switch {
			case !h.state.final():
				h.state = classifyRollout(ro, rw.ver)
				if h.state.final() {
					fin := *ro
					h.fin = &fin
				}
				h.printed = h.printed && h.state == rolloutSwapped
			case h.state != rolloutFailed && ro.Installed != rw.ver:
				h.state, h.fin, h.printed = rolloutPending, nil, false
			}
		case !ok:
			// Absent from this list: cut short, or a record the console could
			// not read. Neither says the host uses the old path, so it keeps
			// its mode and state and is never moved to URL watching here
			// (Astra plan 6 R3). On an untruncated list the console left it
			// out because it could not read something, which is what
			// watchPendingUnread says.
			h.unread = !l.truncated
		case h.mode == modeURL && h.urlLive:
			// Live by URL is final too, until a report says otherwise.
		case l.incomplete:
			h.unread = h.mode == modeUnknown
		case h.mode == modeRollout:
			h.dark = true
		default:
			h.mode = modeURL
		}
	}
}

func groupByBox(hosts []*watchedHost) (map[string][]*watchedHost, []string) {
	boxes := map[string][]*watchedHost{}
	var labels []string
	for _, h := range hosts {
		if h.mode != modeRollout || h.ro == nil {
			continue
		}
		if _, ok := boxes[h.ro.Box]; !ok {
			labels = append(labels, h.ro.Box)
		}
		boxes[h.ro.Box] = append(boxes[h.ro.Box], h)
	}
	sort.Strings(labels)
	return boxes, labels
}

// boxDone is spec §9.1's rule: every targeted instance on the box is final,
// and its scheduler has nothing queued, running or settling for the app, by
// the latest word the watch has from the box.
func boxDone(hs []*watchedHost) bool {
	for _, h := range hs {
		if !h.state.final() || !h.ro.SchedulerIdle {
			return false
		}
	}
	return true
}

func boxSummary(label string, hs []*watchedHost) string {
	counts := map[rolloutState]int{}
	for _, h := range hs {
		counts[h.state]++
	}
	var parts []string
	for _, p := range []struct {
		s      rolloutState
		format string
	}{{rolloutSwapped, watchCountSwapped}, {rolloutAsleep, watchCountAsleep}, {rolloutFailed, watchCountFailed}, {rolloutHeld, watchCountHeld}} {
		if counts[p.s] > 0 {
			parts = append(parts, fmt.Sprintf(p.format, counts[p.s]))
		}
	}
	return fmt.Sprintf(watchBox, label, strings.Join(parts, ", "))
}

// stayedOn is the build a held instance is still on: what it serves, or,
// asleep, its last good build.
func stayedOn(r *apiRollout) string {
	if r.Serving != "" {
		return r.Serving
	}
	return r.Good
}

func (rw rolloutWatch) printHost(c colors, h *watchedHost) {
	host := h.t.Host
	if h.mode == modeURL {
		fmt.Fprintf(rw.w, "%s https://%s\n", c.ok("live"), host)
		return
	}
	switch h.state {
	case rolloutSwapped:
		fmt.Fprintf(rw.w, "%s https://%s\n", c.ok(watchWordSwapped), host)
	case rolloutAsleep:
		fmt.Fprintf(rw.w, "%s https://%s (%s)\n", c.ok(watchWordInstalled), host, watchAsleepTail)
	case rolloutFailed:
		fmt.Fprintf(rw.w, "%s https://%s: %s\n", c.warn(watchWordFailed), host, rolloutwords.Reason(h.fin.Reason))
	case rolloutHeld:
		fmt.Fprintf(rw.w, "%s https://%s: %s\n", c.warn(watchWordHeld), host, fmt.Sprintf(watchHeldTail, stayedOn(h.fin), rw.ver))
	}
}

func (rw rolloutWatch) result(hosts []*watchedHost) error {
	failed, held, down := 0, 0, false
	for _, h := range hosts {
		if h.mode != modeRollout {
			continue
		}
		switch h.state {
		case rolloutFailed:
			failed++
			down = down || h.fin.Attempt == status.AttemptDown
		case rolloutHeld:
			held++
		}
	}
	if failed == 0 && held == 0 {
		return nil
	}
	msg := fmt.Sprintf(watchFailedSummary, rw.ver, failed, held, rw.app, rw.ver, rw.channel)
	if down {
		msg += "\n" + fmt.Sprintf(watchHoldDownSummary, rw.app)
	}
	return errors.New(msg)
}

// pendingDetail is what a timeout says about a host still pending.
func (rw rolloutWatch) pendingDetail(h *watchedHost) string {
	r := h.ro
	switch {
	case h.mode == modeUnknown && h.unread:
		return watchPendingUnread
	case h.mode == modeUnknown:
		return watchPendingUnlisted
	case h.mode == modeURL && h.absent && h.urlDiag == "":
		// Dropped off the list before any poll said anything.
		if h.unread {
			return watchPendingUnread
		}
		return watchPendingUnlisted
	case h.mode == modeURL:
		return h.urlDiag
	case h.dark:
		return watchPendingDark
	case r.Unavailable != "":
		return rolloutwords.Unavailable(r.Unavailable)
	case r.Installed == "":
		return watchPendingNotInstalled
	case r.Installed != rw.ver:
		return fmt.Sprintf(watchPendingOther, r.Installed)
	case r.Attempt == status.AttemptQueued:
		return watchPendingQueued
	case r.Attempt == status.AttemptRunning:
		return watchPendingRunning
	case r.Attempt == status.AttemptSettling:
		return watchPendingSettling
	}
	return watchPendingWaiting
}

func (rw rolloutWatch) timedOut(hosts []*watchedHost, boxes map[string][]*watchedHost, labels []string) error {
	var b strings.Builder
	fmt.Fprintf(&b, watchTimeout, rw.ver)
	var pending []*watchedHost
	for _, h := range hosts {
		if !h.done() {
			pending = append(pending, h)
		}
	}
	sort.Slice(pending, func(i, j int) bool { return pending[i].t.Host < pending[j].t.Host })
	for _, h := range pending {
		fmt.Fprintf(&b, "\n  %s: %s", h.t.Host, rw.pendingDetail(h))
		// The reason a wake failed, from the platform log stream, as
		// watchVersion gives it: only for a host polled by URL, whose poll
		// is what woke it.
		if h.mode == modeURL && rw.diag != nil && rw.diag.wakeReason != nil {
			if r := rw.diag.wakeReason(h.t.Host); r != "" {
				fmt.Fprintf(&b, "\n      %s", r)
			}
		}
	}
	for _, l := range labels {
		allFinal := true
		for _, h := range boxes[l] {
			allFinal = allFinal && h.state.final()
		}
		if allFinal && !boxDone(boxes[l]) {
			fmt.Fprintf(&b, "\n  "+watchBoxBusy, l)
		}
	}
	if rw.diag != nil && rw.diag.logsHint != "" {
		fmt.Fprintf(&b, "\nfull timeline: %s", rw.diag.logsHint)
	}
	return errors.New(b.String())
}

// watchDeployed is the watch for an instance (non-static) deploy of ver onto
// channel.
func watchDeployed(ctx context.Context, sess *apiSession, acct apiAcct, app, ver, channel string, hosts []string, probe string, timeout time.Duration) error {
	targets := make([]watchTarget, 0, len(hosts))
	for _, h := range hosts {
		targets = append(targets, watchTarget{Host: h, Origin: watchOrigin(h), Probe: probe})
	}
	return newDeployWatch(ctx, sess, acct, app, ver, channel, probe, timeout).run(targets)
}

// newDeployWatch is watchDeployed's watch, built apart so a test can read
// its cadence without a console.
func newDeployWatch(ctx context.Context, sess *apiSession, acct apiAcct, app, ver, channel, probe string, timeout time.Duration) rolloutWatch {
	client := &http.Client{Timeout: 10 * time.Second}
	return rolloutWatch{
		w: stdout, app: app, ver: ver, channel: channel, timeout: timeout,
		interval: deployWatchPollEvery, listEvery: deployWatchListEvery,
		list: func(deadline time.Time) (watchList, error) {
			lctx, cancel := context.WithDeadline(ctx, deadline)
			defer cancel()
			return sess.watchInstanceList(lctx, acct, app)
		},
		check: func(t watchTarget, n int64) (bool, string) { return checkTarget(client, t, ver, n) },
		onURL: func() { printWatchNote(probe, false, app) },
		diag:  deployDiagnostics(ctx, sess, acct, app),
	}
}

// deployDiagnostics is the timeout report's help, moved out of runDeploy
// unchanged.
func deployDiagnostics(ctx context.Context, sess *apiSession, acct apiAcct, app string) *watchDiagnostics {
	return &watchDiagnostics{
		// A timed-out host's wake reason, from the platform log stream —
		// scoped tight (one host, errors only, the watch's own recent past)
		// so a diagnosis this may print was caused by THIS deploy, not an
		// incident from earlier in the day.
		wakeReason: func(host string) string {
			q := url.Values{"stream": {"platform"}, "level": {"error"},
				"host": {host}, "since": {"15m"}, "limit": {"1"}}
			res, err := sess.logsGet(ctx, acct, app, q)
			if err != nil || len(res.Records) == 0 {
				return ""
			}
			return res.Records[len(res.Records)-1].Msg
		},
		logsHint: fmt.Sprintf("carlos logs --app %s --account %s --stream platform", app, acct.Seg),
	}
}

// printWatchNote is runDeploy's probe-or-header note, moved out unchanged.
func printWatchNote(probe string, staticDeploy bool, app string) {
	if probe != "" {
		mutedf(stdout, "watching %s for the build %s reports itself\n", probe, app)
		return
	}
	// Said out loud, every time, because the silence is what the bug was
	// made of (discussion #13). With no probe this watch can only ask
	// the ROUTER what release was adopted; it cannot ask the process
	// what it is running. A deploy that goes green here has proved that
	// adoption happened, not that anything restarted — and an app whose
	// instance failed to cycle looks identical. It is a one-line fix
	// per app, so the line names it rather than describing the problem
	// and leaving the reader to find the verb.
	mutedf(stdout, "watching the X-Carlos-Version header: it reports the release ADOPTED, not the build running\n")
	// The advisory below assumes a process to probe. A static route has
	// none — no /api/version, no exec child to cycle — so the suggested
	// command would only teach the operator to configure a probe that
	// 404s, which then fails and times out every future `carlos deploy`
	// for that site. Static deploys get the line above (still true: the
	// header reports adoption, not a running build) and stop there.
	if !staticDeploy {
		mutedf(stdout, "  a probe proves the process actually cycled — carlos apps build-probe --app %s /api/version\n", app)
	}
}