package api
import (
"context"
"errors"
"fmt"
"io"
"log"
"net/http"
"os"
"strings"
"time"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/s3"
"craftdesign.group/quarry/internal/work"
"craftdesign.group/quarry/internal/zitcli"
)
// Archives live in a bucket once built. A tarball costs seconds of
// compression, the route is anonymous, and any check-in may be asked
// for, so the forge builds each check-in's archive in each container
// once, puts it in the repository's bucket (its RFC 0016 binding, else
// the forge's own) and answers every request for it with a redirect to
// a presigned GET. The bytes never pass through the forge after the
// build. With no bucket anywhere the archive is built per request, as
// it always was, through the same lanes.
// archiveLinkExpiry is how long a redirect's link works. Short, so a
// link copied out of a private repository's download stops working
// soon after: the 302 is re-issued on every request, so a reader loses
// nothing by it.
const archiveLinkExpiry = 5 * time.Minute
// archiveUploadClient sends a built archive to its bucket. An archive
// can be up to maxArchiveBytes, which a 20-second client cannot always
// carry.
var archiveUploadClient = &http.Client{Timeout: 10 * time.Minute}
// archiveObjectKey is where one archive lives in a bucket:
// archive/<repo>/<producer>/<check-in>.<container>.
//
// The repository keeps two repositories that share a bucket apart, and
// gives repo.delete one prefix to remove. The producer is there for the
// reason archiveKey records: zit's container bytes changed between
// releases (zit#207), and a key without the producer kept serving the
// old container. A key, once written, never changes meaning.
func archiveObjectKey(repo, producer, version, format string) string {
ext := ".tar.gz"
if format == "zip" {
ext = ".zip"
}
key := archivePrefix(repo)
if seg := keySegment(producer); seg != "" {
key += seg + "/"
}
return key + version + ext
}
func archivePrefix(repo string) string { return "archive/" + repo + "/" }
// keySegment makes "zit 0.3.30" a key segment: what is not a letter, a
// digit, '.', '_' or '-' becomes '-'.
func keySegment(s string) string {
return strings.Map(func(r rune) rune {
switch {
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '.', r == '_', r == '-':
return r
}
return '-'
}, s)
}
// archiveBucket is the bucket repo's archives are kept in: the
// repository's own binding (its parent's, for a companion store), else
// the forge's. A binding whose last probe failed is not used; the
// archive is then built per request rather than failing.
//
// Unlike uploads, archives may go to the forge's bucket: an archive is
// the forge's own derived output, built by the forge from a store it
// serves, not a reader's file filed into the operator's account.
func (d Deps) archiveBucket(repo string) (s3.Bucket, bool) {
if d.Buckets == nil {
return s3.Bucket{}, false
}
owner := repo
if parent := d.companionParent(repo); parent != "" {
owner = parent
}
for _, name := range []string{owner, oltp.ForgeBucket} {
b, ok, err := d.Buckets.Bucket(name)
if err != nil {
log.Printf("archive: %s: reading bucket binding: %v", repo, err)
return s3.Bucket{}, false
}
if !ok {
continue
}
if b.ProbeErr != "" {
return s3.Bucket{}, false
}
return s3.Bucket{Endpoint: b.Endpoint, Name: b.Bucket, Region: b.Region, AccessKey: b.AccessKey, Secret: b.Secret}, true
}
return s3.Bucket{}, false
}
// archiveObjects is a bucket as archive.Bucketed asks of it.
type archiveObjects struct{ b s3.Bucket }
func (o archiveObjects) Where() string { return o.b.Endpoint + "/" + o.b.Name }
func (o archiveObjects) Has(ctx context.Context, key string) (bool, error) {
_, ok, err := o.b.Stat(ctx, attachClient(), key, time.Now())
return ok, err
}
func (o archiveObjects) Upload(ctx context.Context, key, path string) error {
ctype := "application/gzip"
if strings.HasSuffix(key, ".zip") {
ctype = "application/zip"
}
return o.b.PutFile(ctx, archiveUploadClient, key, path, s3.PutOptions{ContentType: ctype}, time.Now())
}
// dropArchives removes every archive of repo's from each bucket, for
// repo.delete: a repository created later under the same name starts
// with none, and a deleted repository's content does not outlive it in
// a bucket. Forget comes first: it waits for any upload under way and
// refuses every later one from a build begun before the delete, so the
// listings that follow miss nothing.
func (d Deps) dropArchives(ctx context.Context, repo string, buckets ...s3.Bucket) error {
prefix := archivePrefix(repo)
d.ArchiveObjects.Forget(prefix)
var errs []error
for _, b := range buckets {
objects, err := b.List(ctx, attachClient(), prefix, time.Now())
if err != nil {
errs = append(errs, err)
continue
}
for _, o := range objects {
if err := b.Delete(ctx, attachClient(), o.Key, time.Now()); err != nil {
errs = append(errs, fmt.Errorf("%s: %w", o.Key, err))
}
}
}
return errors.Join(errs...)
}
// archiveBucketsToClear is every bucket that may hold repo's archives:
// its own (its parent's, for a companion) and the forge's, whatever
// their last probe said, since a binding that failed a probe may have
// worked when the archives were written.
func (d Deps) archiveBucketsToClear(repo string) []s3.Bucket {
if d.Buckets == nil {
return nil
}
owner := repo
if parent := d.companionParent(repo); parent != "" {
owner = parent
}
var out []s3.Bucket
for _, name := range []string{owner, oltp.ForgeBucket} {
if b, ok, err := d.Buckets.Bucket(name); err == nil && ok {
out = append(out, s3.Bucket{Endpoint: b.Endpoint, Name: b.Bucket, Region: b.Region, AccessKey: b.AccessKey, Secret: b.Secret})
}
}
return out
}
// archiveSweepCap bounds one sweep's deletions; what is left goes in a
// later process's sweep.
const archiveSweepCap = 1000
// sweepArchives deletes, under repo's prefix, the archives a zit other
// than this forge's built: archive/<repo>/<producer>/ for every other
// producer, and the keys written with none. They can never be served
// again, since the key the forge asks for names the current producer.
// At most max are deleted. With no producer known nothing is swept:
// there is then no telling current from old.
func (d Deps) sweepArchives(ctx context.Context, repo string, b s3.Bucket, max int) (int, error) {
current := keySegment(d.ZitVersion)
if current == "" {
return 0, nil
}
prefix := archivePrefix(repo)
objects, err := b.List(ctx, attachClient(), prefix, time.Now())
if err != nil {
return 0, err
}
n := 0
var errs []error
for _, o := range objects {
if n >= max {
break
}
producer, _, nested := strings.Cut(strings.TrimPrefix(o.Key, prefix), "/")
if nested && producer == current {
continue
}
if err := b.Delete(ctx, attachClient(), o.Key, time.Now()); err != nil {
errs = append(errs, fmt.Errorf("%s: %w", o.Key, err))
continue
}
n++
}
return n, errors.Join(errs...)
}
// sweepOnce starts sweepArchives in the background the first time this
// process builds one of repo's archives into a bucket.
func (d Deps) sweepOnce(repo string, b s3.Bucket) {
if d.ZitVersion == "" || !d.ArchiveObjects.FirstSweep(repo) {
return
}
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
n, err := d.sweepArchives(ctx, repo, b, archiveSweepCap)
if n > 0 || err != nil {
log.Printf("archive: swept repo=%s removed=%d other-producer archives, err=%v", repo, n, err)
}
}()
}
// archiveVersion resolves what an archive request names to a check-in.
// held says the index has that check-in, which is the condition for
// asking a bucket about it: a hash this repository does not hold is
// never answered from a key someone else's build left behind.
func (d Deps) archiveVersion(repo *Repo, ref string) (version string, held bool) {
if repo.Index == nil {
return d.resolveVersion(repo, ref), false
}
if ref != "" {
if e, _, ok, err := repo.Index.CheckinByPrefix(ref); err == nil && ok {
return e.Hash, true
}
}
name := d.resolveVersion(repo, ref)
tag, isTag := strings.CutPrefix(name, "tag:")
if !isTag {
if b, ok, err := repo.Index.TipOf(name); err == nil && ok {
return b.Tip, true
}
}
if hash, ok, err := repo.Index.TagTarget(tag); err == nil && ok {
return hash, true
}
return name, false
}
// archiveLaneFor picks the lane a check-in's build runs on: a signed-in
// reader's own; the normal one for an open branch's tip or a tagged
// check-in, which many readers share; cold for the rest, a closed
// branch's tip included.
func (d Deps) archiveLaneFor(ctx context.Context, repo *Repo, version string) (*work.Lane, string) {
if ops.User(ctx) != "" {
return d.ArchiveReaderLane, "reader"
}
if repo.Index == nil {
return d.ArchiveLane, "normal"
}
if tags, err := repo.Index.TagsOf(version); err == nil && len(tags) > 0 {
return d.ArchiveLane, "normal"
}
if openTip(repo, version) {
return d.ArchiveLane, "normal"
}
return d.ArchiveColdLane, "cold"
}
// pairLaneFor is archiveLaneFor for a premerge pair: shared work only
// when it is an open landing's branch tip and target tip, which is the
// pair CI and reviewers ask for.
func (d Deps) pairLaneFor(ctx context.Context, repo *Repo, branchTip, targetTip string) (*work.Lane, string) {
if ops.User(ctx) != "" {
return d.ArchiveReaderLane, "reader"
}
if repo.Index == nil {
return d.ArchiveColdLane, "cold"
}
bs, _ := repo.Index.TipBranches(branchTip)
for _, b := range bs {
l, ok, err := repo.Index.OpenLandingOn(b.Name)
if err != nil || !ok {
continue
}
if t, ok, err := repo.Index.TipOf(l.Target); err == nil && ok && t.Tip == targetTip {
return d.ArchiveLane, "normal"
}
}
return d.ArchiveColdLane, "cold"
}
// openTip says version is the tip of a branch that is not closed.
func openTip(repo *Repo, version string) bool {
bs, err := repo.Index.TipBranches(version)
if err != nil {
return false
}
for _, b := range bs {
if !b.Closed {
return true
}
}
return false
}
// moveFile puts src at dst: a rename when they share a filesystem, a
// copy when they do not.
func moveFile(src, dst string) error {
if err := os.Rename(src, dst); err == nil {
return nil
}
in, err := os.Open(src)
if err != nil {
return err
}
defer in.Close()
out, err := os.Create(dst)
if err != nil {
return err
}
if _, err := io.Copy(out, in); err != nil {
out.Close()
return err
}
return out.Close()
}
// archiveTo and mergeArchiveTo build through the repository's
// builders, zit's unless a test supplied them.
func (r *Repo) archiveTo(ctx context.Context, format, version, name, dst string, max int64) error {
if r.Archive != nil {
return r.Archive(ctx, format, version, name, dst, max)
}
return r.Zit.ArchiveTo(ctx, format, version, name, dst, max)
}
func (r *Repo) mergeArchiveTo(ctx context.Context, branchTip, targetTip, name, dst string, max int64) error {
if r.MergeArchive != nil {
return r.MergeArchive(ctx, branchTip, targetTip, name, dst, max)
}
return r.Zit.MergeArchiveTo(ctx, branchTip, targetTip, name, dst, max)
}
// archiveBuildErr is a build's failure as the op answers it.
func archiveBuildErr(repo, version, short, format string, err error) error {
switch {
case errors.Is(err, work.ErrBusy):
return fmt.Errorf("archive queue is full, retry shortly (%w): %w", work.ErrBusy, ops.ErrUnavailable)
case errors.Is(err, zitcli.ErrNoVersion):
return fmt.Errorf("no version %q in %s: %w", short, repo, ops.ErrNotFound)
}
log.Printf("archive %s@%s (%s): %v", repo, version, format, err)
return fmt.Errorf("%w: %v", ops.ErrUnavailable, err)
}
// logBuild is the one line every archive build leaves.
func logBuild(repo, version, format, lane, into, dst string, start time.Time) {
var size int64
if fi, err := os.Stat(dst); err == nil {
size = fi.Size()
}
log.Printf("archive: built repo=%s checkin=%s format=%s size=%d duration=%s queue=%s into=%s",
repo, version, format, size, time.Since(start).Round(time.Millisecond), lane, into)
}
// laneOf is a work lane as the ops page lists it.
func laneOf(l *work.Lane) zitcli.Lane {
st := l.Stats()
lane := zitcli.Lane{Name: st.Name, Bound: st.Limit, Running: int(st.InFlight),
Admitted: int(st.Admitted), Refused: int(st.Shed), Waiting: st.Waiting, Queue: st.Queue}
for _, it := range l.Items() {
lane.Items = append(lane.Items, zitcli.LaneItem{Verb: it.Verb, Subject: it.Subject,
AgeMs: it.Age.Milliseconds(), Running: it.Running})
}
return lane
}