pages/internal/client/deploy.go

package client

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

	"golang.org/x/sync/errgroup"

	"github.com/iceBear67/simplepages/api"
)

// ------------------------------------------------------------ single calls

// CreateDeployment opens a deployment. Nothing is served until it is finalized
// and activated.
func (c *Client) CreateDeployment(ctx context.Context, project string, meta map[string]string) (api.Deployment, error) {
	var out api.Deployment
	req := api.CreateDeploymentRequest{Meta: meta}
	err := c.do(ctx, http.MethodPost, api.PathDeployments(project), req, &out)
	return out, err
}

// SetManifest declares the deployment's complete file list and returns the
// digests the server does not have yet.
//
// A large manifest is both slow to send and slow to insert, so it gets a
// deadline scaled to its size rather than the management timeout.
func (c *Client) SetManifest(ctx context.Context, project, id string, files []api.FileEntry) (api.ManifestResponse, error) {
	var out api.ManifestResponse
	req := api.ManifestRequest{Files: files}
	// Roughly a millisecond per file on top of the base timeout: a 50,000-file
	// manifest gets a minute of headroom, a small one gets no extra.
	extra := time.Duration(len(files)) * time.Millisecond
	err := c.doWithTimeout(ctx, c.timeout+extra, http.MethodPost,
		api.PathManifest(project, id), req, &out)
	return out, err
}

// Finalize verifies every blob arrived and assembles the deployment.
//
// It returns an *api.Error with code blobs_missing when uploads are outstanding;
// the digests are in details["missing"].
func (c *Client) Finalize(ctx context.Context, project, id string, fileCount int) (api.Deployment, error) {
	var out api.Deployment
	// Assembly hardlinks or copies every file, so this scales with the file
	// count in the same way the manifest insert does.
	extra := time.Duration(fileCount) * time.Millisecond
	err := c.doWithTimeout(ctx, c.timeout+extra, http.MethodPost,
		api.PathFinalize(project, id), nil, &out)
	return out, err
}

// Activate switches the project to this deployment. Passing an older id is how
// a rollback is performed.
func (c *Client) Activate(ctx context.Context, project, id string) (api.Deployment, error) {
	var out api.Deployment
	err := c.do(ctx, http.MethodPost, api.PathActivate(project, id), nil, &out)
	return out, err
}

// PutBlob uploads one blob's contents.
//
// It bypasses do, which is JSON-only: the body is raw bytes, the length is
// declared up front so the server can refuse an oversized file before reading
// it, and the deadline has to accommodate a file rather than an API call.
//
// A digest the server already holds is a fast 200 — that is what makes a
// re-run of a failed deploy cheap.
func (c *Client) PutBlob(ctx context.Context, digest string, size int64, body io.Reader) (api.BlobResponse, error) {
	var out api.BlobResponse
	path := api.PathBlob(digest)

	ctx, cancel := context.WithTimeout(ctx, uploadTimeout(c.timeout, size))
	defer cancel()

	req, err := http.NewRequestWithContext(ctx, http.MethodPut, c.base+path, body)
	if err != nil {
		return out, err
	}
	req.ContentLength = size
	req.Header.Set("Authorization", "Bearer "+c.token)
	req.Header.Set("Accept", "application/json")
	req.Header.Set("User-Agent", c.agent)
	req.Header.Set("Content-Type", "application/octet-stream")

	resp, err := c.http.Do(req)
	if err != nil {
		return out, requestError(http.MethodPut, path, err)
	}
	defer func() {
		io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10))
		resp.Body.Close()
	}()

	if resp.StatusCode >= 400 {
		return out, wrapRetryable(resp, responseError(resp))
	}
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		return out, fmt.Errorf("decode PUT %s response: %w", path, err)
	}
	return out, nil
}

// uploadTimeout gives a transfer the base timeout plus enough time to move its
// bytes over a slow link. The floor is deliberately pessimistic — a CI runner
// on a hotel connection should finish, not time out halfway and start over.
func uploadTimeout(base time.Duration, size int64) time.Duration {
	const bytesPerSecond = 128 << 10
	return base + time.Duration(size/bytesPerSecond)*time.Second
}

// doWithTimeout is do with an explicit deadline instead of the client's.
//
// do reads c.timeout, and one Client is shared by every upload goroutine, so
// the field cannot be swapped in place. The struct is a string, a duration and
// two pointers; copying it is cheaper than the synchronisation would be.
func (c *Client) doWithTimeout(ctx context.Context, timeout time.Duration, method, path string, body, out any) error {
	tmp := *c
	tmp.timeout = timeout
	return tmp.do(ctx, method, path, body, out)
}

// ------------------------------------------------------------------ retries

// throttled marks an error the deploy loop should retry, and carries the
// server's Retry-After when it sent one.
//
// The delay rides on a wrapper rather than on api.Error because api.Error is
// the wire type: a field that never appears in JSON does not belong in it.
// api.CodeOf unwraps, so callers still see the underlying code.
type throttled struct {
	err   error
	after time.Duration
}

func (t *throttled) Error() string { return t.err.Error() }
func (t *throttled) Unwrap() error { return t.err }

// wrapRetryable tags the responses that are worth trying again: rate limiting,
// and anything the server reports as a transient failure of its own.
func wrapRetryable(resp *http.Response, err error) error {
	switch {
	case resp.StatusCode == http.StatusTooManyRequests,
		resp.StatusCode >= 500:
		return &throttled{err: err, after: retryAfter(resp)}
	}
	return err
}

// retryAfter reads the header in its delay-seconds form. The HTTP-date form is
// ignored on purpose: honouring it means trusting the server's clock against
// ours, and the backoff below is a perfectly good fallback.
func retryAfter(resp *http.Response) time.Duration {
	v := strings.TrimSpace(resp.Header.Get("Retry-After"))
	if v == "" {
		return 0
	}
	secs, err := strconv.Atoi(v)
	if err != nil || secs < 0 {
		return 0
	}
	const maxWait = 60 * time.Second
	d := time.Duration(secs) * time.Second
	return min(d, maxWait)
}

// retryable reports whether err is worth another attempt, and how long to wait
// before it if the server asked for a specific delay.
//
// Transport errors are retried because a dropped connection mid-upload is the
// single most common failure on a CI runner. A 4xx other than 429 is not: the
// request is wrong and repeating it will not fix it.
func retryable(err error) (time.Duration, bool) {
	if err == nil {
		return 0, false
	}
	var t *throttled
	if errors.As(err, &t) {
		return t.after, true
	}
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		// A deadline that came from the caller's context means give up; one from
		// our own per-upload timeout is indistinguishable here, so treat both as
		// fatal rather than risk a retry storm against a wedged server.
		return 0, false
	}
	var apiErr *api.Error
	if errors.As(err, &apiErr) {
		// An envelope arrived, so the server is reachable and answered on
		// purpose. Only rate limiting is worth repeating.
		return 0, apiErr.Code == api.CodeRateLimited
	}
	// Anything else is a transport failure.
	return 0, true
}

// backoff is exponential with jitter, capped. The jitter matters when a CI
// fleet retries in lockstep after a server restart.
func backoff(attempt int, seed uint64) time.Duration {
	const (
		base    = 250 * time.Millisecond
		longest = 15 * time.Second
	)
	d := min(base<<min(attempt, 6), longest)
	// Full jitter over [d/2, d), derived from the digest so the goroutines
	// spread out without needing a shared random source.
	half := d / 2
	return half + time.Duration(seed%uint64(half))
}

// ------------------------------------------------------------- deploy flow

// DeployOptions drives Deploy.
type DeployOptions struct {
	Project string
	Source  *Source
	Meta    map[string]string
	// Activate switches the project over once the upload is complete.
	Activate bool
	// Concurrency bounds simultaneous blob uploads. Zero means 8.
	Concurrency int
	// Retries is the number of additional attempts per blob. Zero means 4.
	Retries int
	// Progress receives human-readable step messages. Calls are serialised, so
	// an implementation needs no locking of its own. It may be nil.
	Progress func(string)
}

// DeployResult reports what a deploy did.
type DeployResult struct {
	Deployment api.Deployment `json:"deployment"`
	FileCount  int            `json:"file_count"`
	TotalBytes int64          `json:"total_bytes"`
	// Uploaded counts blobs this run actually sent; Deduplicated counts the ones
	// the server already had. Their sum is the number of distinct digests.
	Uploaded      int    `json:"uploaded"`
	UploadedBytes int64  `json:"uploaded_bytes"`
	Deduplicated  int    `json:"deduplicated"`
	Activated     bool   `json:"activated"`
	URL           string `json:"url,omitempty"`
}

// Deploy runs the full sequence: create, negotiate the manifest, upload what is
// missing, finalize, and optionally activate.
//
// Interrupting it is safe and cheap to recover from. Blobs that made it are
// already in the content-addressed store, so re-running the same deploy
// negotiates a much smaller missing set — resumption falls out of the protocol
// rather than needing one of its own.
func (c *Client) Deploy(ctx context.Context, opts DeployOptions) (*DeployResult, error) {
	if opts.Source == nil || len(opts.Source.Files) == 0 {
		return nil, errors.New("nothing to deploy")
	}
	if opts.Concurrency <= 0 {
		opts.Concurrency = 8
	}
	if opts.Retries <= 0 {
		opts.Retries = 4
	}

	var mu sync.Mutex
	report := func(format string, args ...any) {
		if opts.Progress == nil {
			return
		}
		mu.Lock()
		defer mu.Unlock()
		opts.Progress(fmt.Sprintf(format, args...))
	}

	src := opts.Source
	res := &DeployResult{FileCount: len(src.Files), TotalBytes: src.TotalBytes}

	dep, err := c.CreateDeployment(ctx, opts.Project, opts.Meta)
	if err != nil {
		return nil, err
	}
	res.Deployment = dep
	report("created deployment %s", dep.ID)

	man, err := c.SetManifest(ctx, opts.Project, dep.ID, src.Manifest())
	if err != nil {
		return nil, err
	}
	res.Deduplicated = man.Have
	report("%d files, %s; %s, %s to upload",
		man.FileCount, humanBytes(man.TotalBytes),
		plural(len(man.Missing), "new blob"), humanBytes(man.MissingBytes))

	if len(man.Missing) > 0 {
		up, bytes, err := c.upload(ctx, src, man.Missing, opts, report)
		res.Uploaded, res.UploadedBytes = up, bytes
		if err != nil {
			return nil, err
		}
	}

	fin, err := c.Finalize(ctx, opts.Project, dep.ID, len(src.Files))
	if err != nil {
		// The server may have lost a blob between our upload and the finalize —
		// a GC race, or a restart mid-write. It tells us exactly which, so send
		// those again and finalize once more rather than failing the build.
		missing, ok := missingFrom(err)
		if !ok {
			return nil, err
		}
		report("server is missing %s after upload; resending", plural(len(missing), "blob"))
		up, bytes, uerr := c.upload(ctx, src, missing, opts, report)
		res.Uploaded += up
		res.UploadedBytes += bytes
		if uerr != nil {
			return nil, uerr
		}
		fin, err = c.Finalize(ctx, opts.Project, dep.ID, len(src.Files))
		if err != nil {
			return nil, err
		}
	}
	res.Deployment = fin
	res.URL = fin.URL
	report("finalized %s (%d files, %s)", fin.ID, fin.FileCount, humanBytes(fin.TotalBytes))

	if opts.Activate {
		act, err := c.Activate(ctx, opts.Project, dep.ID)
		if err != nil {
			return nil, err
		}
		res.Deployment = act
		res.Activated = true
		if act.URL != "" {
			res.URL = act.URL
		}
		report("activated %s", act.ID)
	}
	return res, nil
}

// upload sends the named digests, at most Concurrency at a time.
func (c *Client) upload(ctx context.Context, src *Source, digests []string,
	opts DeployOptions, report func(string, ...any)) (int, int64, error) {

	// A digest may back several paths; any one of them has the bytes.
	byDigest := make(map[string]LocalFile, len(src.Files))
	for _, f := range src.Files {
		if _, ok := byDigest[f.Digest]; !ok {
			byDigest[f.Digest] = f
		}
	}

	var (
		mu    sync.Mutex
		count int
		sent  int64
	)
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(opts.Concurrency)
	for _, digest := range digests {
		f, ok := byDigest[digest]
		if !ok {
			// The server asked for something we never offered. Failing here beats
			// finalizing into a deployment that can never become ready.
			return count, sent, fmt.Errorf("server reported digest %s as missing, "+
				"but it is not in the manifest", digest)
		}
		g.Go(func() error {
			if err := c.putRetrying(ctx, src, f, opts.Retries, report); err != nil {
				return err
			}
			mu.Lock()
			count++
			sent += f.Size
			mu.Unlock()
			return nil
		})
	}
	err := g.Wait()
	if err == nil {
		report("uploaded %s (%s)", plural(count, "blob"), humanBytes(sent))
	}
	return count, sent, err
}

// putRetrying uploads one blob, retrying transient failures.
//
// The body is reopened for every attempt: an io.Reader that has already been
// partly consumed cannot be replayed, and a retry that sent the tail of a file
// would be rejected as a digest mismatch — correctly, but confusingly.
func (c *Client) putRetrying(ctx context.Context, src *Source, f LocalFile,
	retries int, report func(string, ...any)) error {

	var last error
	for attempt := 0; attempt <= retries; attempt++ {
		if attempt > 0 {
			wait, _ := retryable(last)
			if wait == 0 {
				wait = backoff(attempt-1, seedOf(f.Digest))
			}
			report("retrying %s in %s (%v)", shortDigest(f.Digest), wait.Round(time.Millisecond), last)
			t := time.NewTimer(wait)
			select {
			case <-ctx.Done():
				t.Stop()
				return ctx.Err()
			case <-t.C:
			}
		}

		body, err := src.Open(f.Path)
		if err != nil {
			return fmt.Errorf("%s: %w", f.Path, err)
		}
		_, err = c.PutBlob(ctx, f.Digest, f.Size, body)
		body.Close()
		if err == nil {
			return nil
		}
		last = err
		if _, ok := retryable(err); !ok {
			return fmt.Errorf("upload %s: %w", f.Path, err)
		}
	}
	return fmt.Errorf("upload %s: giving up after %d attempts: %w", f.Path, retries+1, last)
}

// missingFrom extracts the digest list from a blobs_missing error.
func missingFrom(err error) ([]string, bool) {
	var apiErr *api.Error
	if !errors.As(err, &apiErr) || apiErr.Code != api.CodeBlobsMissing {
		return nil, false
	}
	raw, ok := apiErr.Details["missing"].([]any)
	if !ok || len(raw) == 0 {
		return nil, false
	}
	out := make([]string, 0, len(raw))
	for _, v := range raw {
		s, ok := v.(string)
		if !ok {
			return nil, false
		}
		out = append(out, s)
	}
	return out, true
}

// seedOf derives a per-blob jitter seed from its digest, so retries of
// different blobs spread out without a shared random source.
func seedOf(digest string) uint64 {
	var h uint64 = 1469598103934665603 // FNV-1a offset basis
	for i := 0; i < len(digest); i++ {
		h ^= uint64(digest[i])
		h *= 1099511628211
	}
	return h
}

func shortDigest(d string) string {
	if len(d) > 12 {
		return d[:12]
	}
	return d
}

func plural(n int, noun string) string {
	if n == 1 {
		return "1 " + noun
	}
	return strconv.Itoa(n) + " " + noun + "s"
}

// humanBytes renders a size the way a person reads it. It lives here rather
// than in cliutil because the progress messages are produced by this package.
func humanBytes(n int64) string {
	const unit = 1024
	if n < unit {
		return strconv.FormatInt(n, 10) + " B"
	}
	div, exp := int64(unit), 0
	for v := n / unit; v >= unit && exp < 4; v /= unit {
		div *= unit
		exp++
	}
	return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTP"[exp])
}