anonymous_github/backend/internal/server/admin_queue_integration_test.go

package server

import (
	"context"
	"errors"
	"fmt"
	"net/http/httptest"
	"testing"
	"time"

	"go.mongodb.org/mongo-driver/v2/bson"
)

func runAdminQueueIntegration(t *testing.T, base *Server) {
	t.Run("queue controls preserve job identities and group expiry", func(t *testing.T) {
		ctx := context.Background()
		db := base.Store.DB.Client().Database("admin_queue_" + bson.NewObjectID().Hex())
		defer db.Drop(ctx)
		s := &Server{Store: &Store{DB: db, Redis: base.Store.Redis}}
		if err := s.Store.ensureIndexes(ctx); err != nil {
			t.Fatal(err)
		}
		call := func(method, queue, operation string) (any, error) {
			return s.adminQueues(&request{Server: s, R: httptest.NewRequest(method, "/?queue="+queue, nil), Params: map[string]string{"name": queue, "operation": operation}})
		}
		expectError := func(method, queue, operation, code string, status int) {
			t.Helper()
			_, err := call(method, queue, operation)
			var api *APIError
			if !errors.As(err, &api) || api.Code != code || api.Status != status {
				t.Fatalf("%s %s %s: %v", method, queue, operation, err)
			}
		}
		insert := func(id, kind, action, state string) {
			t.Helper()
			_, err := db.Collection("gojobs").InsertOne(ctx, M{"_id": id, "kind": kind, "resourceId": "shared", "action": action, "state": state, "createdAt": time.Now(), "attempts": 3})
			if err != nil {
				t.Fatal(err)
			}
		}
		state := func(id, want string) {
			t.Helper()
			row, err := s.Store.one(ctx, "gojobs", M{"_id": id})
			if err != nil || (want == "" && row != nil) || (want != "" && str(row["state"]) != want) {
				t.Fatalf("%s want %s got %v: %v", id, want, row, err)
			}
		}
		insert("repo:shared:remove", "repo", "remove", "failed")
		insert("gist:shared:expire", "gist", "expire", "failed")
		insert("pr:shared:prepare", "pr", "prepare", "failed")
		insert("repo:shared:cache", "repo", "cache", "active")
		_, err := db.Collection(collection("pr")).InsertOne(ctx, M{idKey("pr"): "shared", "status": "error", "source": M{"provider": "fixture"}})
		if err != nil {
			t.Fatal(err)
		}
		output, err := call("GET", "remove", "")
		if err != nil {
			t.Fatal(err)
		}
		result := output.(M)
		jobs := result["jobs"].([]M)
		if len(jobs) != 2 {
			t.Fatalf("expiry absent from remove queue: %v", jobs)
		}
		for _, queue := range result["queues"].([]M) {
			if queue["key"] == "remove" && number(obj(queue["counts"])["failed"]) != 2 {
				t.Fatal("expiry missing from counts", queue)
			}
		}
		for _, method := range []string{"POST", "DELETE"} {
			expectError(method, "download", "absent", "job_not_found", 404)
			expectError(method, "download", "gist:shared:expire", "job_not_found", 404)
			expectError(method, "unknown", "absent", "queue_not_found", 404)
		}
		expectError("POST", "cache", "repo:shared:cache", "job_is_active", 409)
		if _, err := call("POST", "download", "pr:shared:prepare"); err != nil {
			t.Fatal(err)
		}
		state("pr:shared:prepare", "waiting")
		resource, err := s.Store.one(ctx, collection("pr"), M{idKey("pr"): "shared"})
		if err != nil || resource["status"] != "preparing" {
			t.Fatal("document retry did not prepare its resource", resource, err)
		}
		if count, err := db.Collection("gojobs").CountDocuments(ctx, M{}); err != nil || count != 4 {
			t.Fatal("single retry created a different job", count, err)
		}
		output, err = call("POST", "remove", "retry-failed")
		if err != nil || number(obj(output)["total"]) != 2 || number(obj(output)["retried"]) != 2 {
			t.Fatal("remove retry omitted expiry", output, err)
		}
		state("gist:shared:expire", "waiting")
		state("repo:shared:remove", "waiting")
		insert("expiry-delayed", "gist", "expire", "delayed")
		if _, err := call("POST", "remove", "drain"); err != nil {
			t.Fatal(err)
		}
		state("gist:shared:expire", "")
		state("expiry-delayed", "")
		state("repo:shared:remove", "")
		state("pr:shared:prepare", "waiting")
		insert("expiry-completed", "gist", "expire", "completed")
		insert("expiry-active", "gist", "expire", "active")
		if _, err := call("POST", "remove", "empty"); err != nil {
			t.Fatal(err)
		}
		state("expiry-completed", "")
		state("expiry-active", "")
		insert("expiry-active", "gist", "expire", "active")
		if _, err := call("DELETE", "remove", "expiry-active"); err != nil {
			t.Fatal(err)
		}
		state("expiry-active", "")
		insert("empty", "gist", "expire", "failed")
		if _, err := call("DELETE", "remove", "empty"); err != nil {
			t.Fatal(err)
		}
		state("empty", "")
		state("pr:shared:prepare", "waiting")
		// Removing an active job must not let its old worker finish a replacement.
		if _, err := call("DELETE", "cache", "repo:shared:cache"); err != nil {
			t.Fatal(err)
		}
		if err := s.enqueue(ctx, "repo", "shared", "cache"); err != nil {
			t.Fatal(err)
		}
		changed, err := s.Store.update(ctx, "gojobs", M{"_id": "repo:shared:cache", "lease": "old-worker"}, M{"$set": M{"state": "completed"}})
		if err != nil || changed {
			t.Fatal("old worker changed replacement", err)
		}
		state("repo:shared:cache", "waiting")

		if _, err := db.Collection("gojobs").DeleteMany(ctx, M{}); err != nil {
			t.Fatal(err)
		}
		rows := []any{}
		states := []string{"active", "waiting", "delayed", "failed", "completed"}
		for _, status := range states {
			for i := 0; i < 201; i++ {
				rows = append(rows, M{"_id": fmt.Sprintf("%s-%03d", status, i), "kind": "repo", "resourceId": "resource-name", "action": "prepare", "state": status, "createdAt": time.Unix(int64(i), 0)})
			}
		}
		if _, err := db.Collection("gojobs").InsertMany(ctx, rows); err != nil {
			t.Fatal(err)
		}
		output, err = call("GET", "unknown", "")
		if err != nil {
			t.Fatal(err)
		}
		result = output.(M)
		jobs = result["jobs"].([]M)
		if result["selectedQueue"] != "download" || len(jobs) != 1000 {
			t.Fatal("queue fallback or per-state bounds differ", result["selectedQueue"], len(jobs))
		}
		for i, job := range jobs {
			if job["_state"] != states[i/200] {
				t.Fatal("job priority order differs", i, job)
			}
		}
		output, err = s.adminQueues(&request{Server: s, R: httptest.NewRequest("GET", "/?search=ACTIVE-200", nil), Params: map[string]string{}})
		if err != nil {
			t.Fatal(err)
		}
		jobs = output.(M)["jobs"].([]M)
		if len(jobs) != 1 || jobs[0]["id"] != "active-200" {
			t.Fatal("job ID search differs", jobs)
		}
	})
}