package api
import (
"context"
"net/http"
"net/http/httptest"
"net/url"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"craftdesign.group/quarry/internal/archive"
"craftdesign.group/quarry/internal/artifact"
"craftdesign.group/quarry/internal/index"
"craftdesign.group/quarry/internal/oltp"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/work"
)
// Four check-ins: v1 is tagged, mid is neither a tip nor tagged (the
// expensive, crawlable case), tip is trunk's tip and dev a branch tip.
var (
arcV1 = strings.Repeat("a1", 32)
arcMid = strings.Repeat("b2", 32)
arcTip = strings.Repeat("c3", 32)
arcDev = strings.Repeat("d4", 32)
)
func archiveIndex(t *testing.T) *index.Index {
t.Helper()
ix, err := index.Open(filepath.Join(t.TempDir(), "index.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { ix.Close() })
if err := ix.FoldNow([]index.Parsed{
{Hash: arcV1, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:00:00", User: "dan", Comment: "root",
Tags: []artifact.Tag{{Name: "*branch", Target: "*", Value: "trunk"}, {Name: "*sym-trunk", Target: "*"}, {Name: "+sym-v1.0", Target: "*"}}}},
{Hash: arcMid, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:10:00", User: "dan", Comment: "mid",
Parents: []string{arcV1}}},
{Hash: arcTip, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:20:00", User: "dan", Comment: "tip",
Parents: []string{arcMid}}},
{Hash: arcDev, Art: &artifact.Artifact{Type: artifact.Manifest, Date: "2026-08-15T09:30:00", User: "dan", Comment: "dev",
Parents: []string{arcMid}, Tags: []artifact.Tag{{Name: "*branch", Target: "*", Value: "dev"}, {Name: "*sym-dev", Target: "*"}, {Name: "-sym-trunk", Target: "*"}}}},
}); err != nil {
t.Fatal(err)
}
return ix
}
type archiveRig struct {
r *ops.Registry
d Deps
bucket *uvBucket
builds atomic.Int32
// gate, when set, holds every build until it is closed.
gate chan struct{}
// started hears each build as it begins.
started chan string
}
const arcProducer = "zit 0.3.30"
func newArchiveRig(t *testing.T, mutate func(*Deps, *memBuckets, string)) *archiveRig {
t.Helper()
srv, bucket := newUVBucket(t)
rig := &archiveRig{bucket: bucket, started: make(chan string, 64)}
store := newMemBuckets()
store.BindBucket("test", oltp.BucketBinding{Endpoint: srv.URL, Bucket: "files", Region: "auto", AccessKey: "AK", Secret: "s3cret"})
repo := &Repo{Index: archiveIndex(t), Archive: func(ctx context.Context, format, version, name, dst string, max int64) error {
rig.builds.Add(1)
rig.started <- version
if rig.gate != nil {
<-rig.gate
}
return os.WriteFile(dst, []byte(format+" of "+version), 0o644)
}}
rig.d = Deps{Version: "v", ZitVersion: arcProducer, Repos: NewRepos(map[string]*Repo{"test": repo}), Buckets: store,
ArchiveLane: work.NewLane("archive builds", 2, 8), ArchiveColdLane: work.NewLane("cold archive builds", 1, 0), ArchiveObjects: archive.NewBucketed()}
if mutate != nil {
mutate(&rig.d, store, srv.URL)
}
rig.r = ops.NewRegistry()
if err := RegisterAll(rig.r, rig.d); err != nil {
t.Fatal(err)
}
return rig
}
func signedIn(user string) context.Context { return ops.WithUser(context.Background(), user) }
// get serves archive.get the way both the web route and the API do:
// the op, then ServeFile.
func (rig *archiveRig) get(ctx context.Context, version, format string) *httptest.ResponseRecorder {
w := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/r/test/archive/"+version, nil).WithContext(ctx)
out, err := rig.r.Invoke(ctx, "archive.get", archiveIn{Repo: "test", Version: version, Format: format})
if err != nil {
writeOpError(w, req, "archive.get", err)
return w
}
ServeFile(w, req, "archive.get", out.(ops.File))
return w
}
// put stores an object under the fake bucket's lock: a sweep may be
// running in the background.
func (rig *archiveRig) put(key string) {
rig.bucket.mu.Lock()
defer rig.bucket.mu.Unlock()
rig.bucket.objects[key] = uvObject{body: []byte("x")}
}
func (rig *archiveRig) objectKeys() []string {
rig.bucket.mu.Lock()
defer rig.bucket.mu.Unlock()
var keys []string
for k := range rig.bucket.objects {
keys = append(keys, k)
}
return keys
}
// The first request for an archive builds it once however many readers
// arrive together, puts it in the bucket, and every one of them is sent
// there; a later request is redirected without building.
func TestArchiveIsBuiltOnceIntoTheBucketAndRedirected(t *testing.T) {
rig := newArchiveRig(t, nil)
rig.gate = make(chan struct{})
const n = 8
codes := make(chan *httptest.ResponseRecorder, n)
var wg sync.WaitGroup
for range n {
wg.Add(1)
go func() { defer wg.Done(); codes <- rig.get(context.Background(), arcTip, "tar") }()
}
<-rig.started
close(rig.gate)
wg.Wait()
close(codes)
for w := range codes {
if w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d, want 302: %s", w.Code, w.Body.String())
}
}
if got := rig.builds.Load(); got != 1 {
t.Fatalf("FAIL: %d builds for %d concurrent readers, want 1", got, n)
}
want := "files/archive/test/zit-0.3.30/" + arcTip + ".tar.gz"
if keys := rig.objectKeys(); len(keys) != 1 || keys[0] != want {
t.Fatalf("FAIL: bucket holds %v, want [%s]", keys, want)
}
if body := string(rig.bucket.objects[want].body); body != "tar of "+arcTip {
t.Fatalf("FAIL: object body %q", body)
}
w := rig.get(context.Background(), arcTip, "tar")
if w.Code != http.StatusFound || rig.builds.Load() != 1 {
t.Fatalf("FAIL: second request: status %d, builds %d", w.Code, rig.builds.Load())
}
loc, err := url.Parse(w.Header().Get("Location"))
if err != nil {
t.Fatal(err)
}
q := loc.Query()
if !strings.HasSuffix(loc.Path, "/archive/test/zit-0.3.30/"+arcTip+".tar.gz") ||
q.Get("X-Amz-Expires") != "300" ||
q.Get("response-content-disposition") != "attachment; filename*=UTF-8''test-"+arcTip[:12]+".tar.gz" ||
q.Get("response-content-type") != "application/gzip" {
t.Fatalf("FAIL: Location %s", loc)
}
if cc := w.Header().Get("Cache-Control"); !strings.Contains(cc, "no-store") {
t.Fatalf("FAIL: a redirect to a five-minute link was cacheable: %q", cc)
}
if w.Body.Len() > 512 {
t.Fatalf("FAIL: the redirect carried %d bytes; archive bytes must not pass through the forge", w.Body.Len())
}
t.Logf("ok: %d readers, 1 build, then 302 to %s", n, loc.Path)
}
// Keys carry the repository, so two repositories sharing one bucket
// never answer each other's archives.
func TestArchiveKeyIncludesTheRepository(t *testing.T) {
a := archiveObjectKey("alpha", arcProducer, arcTip, "zip")
b := archiveObjectKey("beta", arcProducer, arcTip, "zip")
if a == b || !strings.HasPrefix(a, "archive/alpha/") || !strings.HasSuffix(a, arcTip+".zip") {
t.Fatalf("FAIL: %q vs %q", a, b)
}
if archiveObjectKey("alpha", "zit 0.3.31", arcTip, "zip") == a {
t.Fatal("FAIL: a new producer reuses the old producer's object")
}
t.Logf("ok: %s", a)
}
// An anonymous request for a check-in that is neither a tip nor tagged
// goes through the small queue, and is refused with 503 and a
// Retry-After when that queue is full. Tips, tags and signed-in readers
// use the normal queue and are served meanwhile.
func TestAnonymousColdArchivesUseTheSmallQueue(t *testing.T) {
rig := newArchiveRig(t, nil)
rig.gate = make(chan struct{})
first := make(chan *httptest.ResponseRecorder, 1)
go func() { first <- rig.get(context.Background(), arcMid, "tar") }()
<-rig.started
w := rig.get(context.Background(), arcMid, "zip")
if w.Code != http.StatusServiceUnavailable || w.Header().Get("Retry-After") == "" {
t.Fatalf("FAIL: a second cold build: status %d Retry-After %q, want 503 with one", w.Code, w.Header().Get("Retry-After"))
}
if st := rig.d.ArchiveColdLane.Stats(); st.Admitted != 1 || st.Shed != 1 {
t.Fatalf("FAIL: cold lane %+v, want 1 admitted and 1 shed", st)
}
served := make(chan *httptest.ResponseRecorder, 4)
go func() { served <- rig.get(context.Background(), arcTip, "zip") }() // trunk's tip
go func() { served <- rig.get(context.Background(), arcV1, "zip") }() // tagged
go func() { served <- rig.get(context.Background(), "tag:v1.0", "tar") }() // tag by name
go func() { served <- rig.get(signedIn("dan"), arcMid, "zip") }() // signed in
// Two run on the normal lane while the cold one is still held;
// the other two wait behind them there, not on the cold lane.
for range 2 {
<-rig.started
}
close(rig.gate)
for range 4 {
if w := <-served; w.Code != http.StatusFound {
t.Fatalf("FAIL: a warm request: status %d %s", w.Code, w.Body.String())
}
}
if w := <-first; w.Code != http.StatusFound {
t.Fatalf("FAIL: the first cold request: status %d", w.Code)
}
if st := rig.d.ArchiveLane.Stats(); st.Admitted != 4 || st.Shed != 0 {
t.Fatalf("FAIL: normal lane %+v, want 4 admitted", st)
}
if st := rig.d.ArchiveColdLane.Stats(); st.Admitted != 1 {
t.Fatalf("FAIL: cold lane %+v after the warm requests, want still 1 admitted", st)
}
t.Log("ok: cold anonymous on the small lane and shed when full; tips, tags and readers on the normal one")
}
// A private repository refuses a reader who may not see it before
// anything is built or asked of the bucket.
func TestPrivateArchiveRefusesBeforeAnyBuild(t *testing.T) {
rig := newArchiveRig(t, func(d *Deps, _ *memBuckets, _ string) {
d.CanRead = func(ctx context.Context, repo string) bool { return ops.User(ctx) == "dan" }
})
w := rig.get(context.Background(), arcTip, "tar")
if w.Code != http.StatusNotFound {
t.Fatalf("FAIL: status %d, want 404", w.Code)
}
if rig.builds.Load() != 0 || len(rig.objectKeys()) != 0 {
t.Fatalf("FAIL: builds=%d objects=%v for a refused reader", rig.builds.Load(), rig.objectKeys())
}
if w := rig.get(signedIn("dan"), arcTip, "tar"); w.Code != http.StatusFound {
t.Fatalf("FAIL: a permitted reader: status %d", w.Code)
}
t.Log("ok: refused with nothing built; the permitted reader is redirected")
}
// With no bucket anywhere the archive is built and served as before,
// through the same bounded queue.
func TestArchiveWithoutABucketIsServedThroughTheQueue(t *testing.T) {
rig := newArchiveRig(t, func(d *Deps, store *memBuckets, _ string) { store.UnbindBucket("test") })
w := rig.get(context.Background(), arcTip, "zip")
if w.Code != http.StatusOK || w.Body.String() != "zip of "+arcTip {
t.Fatalf("FAIL: status %d body %q", w.Code, w.Body.String())
}
if st := rig.d.ArchiveLane.Stats(); st.Admitted != 1 {
t.Fatalf("FAIL: normal lane %+v, want the build admitted through it", st)
}
w = rig.get(context.Background(), arcMid, "zip")
if w.Code != http.StatusOK {
t.Fatalf("FAIL: cold, no bucket: status %d", w.Code)
}
if st := rig.d.ArchiveColdLane.Stats(); st.Admitted != 1 {
t.Fatalf("FAIL: cold lane %+v, want the cold build admitted through it", st)
}
t.Log("ok: built per request, through the lanes")
}
// A repository with no bucket of its own keeps its archives in the
// forge's.
func TestArchiveFallsBackToTheForgeBucket(t *testing.T) {
rig := newArchiveRig(t, func(d *Deps, store *memBuckets, endpoint string) {
store.UnbindBucket("test")
store.BindBucket(oltp.ForgeBucket, oltp.BucketBinding{Endpoint: endpoint, Bucket: "forge", Region: "auto", AccessKey: "AK", Secret: "s3cret"})
})
if w := rig.get(context.Background(), arcTip, "tar"); w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d", w.Code)
}
if keys := rig.objectKeys(); len(keys) != 1 || !strings.HasPrefix(keys[0], "forge/archive/test/") {
t.Fatalf("FAIL: objects %v", keys)
}
t.Log("ok: built into the forge's bucket")
}
// A check-in this repository does not hold is not served from the
// bucket, even when an object for that hash is there: the bucket is
// asked only about check-ins the index has.
func TestArchiveOfAnUnknownCheckinIsNotRedirected(t *testing.T) {
rig := newArchiveRig(t, nil)
ghost := strings.Repeat("ee", 32)
rig.put("files/" + archiveObjectKey("test", arcProducer, ghost, "tar"))
w := rig.get(context.Background(), ghost, "tar")
if w.Code == http.StatusFound {
t.Fatalf("FAIL: a check-in the repository does not hold was redirected to %s", w.Header().Get("Location"))
}
t.Logf("ok: status %d", w.Code)
}
// Deleting a repository removes its archives from the bucket, and only
// its own.
func TestDroppingArchivesRemovesOnlyThatRepository(t *testing.T) {
rig := newArchiveRig(t, nil)
if w := rig.get(context.Background(), arcTip, "tar"); w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d", w.Code)
}
rig.put("files/archive/test2/zit-0.3.30/" + arcTip + ".tar.gz")
b, ok := rig.d.archiveBucket("test")
if !ok {
t.Fatal("FAIL: no archive bucket")
}
if err := rig.d.dropArchives(context.Background(), "test", b); err != nil {
t.Fatalf("FAIL: %v", err)
}
if keys := rig.objectKeys(); len(keys) != 1 || !strings.Contains(keys[0], "/test2/") {
t.Fatalf("FAIL: objects after drop %v", keys)
}
// The process forgot it too: the next request builds again.
if w := rig.get(context.Background(), arcTip, "tar"); w.Code != http.StatusFound || rig.builds.Load() != 2 {
t.Fatalf("FAIL: after drop: status %d builds %d", w.Code, rig.builds.Load())
}
t.Log("ok")
}
// Archives an older zit built are swept from a repository's prefix in
// the background, after the first build for it in this process: the
// current producer's stay, another repository's are never touched, and
// a run deletes at most its cap.
func TestOldProducersArchivesAreSwept(t *testing.T) {
rig := newArchiveRig(t, nil)
put := func(key string) { rig.put("files/" + key) }
put("archive/test/zit-0.3.29/" + arcMid + ".zip")
put("archive/test/zit-0.3.28/" + arcMid + ".tar.gz")
put("archive/test/" + arcMid + ".zip") // written with no producer
put("archive/test/zit-0.3.30/" + arcV1 + ".zip")
put("archive/test2/zit-0.3.29/" + arcMid + ".zip")
put("unversioned/test/zit-0.3.29/file")
b, _ := rig.d.archiveBucket("test")
n, err := rig.d.sweepArchives(context.Background(), "test", b, 2)
if err != nil || n != 2 {
t.Fatalf("FAIL: first run deleted %d (err %v), want the cap of 2", n, err)
}
if w := rig.get(context.Background(), arcTip, "tar"); w.Code != http.StatusFound {
t.Fatalf("FAIL: status %d", w.Code)
}
// The build starts the sweep; it runs in the background.
want := map[string]bool{
"files/archive/test/zit-0.3.30/" + arcV1 + ".zip": true,
"files/archive/test/zit-0.3.30/" + arcTip + ".tar.gz": true,
"files/archive/test2/zit-0.3.29/" + arcMid + ".zip": true,
"files/unversioned/test/zit-0.3.29/file": true,
}
deadline := time.Now().Add(5 * time.Second)
for {
keys := rig.objectKeys()
ok := len(keys) == len(want)
for _, k := range keys {
ok = ok && want[k]
}
if ok {
break
}
if time.Now().After(deadline) {
t.Fatalf("FAIL: bucket holds %v, want %v", keys, want)
}
time.Sleep(20 * time.Millisecond)
}
t.Log("ok: old producers swept, current kept, other prefixes untouched")
}