feat(api): 集群列表读改走 informer 缓存,审核队列、我的提交与备份列表改为服务端分页筛选并一次批量查构建,面板三页跟进

This commit is contained in:
Lemon-miaow committed 2026-09-25 18:13:21 +08:00
1 parent 0a66385d9f
commit 1054e62fa9
41 files changed
+1642 -295

No files matched your search

+23 -14
View File
@@ -63,7 +63,8 @@ type fakeRepo struct {
// the value copied from the consumed code at verify (migration 0005).
linkAuthSource map[string]string
// world backups (spec §7, §22). A nil slice lists empty.
backups []fakeBackup
backups []fakeBackup
backupListOpts BackupListOpts // the last AllBackups or BackupsForUser options
// session auth (spec §B, passwordless). staff is keyed by username (the login key);
// sessions by token_hash; settings by key. They mirror the PG contract so the
// hermetic tests exercise the same fail-closed semantics the integration impl
@@ -912,23 +913,31 @@ func (f *fakeRepo) BackupStoreBytes(context.Context) (int64, error) {
// the hermetic tests can't pass against a too-lenient fake: only status='present'
// rows are visible, the user scope is the former_owner column, and LatestBackup
// is the newest present row for a server (or ErrNotFound).
func (f *fakeRepo) AllBackups(_ context.Context) ([]BackupView, error) {
var out []BackupView
for _, b := range f.backups {
if b.view.Status == "present" {
out = append(out, b.view)
}
}
return out, nil
func (f *fakeRepo) AllBackups(_ context.Context, opts BackupListOpts) ([]BackupView, int, error) {
return f.pageBackups("", opts)
}
func (f *fakeRepo) BackupsForUser(_ context.Context, userID string) ([]BackupView, error) {
var out []BackupView
func (f *fakeRepo) BackupsForUser(_ context.Context, userID string, opts BackupListOpts) ([]BackupView, int, error) {
if userID == "" { // an empty id is no former owner; "" below means every owner
return nil, 0, nil
}
return f.pageBackups(userID, opts)
}
// pageBackups filters like the PG query (present, in scope, on the server) and
// pages newest first, recording the options it was asked for.
func (f *fakeRepo) pageBackups(owner string, opts BackupListOpts) ([]BackupView, int, error) {
f.backupListOpts = opts
var match []BackupView
for _, b := range f.backups {
if b.view.Status == "present" && b.view.FormerOwner == userID {
out = append(out, b.view)
if b.view.Status == "present" && (owner == "" || b.view.FormerOwner == owner) &&
(opts.Server == "" || b.view.ServerName == opts.Server) {
match = append(match, b.view)
}
}
return out, nil
sort.SliceStable(match, func(i, j int) bool { return match[i].CreatedAt.After(match[j].CreatedAt) })
lo := min(opts.Offset, len(match))
hi := min(lo+opts.Limit, len(match))
return match[lo:hi], len(match), nil
}
func (f *fakeRepo) LatestBackup(_ context.Context, serverName string) (*BackupRecord, error) {
var latest *fakeBackup
+3 -2
View File
@@ -84,8 +84,9 @@ type ServerSpecPatch struct {
// so handlers are tested against a fake; the controller-runtime implementation
// (k8sCluster) is integration-tested only — it requires a live cluster.
type Cluster interface {
// Ping verifies the K8s API and CRD informer are healthy — used by /readyz
// (spec §7) to confirm the lifecycle store is reachable and synced.
// Ping verifies the K8s API is reachable and the MinecraftServer cache has
// synced — used by /readyz (spec §7), so a replica serves the fleet reads only
// once it holds the whole fleet.
Ping(ctx context.Context) error
// GetServer reads one MinecraftServer's lifecycle view, or ErrNotFound.
+26 -3
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"net/http"
"strconv"
"strings"
"time"
@@ -29,17 +30,39 @@ func errNoWorldVolume() error {
// query runs (AllBackups vs BackupsForUser) — there is no client-supplied filter
// a user could widen, so "a user cannot see another's backups" is a property of
// the query, not of request parsing.
//
// What the request may choose is the page: ?server= narrows the list to one
// server (inside the caller's scope, never beyond it), ?limit= and ?offset= page
// it, and the answer carries how many match in all.
func (a *API) handleListBackups(w http.ResponseWriter, r *http.Request) {
p := principalFromContext(r.Context())
q := r.URL.Query()
opts := BackupListOpts{Server: q.Get("server")}
// Format only: a system server's reserved name is still a server whose
// backups an admin may list.
if opts.Server != "" {
if err := naming.ValidateSystemServerName(opts.Server); err != nil {
writeError(w, r, newError(http.StatusBadRequest, "bad_name", "invalid server name: %v", err))
return
}
}
opts.Limit, _ = strconv.Atoi(q.Get("limit"))
opts.Offset, _ = strconv.Atoi(q.Get("offset"))
if opts.Limit <= 0 {
opts.Limit = DefaultBackupListLimit
}
opts.Limit = min(opts.Limit, MaxBackupListLimit)
opts.Offset = max(opts.Offset, 0)
var (
backups []BackupView
total int
err error
)
if p.IsAdmin() {
backups, err = a.Repo.AllBackups(r.Context())
backups, total, err = a.Repo.AllBackups(r.Context(), opts)
} else {
backups, err = a.Repo.BackupsForUser(r.Context(), p.UserID)
backups, total, err = a.Repo.BackupsForUser(r.Context(), p.UserID, opts)
}
if err != nil {
writeError(w, r, err)
@@ -48,7 +71,7 @@ func (a *API) handleListBackups(w http.ResponseWriter, r *http.Request) {
if backups == nil {
backups = []BackupView{}
}
writeJSON(w, http.StatusOK, map[string]any{"backups": backups})
writeJSON(w, http.StatusOK, map[string]any{"backups": backups, "total": total})
}
// handleRestoreBackup starts restoring a server's world from a backup (spec §7
+70
View File
@@ -83,6 +83,76 @@ func TestListBackups(t *testing.T) {
}
})
admin := &Principal{UserID: "admin1", Role: "admin", ViaAdminAccess: true}
total := func(t *testing.T, w *httptest.ResponseRecorder) int {
t.Helper()
var resp struct {
Total *int `json:"total"`
}
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil || resp.Total == nil {
t.Fatalf("body carries no total: %v (%s)", err, w.Body.String())
}
return *resp.Total
}
t.Run("server filter narrows inside the scope, never past it", func(t *testing.T) {
api := mk()
api.External = staticExternal{p: admin}
w := do(api.ExternalHandler(), "GET", "/api/v1/backups?server=beta", "", nil)
if got := ids(list(t, w)); !got["b2"] || len(got) != 1 || total(t, w) != 1 {
t.Fatalf("admin ?server=beta = %v (total %d), want {b2} of 1", got, total(t, w))
}
api.External = staticExternal{p: &Principal{UserID: "owner1", Role: "user"}}
w = do(api.ExternalHandler(), "GET", "/api/v1/backups?server=beta", "", nil)
if got := ids(list(t, w)); len(got) != 0 || total(t, w) != 0 {
t.Fatalf("owner1 ?server=beta = %v (total %d), want nothing: beta's backup is owner2's", got, total(t, w))
}
})
t.Run("pages newest first with the match total", func(t *testing.T) {
api := mk()
api.External = staticExternal{p: admin}
w := do(api.ExternalHandler(), "GET", "/api/v1/backups?limit=1", "", nil)
if vs := list(t, w); len(vs) != 1 || vs[0].ID != "b2" || total(t, w) != 2 {
t.Fatalf("?limit=1 = %+v (total %d), want [b2] of 2", vs, total(t, w))
}
w = do(api.ExternalHandler(), "GET", "/api/v1/backups?limit=1&offset=1", "", nil)
if vs := list(t, w); len(vs) != 1 || vs[0].ID != "b1" {
t.Fatalf("?limit=1&offset=1 = %+v, want [b1]", vs)
}
})
t.Run("page bounds", func(t *testing.T) {
cases := []struct {
query string
want BackupListOpts
}{
{"", BackupListOpts{Limit: 20}},
{"?limit=5000&offset=-3", BackupListOpts{Limit: 100}},
{"?limit=x&offset=40&server=alpha", BackupListOpts{Server: "alpha", Limit: 20, Offset: 40}},
}
for _, c := range cases {
repo := newFakeRepo()
api := newTestAPI(repo, newFakeCluster())
api.External = staticExternal{p: admin}
if w := do(api.ExternalHandler(), "GET", "/api/v1/backups"+c.query, "", nil); w.Code != http.StatusOK {
t.Fatalf("%q: code = %d (%s)", c.query, w.Code, w.Body.String())
}
if repo.backupListOpts != c.want {
t.Errorf("%q asked the repo for %+v, want %+v", c.query, repo.backupListOpts, c.want)
}
}
})
t.Run("malformed server name is a 400", func(t *testing.T) {
api := mk()
api.External = staticExternal{p: admin}
w := do(api.ExternalHandler(), "GET", "/api/v1/backups?server=Bad%20Name", "", nil)
if w.Code != http.StatusBadRequest || !strings.Contains(w.Body.String(), "bad_name") {
t.Fatalf("code = %d (%s), want 400 bad_name", w.Code, w.Body.String())
}
})
t.Run("backup_ref never serialized", func(t *testing.T) {
api := mk()
api.External = staticExternal{p: &Principal{UserID: "admin1", Role: "admin", ViaAdminAccess: true}}
+6 -5
View File
@@ -17,11 +17,12 @@ import (
// admin-tier (Zero Trust), enforced by adminOnly before these handlers run.
type ImageBuilder interface {
Submit(ctx context.Context, req build.Request) (*build.Build, error)
// Get reads one build row without reconciling it — the read-only lookup the
// submission views use to surface a build's outcome to its submitter (the
// /images/build routes are admin-tier). State advance belongs to the
// reconcile loop (Sync/SyncAll), so a list render never touches the cluster.
Get(ctx context.Context, id string) (*build.Build, error)
// GetMany reads the build rows among ids without reconciling them, keyed by
// id — the one read-only lookup a submission list makes to surface each
// build's outcome to its submitter (the /images/build routes are admin-tier).
// State advance belongs to the reconcile loop (Sync/SyncAll), so a list render
// never touches the cluster. An id with no row is absent from the map.
GetMany(ctx context.Context, ids []string) (map[string]build.Build, error)
// Sync reconciles a build against its Job and returns the current view, so a
// GET doubles as the reconcile tick (idempotent on terminal builds).
Sync(ctx context.Context, id string) (*build.Build, error)
+9 -5
View File
@@ -18,6 +18,7 @@ type fakeBuilder struct {
submitErr error
getBuilds map[string]*build.Build
getErr error
getManyIDs [][]string // one entry per GetMany call
syncErr error
cancelErr error
addedRef string
@@ -46,15 +47,18 @@ func (f *fakeBuilder) Submit(_ context.Context, req build.Request) (*build.Build
RequestedBy: req.RequestedBy}, nil
}
func (f *fakeBuilder) Get(_ context.Context, id string) (*build.Build, error) {
f.lastBuildID = id
func (f *fakeBuilder) GetMany(_ context.Context, ids []string) (map[string]build.Build, error) {
f.getManyIDs = append(f.getManyIDs, ids)
if f.getErr != nil {
return nil, f.getErr
}
if b, ok := f.getBuilds[id]; ok {
return b, nil
out := map[string]build.Build{}
for _, id := range ids {
if b, ok := f.getBuilds[id]; ok {
out[id] = *b
}
}
return nil, build.ErrNotFound
return out, nil
}
func (f *fakeBuilder) Sync(_ context.Context, id string) (*build.Build, error) {
+54 -2
View File
@@ -23,19 +23,64 @@ import (
// — the app-tier desiredState lever, the admin-tier create, and the admin-tier
// spec patch — each via a merge patch. It is integration-tested against a live
// cluster, not the hermetic api_test.go suite.
//
// The fleet-wide reads (ListServers, GetBySubdomain) come from servers, an
// informer cache of MinecraftServers, when one is wired (WithServerCache): velocity's
// registration pull, the fleet page and every wake's running-cap count would each
// be a full List against the apiserver otherwise. Everything that reads one server
// to act on it — GetServer, and the read-modify-write behind every patch — stays on
// the direct client c, so a write never works from a copy the watch has not caught
// up with yet.
type K8sCluster struct {
c client.Client
namespace string
// servers serves the fleet-wide reads; nil means c.
servers client.Reader
// synced reports whether servers has its first full list; nil means no cache.
synced func() bool
// now is injectable for the maintenance-lock tests; nil means time.Now.
now func() time.Time
}
// SubdomainIndex is the field index GetBySubdomain matches on in the server cache.
// The cache must register it with SubdomainOf.
const SubdomainIndex = "spec.subdomain"
// SubdomainOf is the SubdomainIndex extractor.
func SubdomainOf(o client.Object) []string {
ms, ok := o.(*v1alpha1.MinecraftServer)
if !ok || ms.Spec.Subdomain == "" {
return nil
}
return []string{ms.Spec.Subdomain}
}
// NewK8sCluster builds a Cluster over c, scoped to namespace.
func NewK8sCluster(c client.Client, namespace string) *K8sCluster {
return &K8sCluster{c: c, namespace: namespace}
}
// WithServerCache serves ListServers and GetBySubdomain from servers, an informer
// cache of MinecraftServers in the namespace that indexes SubdomainIndex. synced
// reports whether its first list has landed; Ping fails until it has.
func (k *K8sCluster) WithServerCache(servers client.Reader, synced func() bool) *K8sCluster {
k.servers, k.synced = servers, synced
return k
}
func (k *K8sCluster) fleet() client.Reader {
if k.servers != nil {
return k.servers
}
return k.c
}
// Ping checks the apiserver with a one-item list on the direct client and, when a
// server cache is wired, that its informer has synced.
func (k *K8sCluster) Ping(ctx context.Context) error {
if k.synced != nil && !k.synced() {
return errors.New("MinecraftServer cache has not synced")
}
var list v1alpha1.MinecraftServerList
return k.c.List(ctx, &list, client.InNamespace(k.namespace), client.Limit(1))
}
@@ -87,9 +132,16 @@ func (k *K8sCluster) PodImages(ctx context.Context) ([]string, error) {
return images, nil
}
// GetBySubdomain looks the subdomain up in the cache's index. Without a cache it
// lists the namespace, since a CRD field selector is not something the apiserver
// serves.
func (k *K8sCluster) GetBySubdomain(ctx context.Context, subdomain string) (*ServerInfo, error) {
var list v1alpha1.MinecraftServerList
if err := k.c.List(ctx, &list, client.InNamespace(k.namespace)); err != nil {
opts := []client.ListOption{client.InNamespace(k.namespace)}
if k.servers != nil {
opts = append(opts, client.MatchingFields{SubdomainIndex: subdomain})
}
if err := k.fleet().List(ctx, &list, opts...); err != nil {
return nil, err
}
for i := range list.Items {
@@ -102,7 +154,7 @@ func (k *K8sCluster) GetBySubdomain(ctx context.Context, subdomain string) (*Ser
func (k *K8sCluster) ListServers(ctx context.Context) ([]ServerInfo, error) {
var list v1alpha1.MinecraftServerList
if err := k.c.List(ctx, &list, client.InNamespace(k.namespace)); err != nil {
if err := k.fleet().List(ctx, &list, client.InNamespace(k.namespace)); err != nil {
return nil, err
}
out := make([]ServerInfo, 0, len(list.Items))
+121
View File
@@ -2,12 +2,17 @@ package api
import (
"context"
"errors"
"sort"
"strings"
"testing"
"felis.lolicon.best/internal/apis/felis/v1alpha1"
"felis.lolicon.best/internal/naming"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
)
@@ -132,3 +137,119 @@ func TestPatchIdleStopKeepsTheChoiceVisible(t *testing.T) {
t.Fatalf("view idleStopSeconds = %d, want 1800", info.IdleStopSeconds)
}
}
// spyReader records the options of every List it serves.
type spyReader struct {
client.Reader
lists []*client.ListOptions
}
func (s *spyReader) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
lo := &client.ListOptions{}
lo.ApplyOptions(opts)
s.lists = append(s.lists, lo)
return s.Reader.List(ctx, list, opts...)
}
func testServer(name, subdomain string) *v1alpha1.MinecraftServer {
return &v1alpha1.MinecraftServer{
ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "minecraft"},
Spec: v1alpha1.MinecraftServerSpec{Subdomain: subdomain, DesiredState: v1alpha1.DesiredRunning},
}
}
// The fleet reads come from the informer cache, the subdomain lookup through its
// index, and everything that reads one server to act on it from the apiserver. The
// two fakes hold different fleets so each read shows which one it asked.
func TestFleetReadsComeFromTheServerCache(t *testing.T) {
ctx := context.Background()
scheme := runtime.NewScheme()
if err := v1alpha1.AddToScheme(scheme); err != nil {
t.Fatalf("scheme: %v", err)
}
direct := fake.NewClientBuilder().WithScheme(scheme).WithObjects(testServer("fresh", "fresh")).Build()
cached := fake.NewClientBuilder().WithScheme(scheme).
WithObjects(testServer("alpha", "alpha"), testServer("beta", "beta")).
WithIndex(&v1alpha1.MinecraftServer{}, SubdomainIndex, SubdomainOf).Build()
spy := &spyReader{Reader: cached}
synced := false
k := NewK8sCluster(direct, "minecraft").WithServerCache(spy, func() bool { return synced })
if err := k.Ping(ctx); err == nil || err.Error() != "MinecraftServer cache has not synced" {
t.Fatalf("Ping before the cache synced = %v, want the not-synced error", err)
}
synced = true
if err := k.Ping(ctx); err != nil {
t.Fatalf("Ping once synced: %v", err)
}
infos, err := k.ListServers(ctx)
if err != nil {
t.Fatalf("ListServers: %v", err)
}
var names []string
for _, i := range infos {
names = append(names, i.Name)
}
sort.Strings(names)
if got := strings.Join(names, ","); got != "alpha,beta" {
t.Fatalf("ListServers = %s, want alpha,beta (the cache's fleet)", got)
}
info, err := k.GetBySubdomain(ctx, "beta")
if err != nil || info.Name != "beta" {
t.Fatalf("GetBySubdomain(beta) = %+v, %v; want beta", info, err)
}
if got := spy.lists[len(spy.lists)-1].FieldSelector.String(); got != "spec.subdomain=beta" {
t.Fatalf("subdomain lookup selector = %q, want spec.subdomain=beta (served by the index)", got)
}
if _, err := k.GetBySubdomain(ctx, "fresh"); !errors.Is(err, ErrNotFound) {
t.Fatalf("GetBySubdomain(fresh) = %v, want ErrNotFound (only the apiserver has it)", err)
}
if info, err := k.GetServer(ctx, "fresh"); err != nil || info.Name != "fresh" {
t.Fatalf("GetServer(fresh) = %+v, %v; want it from the apiserver", info, err)
}
if _, err := k.GetServer(ctx, "alpha"); !errors.Is(err, ErrNotFound) {
t.Fatalf("GetServer(alpha) = %v, want ErrNotFound (a single read never trusts the cache)", err)
}
if err := k.SetDesiredState(ctx, "fresh", v1alpha1.DesiredStopped); err != nil {
t.Fatalf("SetDesiredState(fresh): %v", err)
}
var ms v1alpha1.MinecraftServer
if err := direct.Get(ctx, types.NamespacedName{Namespace: "minecraft", Name: "fresh"}, &ms); err != nil {
t.Fatalf("read back: %v", err)
}
if ms.Spec.DesiredState != v1alpha1.DesiredStopped {
t.Fatalf("desiredState = %s, want Stopped", ms.Spec.DesiredState)
}
if err := k.SetDesiredState(ctx, "alpha", v1alpha1.DesiredStopped); !errors.Is(err, ErrNotFound) {
t.Fatalf("SetDesiredState(alpha) = %v, want ErrNotFound (writes read the apiserver)", err)
}
}
// Without a cache the subdomain lookup lists the namespace: the apiserver serves no
// field selector on a CRD, and the fake refuses one it has no index for.
func TestGetBySubdomainWithoutCache(t *testing.T) {
scheme := runtime.NewScheme()
if err := v1alpha1.AddToScheme(scheme); err != nil {
t.Fatalf("scheme: %v", err)
}
k := NewK8sCluster(fake.NewClientBuilder().WithScheme(scheme).
WithObjects(testServer("alpha", "alpha"), testServer("beta", "survival")).Build(), "minecraft")
info, err := k.GetBySubdomain(context.Background(), "survival")
if err != nil || info.Name != "beta" {
t.Fatalf("GetBySubdomain(survival) = %+v, %v; want beta", info, err)
}
if err := k.Ping(context.Background()); err != nil {
t.Fatalf("Ping with no cache: %v", err)
}
}
func TestSubdomainOf(t *testing.T) {
if got := SubdomainOf(testServer("a", "survival")); len(got) != 1 || got[0] != "survival" {
t.Fatalf("SubdomainOf = %v, want [survival]", got)
}
if got := SubdomainOf(testServer("a", "")); got != nil {
t.Fatalf("SubdomainOf(no subdomain) = %v, want nil", got)
}
}
+29 -22
View File
@@ -670,32 +670,39 @@ func (p *PGRepo) SeedServer(ctx context.Context, name, subdomain string, cpuMill
return tx.Commit()
}
// AllBackups lists every present world backup, newest first (spec §7 GET
// /backups, admin scope; world_backups in §22). Only status='present' rows are
// listed — an expired or deleted backup is gone (spec §466).
func (p *PGRepo) AllBackups(ctx context.Context) ([]BackupView, error) {
const q = `SELECT id, server_name, COALESCE(former_owner, ''), COALESCE(size_bytes, 0),
reason, status, created_at, expires_at, corrupt_at IS NOT NULL, verified_at, skipped_entries
FROM world_backups WHERE status = 'present' ORDER BY created_at DESC`
rows, err := p.db.QueryContext(ctx, q)
if err != nil {
return nil, err
}
return scanBackupViews(rows)
// AllBackups lists one page of every present world backup, newest first (spec
// §7 GET /backups, admin scope; world_backups in §22). Only status='present' rows
// are listed — an expired or deleted backup is gone (spec §466).
func (p *PGRepo) AllBackups(ctx context.Context, opts BackupListOpts) ([]BackupView, int, error) {
return p.pageBackups(ctx, ``, opts)
}
// BackupsForUser lists the present world backups of worlds the user formerly
// owned, newest first (spec §7 GET /backups, former_owner scope). A NULL
// BackupsForUser lists one page of the present world backups of worlds the user
// formerly owned, newest first (spec §7 GET /backups, former_owner scope). A NULL
// former_owner never matches a user id, so orphaned backups stay admin-only.
func (p *PGRepo) BackupsForUser(ctx context.Context, userID string) ([]BackupView, error) {
const q = `SELECT id, server_name, COALESCE(former_owner, ''), COALESCE(size_bytes, 0),
reason, status, created_at, expires_at, corrupt_at IS NOT NULL, verified_at, skipped_entries
FROM world_backups WHERE status = 'present' AND former_owner = $1 ORDER BY created_at DESC`
rows, err := p.db.QueryContext(ctx, q, userID)
if err != nil {
return nil, err
func (p *PGRepo) BackupsForUser(ctx context.Context, userID string, opts BackupListOpts) ([]BackupView, int, error) {
return p.pageBackups(ctx, ` AND former_owner = $2`, opts, userID)
}
// pageBackups counts and reads one page of present backups under the caller's
// scope clause, whose parameters follow the server filter ($1). id breaks
// created_at ties so a page boundary never repeats or skips a row.
func (p *PGRepo) pageBackups(ctx context.Context, scope string, opts BackupListOpts, scopeArgs ...any) ([]BackupView, int, error) {
match := ` FROM world_backups WHERE status = 'present' AND ($1::text = '' OR server_name = $1::text)` + scope
args := append([]any{opts.Server}, scopeArgs...)
var total int
if err := p.db.QueryRowContext(ctx, `SELECT count(*)`+match, args...).Scan(&total); err != nil {
return nil, 0, err
}
return scanBackupViews(rows)
n := len(args)
rows, err := p.db.QueryContext(ctx, fmt.Sprintf(`SELECT id, server_name, COALESCE(former_owner, ''), COALESCE(size_bytes, 0),
reason, status, created_at, expires_at, corrupt_at IS NOT NULL, verified_at, skipped_entries%s
ORDER BY created_at DESC, id DESC LIMIT $%d OFFSET $%d`, match, n+1, n+2), append(args, opts.Limit, opts.Offset)...)
if err != nil {
return nil, 0, err
}
out, err := scanBackupViews(rows)
return out, total, err
}
// scanBackupViews drains a world_backups result set into BackupViews. backup_ref
+26 -8
View File
@@ -95,6 +95,22 @@ type BackupView struct {
SkippedEntries int `json:"skipped_entries,omitempty"`
}
// BackupListOpts selects a page of the backups list. Server narrows it to one
// server's backups (the server's Backups page); empty lists every server in the
// caller's scope. The scope itself is never an option: it is which Repo method
// runs.
type BackupListOpts struct {
Server string
Limit int
Offset int
}
// DefaultBackupListLimit and MaxBackupListLimit bound one page of backups.
const (
DefaultBackupListLimit = 20
MaxBackupListLimit = 100
)
// BackupRecord is the server-side view of a backup used to drive a restore (spec
// §466). It carries the opaque backup_ref the Restorer needs and the former_owner
// the restore authorization compares against — neither is ever serialized to the
@@ -338,14 +354,16 @@ type Repo interface {
// cannot be claimed. The cockpit treats it as best-effort: a lookup error leaves
// ownership unknown rather than failing the fleet read.
ServerOwners(ctx context.Context) (map[string]ServerOwnership, error)
// AllBackups lists every present world backup, newest first (spec §7 GET
// /backups, admin scope). Expired/deleted rows are never returned.
AllBackups(ctx context.Context) ([]BackupView, error)
// BackupsForUser lists the present world backups of worlds the user formerly
// owned, newest first (spec §7 GET /backups, former_owner scope). The
// former_owner column is stamped when the world is archived at release time, so
// a user sees their own released worlds even after the server is re-seeded.
BackupsForUser(ctx context.Context, userID string) ([]BackupView, error)
// AllBackups lists one page of every present world backup, newest first, and
// how many match in all (spec §7 GET /backups, admin scope). Expired/deleted
// rows are never returned.
AllBackups(ctx context.Context, opts BackupListOpts) ([]BackupView, int, error)
// BackupsForUser lists one page of the present world backups of worlds the
// user formerly owned, newest first, and how many match in all (spec §7 GET
// /backups, former_owner scope). The former_owner column is stamped when the
// world is archived at release time, so a user sees their own released worlds
// even after the server is re-seeded.
BackupsForUser(ctx context.Context, userID string, opts BackupListOpts) ([]BackupView, int, error)
// LatestBackup returns the most recent present backup for a server, or
// ErrNotFound when none exists (spec §466 restore). The returned BackupRecord
// carries the server-side backup_ref + former_owner the restore path needs; the
+59 -32
View File
@@ -8,8 +8,8 @@ import (
"io"
"log"
"net/http"
"strconv"
"felis.lolicon.best/internal/build"
"felis.lolicon.best/internal/submit"
)
@@ -36,10 +36,12 @@ type SubmissionService interface {
// submission at the platform-derived context ref. submittedBy is the principal,
// never the body, so a user can only upload to a submission they own.
UploadContext(ctx context.Context, id, submittedBy string, r io.Reader) (*submit.Submission, error)
// ListBy returns one user's submissions, newest first (the "my uploads" view).
ListBy(ctx context.Context, submittedBy string) ([]submit.Submission, error)
// List returns every submission, newest first (the admin review queue).
List(ctx context.Context) ([]submit.Submission, error)
// ListBy returns one page of one user's submissions, newest first (the "my
// uploads" view). The scope is submittedBy, whatever opts says.
ListBy(ctx context.Context, submittedBy string, opts submit.ListOpts) (submit.Page, error)
// List returns one page of every submission, newest first (the admin review
// queue).
List(ctx context.Context, opts submit.ListOpts) (submit.Page, error)
// Approve is the admin gate: it claims pending_review -> approved (CAS) and the
// winner starts the SAME Trivy-gated build as an admin's direct build.
// expectedDigest is the context sha256 the reviewer inspected; a row whose
@@ -206,17 +208,12 @@ func (a *API) handleMySubmissions(w http.ResponseWriter, r *http.Request) {
return
}
p := principalFromContext(r.Context())
subs, err := a.Submissions.ListBy(r.Context(), p.UserID)
page, err := a.Submissions.ListBy(r.Context(), p.UserID, submissionListOpts(r))
if err != nil {
writeSubmitError(w, r, err)
return
}
views, err := a.submissionViews(r.Context(), subs)
if err != nil {
writeSubmitError(w, r, err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"submissions": views})
a.writeSubmissionPage(w, r, page)
}
// handleListSubmissions is the admin review queue: every submission across all
@@ -228,17 +225,42 @@ func (a *API) handleListSubmissions(w http.ResponseWriter, r *http.Request) {
writeError(w, r, errSubmissionsUnavailable)
return
}
subs, err := a.Submissions.List(r.Context())
page, err := a.Submissions.List(r.Context(), submissionListOpts(r))
if err != nil {
writeSubmitError(w, r, err)
return
}
views, err := a.submissionViews(r.Context(), subs)
a.writeSubmissionPage(w, r, page)
}
// submissionListOpts reads the page a submission list asks for: ?status= (exact),
// ?query= (id, submitter or name), ?limit= and ?offset=. An unparsable number
// reads as absent and the submit layer settles the bounds. Scope is never read
// from the request: each handler supplies it.
func submissionListOpts(r *http.Request) submit.ListOpts {
q := r.URL.Query()
limit, _ := strconv.Atoi(q.Get("limit"))
offset, _ := strconv.Atoi(q.Get("offset"))
return submit.ListOpts{
Status: submit.Status(q.Get("status")), Query: q.Get("query"),
Limit: limit, Offset: offset,
}
}
// writeSubmissionPage renders one page: the enriched rows, how many match in all,
// and the scope's count in each status (every status present, zero or not, so the
// panel's summary cards never read a missing key).
func (a *API) writeSubmissionPage(w http.ResponseWriter, r *http.Request, page submit.Page) {
views, err := a.submissionViews(r.Context(), page.Submissions)
if err != nil {
writeSubmitError(w, r, err)
return
}
writeJSON(w, http.StatusOK, map[string]any{"submissions": views})
counts := map[string]int{}
for _, st := range []submit.Status{submit.StatusPendingReview, submit.StatusApproved, submit.StatusRejected} {
counts[string(st)] = page.Counts[st]
}
writeJSON(w, http.StatusOK, map[string]any{"submissions": views, "total": page.Total, "counts": counts})
}
// submissionView is one submission row enriched with its linked build's
@@ -251,29 +273,34 @@ type submissionView struct {
BuildError string `json:"build_error,omitempty"`
}
// submissionViews enriches each submission with its linked build's status via a
// read-only Builder.Get — deliberately never Sync, because the 15s reconcile
// loop owns state advance and rendering a list must not touch the cluster. A
// submission with no linked build (never approved, or approved before the
// hand-off could record the id), no Builder wired, or a build row that is gone
// (ErrNotFound) renders without the extra fields; any other store failure is
// returned so the handler reports it rather than silently dropping the outcome.
// submissionViews enriches each submission with its linked build's status via
// one read-only Builder.GetMany for the whole page — deliberately never Sync,
// because the 15s reconcile loop owns state advance and rendering a list must not
// touch the cluster. A submission with no linked build (never approved, or
// approved before the hand-off could record the id), no Builder wired, or a build
// row that is gone renders without the extra fields; a store failure is returned
// so the handler reports it rather than silently dropping the outcome.
func (a *API) submissionViews(ctx context.Context, subs []submit.Submission) ([]submissionView, error) {
views := make([]submissionView, len(subs))
var ids []string
for i, s := range subs {
views[i] = submissionView{Submission: s}
if a.Builder == nil || s.BuildID == "" {
continue
if s.BuildID != "" {
ids = append(ids, s.BuildID)
}
bld, err := a.Builder.Get(ctx, s.BuildID)
if errors.Is(err, build.ErrNotFound) {
continue
}
if a.Builder == nil || len(ids) == 0 {
return views, nil
}
builds, err := a.Builder.GetMany(ctx, ids)
if err != nil {
return nil, err
}
for i := range views {
if bld, ok := builds[views[i].BuildID]; ok {
views[i].BuildStatus = string(bld.Status)
views[i].BuildError = bld.Error
}
if err != nil {
return nil, err
}
views[i].BuildStatus = string(bld.Status)
views[i].BuildError = bld.Error
}
return views, nil
}
+117 -11
View File
@@ -34,6 +34,9 @@ type fakeSubmissions struct {
byErr error
listed []submit.Submission
listErr error
listOpts submit.ListOpts // the last List or ListBy options
pageTotal int
pageCounts map[submit.Status]int
approvedID string
approvedBy string
approveErr error
@@ -74,13 +77,14 @@ func (f *fakeSubmissions) UploadContext(_ context.Context, id, submittedBy strin
return &submit.Submission{ID: id, SubmittedBy: submittedBy, Status: submit.StatusPendingReview}, nil
}
func (f *fakeSubmissions) ListBy(_ context.Context, submittedBy string) ([]submit.Submission, error) {
f.listedBy = submittedBy
return f.byResult, f.byErr
func (f *fakeSubmissions) ListBy(_ context.Context, submittedBy string, opts submit.ListOpts) (submit.Page, error) {
f.listedBy, f.listOpts = submittedBy, opts
return submit.Page{Submissions: f.byResult, Total: f.pageTotal, Counts: f.pageCounts}, f.byErr
}
func (f *fakeSubmissions) List(_ context.Context) ([]submit.Submission, error) {
return f.listed, f.listErr
func (f *fakeSubmissions) List(_ context.Context, opts submit.ListOpts) (submit.Page, error) {
f.listOpts = opts
return submit.Page{Submissions: f.listed, Total: f.pageTotal, Counts: f.pageCounts}, f.listErr
}
func (f *fakeSubmissions) Approve(_ context.Context, id, reviewedBy, expectedDigest string) (*submit.Submission, error) {
@@ -353,12 +357,14 @@ func TestMySubmissionsScopesToPrincipal(t *testing.T) {
if fs.listedBy != "user-7" {
t.Errorf("ListBy scoped to %q, want the principal id user-7", fs.listedBy)
}
var got map[string][]submit.Submission
var got struct {
Submissions []submit.Submission `json:"submissions"`
}
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil {
t.Fatalf("body not JSON: %v", err)
}
if len(got["submissions"]) != 1 {
t.Fatalf("submissions = %d, want 1", len(got["submissions"]))
if len(got.Submissions) != 1 {
t.Fatalf("submissions = %d, want 1", len(got.Submissions))
}
}
@@ -424,6 +430,104 @@ func TestMySubmissionsBuildLookupSemantics(t *testing.T) {
}
}
// Both lists forward the page the panel asked for and answer with the match total
// and every status count — zero included, so a summary card never reads a missing
// key. A number that does not parse reads as absent (the submit layer then applies
// its default); the scope never comes from the query string.
func TestSubmissionListsForwardThePage(t *testing.T) {
fs := &fakeSubmissions{
listed: []submit.Submission{{ID: "sub-1", Status: submit.StatusApproved}},
pageTotal: 41,
pageCounts: map[submit.Status]int{submit.StatusApproved: 30, submit.StatusPendingReview: 11},
}
w := do(adminSubAPI(fs).ExternalHandler(), "GET",
"/api/v1/submissions?status=approved&query=pack&limit=5&offset=10", "", nil)
if w.Code != http.StatusOK {
t.Fatalf("code = %d, want 200 (%s)", w.Code, w.Body.String())
}
want := submit.ListOpts{Status: "approved", Query: "pack", Limit: 5, Offset: 10}
if fs.listOpts != want {
t.Fatalf("admin list forwarded %+v, want %+v", fs.listOpts, want)
}
var got struct {
Submissions []submit.Submission `json:"submissions"`
Total int `json:"total"`
Counts map[string]int `json:"counts"`
}
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil {
t.Fatalf("body not JSON: %v", err)
}
if len(got.Submissions) != 1 || got.Total != 41 {
t.Fatalf("page = %d rows, total %d; want 1 row, total 41", len(got.Submissions), got.Total)
}
wantCounts := map[string]int{"pending_review": 11, "approved": 30, "rejected": 0}
if len(got.Counts) != 3 || got.Counts["pending_review"] != 11 || got.Counts["approved"] != 30 {
t.Fatalf("counts = %v, want %v", got.Counts, wantCounts)
}
if n, ok := got.Counts["rejected"]; !ok || n != 0 {
t.Fatalf("counts = %v, want rejected present as 0", got.Counts)
}
fs = &fakeSubmissions{}
w = do(appSubAPI(fs).ExternalHandler(), "GET", "/api/v1/me/submissions?limit=many&offset=3&status=rejected", "", nil)
if w.Code != http.StatusOK {
t.Fatalf("mine: code = %d, want 200 (%s)", w.Code, w.Body.String())
}
want = submit.ListOpts{Status: "rejected", Offset: 3}
if fs.listOpts != want || fs.listedBy != "user-7" {
t.Fatalf("mine forwarded %+v for %q, want %+v for user-7", fs.listOpts, fs.listedBy, want)
}
if !strings.Contains(w.Body.String(), `"submissions":[]`) {
t.Fatalf("an empty page must render an empty array: %s", w.Body.String())
}
}
// A page's build outcomes come from one lookup naming every linked build; a page
// with no linked build asks nothing.
func TestSubmissionPageLooksUpBuildsOnce(t *testing.T) {
fs := &fakeSubmissions{listed: []submit.Submission{
{ID: "sub-1", BuildID: "bld-1"},
{ID: "sub-2"},
{ID: "sub-3", BuildID: "bld-3"},
}}
fb := &fakeBuilder{getBuilds: map[string]*build.Build{
"bld-1": {ID: "bld-1", Status: build.StatusSucceeded},
"bld-3": {ID: "bld-3", Status: build.StatusFailed, Error: "scan found a CRITICAL CVE"},
}}
api := adminSubAPI(fs)
api.Builder = fb
w := do(api.ExternalHandler(), "GET", "/api/v1/submissions", "", nil)
if w.Code != http.StatusOK {
t.Fatalf("code = %d, want 200 (%s)", w.Code, w.Body.String())
}
if len(fb.getManyIDs) != 1 || strings.Join(fb.getManyIDs[0], ",") != "bld-1,bld-3" {
t.Fatalf("build lookups = %v, want one naming bld-1,bld-3", fb.getManyIDs)
}
var got struct {
Submissions []struct {
ID string `json:"id"`
BuildStatus string `json:"build_status"`
BuildError string `json:"build_error"`
} `json:"submissions"`
}
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil {
t.Fatalf("body not JSON: %v", err)
}
if got.Submissions[0].BuildStatus != "succeeded" || got.Submissions[1].BuildStatus != "" ||
got.Submissions[2].BuildStatus != "failed" || got.Submissions[2].BuildError != "scan found a CRITICAL CVE" {
t.Fatalf("outcomes = %+v", got.Submissions)
}
fs.listed = []submit.Submission{{ID: "sub-2"}}
fb.getManyIDs = nil
if w := do(api.ExternalHandler(), "GET", "/api/v1/submissions", "", nil); w.Code != http.StatusOK {
t.Fatalf("unlinked page: code = %d", w.Code)
}
if len(fb.getManyIDs) != 0 {
t.Fatalf("a page with no linked build looked up %v", fb.getManyIDs)
}
}
// Every /submissions route is admin-tier: a plain user is rejected before the
// handler runs.
func TestSubmissionAdminRoutesAreAdminOnly(t *testing.T) {
@@ -460,12 +564,14 @@ func TestListSubmissionsAdmin(t *testing.T) {
if w.Code != http.StatusOK {
t.Fatalf("code = %d, want 200 (%s)", w.Code, w.Body.String())
}
var got map[string][]submit.Submission
var got struct {
Submissions []submit.Submission `json:"submissions"`
}
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil {
t.Fatalf("body not JSON: %v", err)
}
if len(got["submissions"]) != 2 {
t.Fatalf("submissions = %d, want 2", len(got["submissions"]))
if len(got.Submissions) != 2 {
t.Fatalf("submissions = %d, want 2", len(got.Submissions))
}
// The admin queue carries the same build outcome enrichment.
if !strings.Contains(w.Body.String(), `"build_status":"succeeded"`) {
+21
View File
@@ -198,6 +198,9 @@ type Store interface {
CreateBuild(ctx context.Context, b *Build) error
// GetBuild loads one build, or ErrNotFound.
GetBuild(ctx context.Context, id string) (*Build, error)
// GetBuilds loads the builds among ids that exist, in no particular order and
// without their Dockerfile — one query for a whole list's worth of lookups.
GetBuilds(ctx context.Context, ids []string) ([]Build, error)
// SetBuildJob records the Job name and advances status to building.
SetBuildJob(ctx context.Context, id, jobName string) error
// FinishBuild sets a terminal status, an optional error, and finished_at.
@@ -519,6 +522,24 @@ func (b *Builder) Get(ctx context.Context, id string) (*Build, error) {
return b.Store.GetBuild(ctx, id)
}
// GetMany returns the builds among ids that exist, keyed by id, read as stored
// (no reconcile) and without their Dockerfile. An id with no row is absent from
// the map.
func (b *Builder) GetMany(ctx context.Context, ids []string) (map[string]Build, error) {
out := make(map[string]Build, len(ids))
if len(ids) == 0 {
return out, nil
}
builds, err := b.Store.GetBuilds(ctx, ids)
if err != nil {
return nil, err
}
for _, bld := range builds {
out[bld.ID] = bld
}
return out, nil
}
// ListBuilds pages the build history for the admin panel, so every admin sees
// every build (and can cancel a running one) from any browser. It reads rows as
// stored: reconcileBuilds advances them in the background, and GET
+38
View File
@@ -25,6 +25,8 @@ type fakeStore struct {
removeErr error
createErr error
listOpts ListOpts
getManyCalls int
}
func newFakeStore() *fakeStore {
@@ -49,6 +51,17 @@ func (f *fakeStore) GetBuild(_ context.Context, id string) (*Build, error) {
return &cp, nil
}
func (f *fakeStore) GetBuilds(_ context.Context, ids []string) ([]Build, error) {
f.getManyCalls++
var out []Build
for _, id := range ids {
if b, ok := f.builds[id]; ok {
out = append(out, *b)
}
}
return out, nil
}
func (f *fakeStore) SetBuildJob(_ context.Context, id, jobName string) error {
b, ok := f.builds[id]
if !ok {
@@ -776,3 +789,28 @@ func TestListBuildsBoundsThePage(t *testing.T) {
}
}
}
// GetMany answers a whole list's lookups with one store read, keys the rows by
// id, leaves an unknown id out, and asks nothing of the store for an empty list.
func TestGetManyIsOneRead(t *testing.T) {
b, st, _ := newBuilder()
st.builds["bld-1"] = &Build{ID: "bld-1", Status: StatusFailed, Error: "scan found a CRITICAL CVE"}
st.builds["bld-2"] = &Build{ID: "bld-2", Status: StatusSucceeded}
got, err := b.GetMany(context.Background(), []string{"bld-1", "bld-2", "bld-gone"})
if err != nil {
t.Fatalf("GetMany: %v", err)
}
if st.getManyCalls != 1 {
t.Fatalf("store reads = %d, want 1", st.getManyCalls)
}
if len(got) != 2 || got["bld-1"].Status != StatusFailed || got["bld-1"].Error != "scan found a CRITICAL CVE" ||
got["bld-2"].Status != StatusSucceeded {
t.Fatalf("GetMany = %+v, want bld-1 failed and bld-2 succeeded", got)
}
if _, ok := got["bld-gone"]; ok {
t.Fatal("an unknown id must be absent")
}
if got, err := b.GetMany(context.Background(), nil); err != nil || len(got) != 0 || st.getManyCalls != 1 {
t.Fatalf("GetMany(nil) = %v, %v with %d reads; want an empty map and no read", got, err, st.getManyCalls)
}
}
+10
View File
@@ -36,6 +36,16 @@ func (s *PGStore) GetBuild(ctx context.Context, id string) (*Build, error) {
return s.scanBuild(s.db.QueryRowContext(ctx, q, id))
}
func (s *PGStore) GetBuilds(ctx context.Context, ids []string) ([]Build, error) {
rows, err := s.db.QueryContext(ctx, `SELECT id, image_ref, status, '', context_ref, base_image,
requested_by, job_name, log_ref, error, created_at, finished_at, context_digest
FROM image_builds WHERE id = ANY($1::text[])`, ids)
if err != nil {
return nil, err
}
return scanBuilds(rows)
}
func (s *PGStore) scanBuild(row *sql.Row) (*Build, error) {
var (
b Build
+194
View File
@@ -0,0 +1,194 @@
//go:build pgint
package pgint
import (
"context"
"fmt"
"strings"
"testing"
"time"
"felis.lolicon.best/internal/api"
"felis.lolicon.best/internal/build"
"felis.lolicon.best/internal/submit"
)
// ---- list pages and batched lookups (api-surface-7) ------------------------------
// GetBuilds answers a page's worth of build lookups in one query: the rows that
// exist, with their outcome and without the Dockerfile, and nothing for an id
// that has no row or for no ids at all.
func TestBuildStoreGetBuilds(t *testing.T) {
ctx := context.Background()
s := build.NewPGStore(db)
tag := suffix(t)
ok, failed := "bld-ok-"+tag, "bld-failed-"+tag
for _, id := range []string{ok, failed} {
if err := s.CreateBuild(ctx, &build.Build{ID: id, ImageRef: "registry.felis.svc:5000/x/" + id + ":1",
Status: build.StatusPending, RequestedBy: "pgint", Dockerfile: "FROM scratch\n", CreatedAt: mustNow()}); err != nil {
t.Fatalf("CreateBuild(%s): %v", id, err)
}
}
if err := s.FinishBuild(ctx, failed, build.StatusFailed, "trivy: CRITICAL", mustNow()); err != nil {
t.Fatalf("FinishBuild: %v", err)
}
got, err := s.GetBuilds(ctx, []string{ok, failed, "bld-gone-" + tag})
if err != nil {
t.Fatalf("GetBuilds: %v", err)
}
byID := map[string]build.Build{}
for _, b := range got {
byID[b.ID] = b
}
if len(got) != 2 || byID[ok].Status != build.StatusPending ||
byID[failed].Status != build.StatusFailed || byID[failed].Error != "trivy: CRITICAL" {
t.Fatalf("GetBuilds = %+v, want %s pending and %s failed with its error", got, ok, failed)
}
if byID[ok].Dockerfile != "" || byID[ok].RequestedBy != "pgint" {
t.Fatalf("row = %+v, want the requester and no Dockerfile", byID[ok])
}
if got, err := s.GetBuilds(ctx, nil); err != nil || len(got) != 0 {
t.Fatalf("GetBuilds(nil) = %v, %v; want nothing", got, err)
}
}
// PageSubmissions pages one scope newest first (id breaking a created_at tie),
// filters by status and by any part of the id, submitter or name, reports the
// filtered total, and counts the scope's statuses regardless of the filters.
func TestSubmitStorePageSubmissions(t *testing.T) {
ctx := context.Background()
s := submit.NewPGStore(db)
tag := suffix(t)
alice, bob := "pgint-alice-"+tag, "pgint-bob-"+tag
base := mustNow().Add(-time.Hour).Truncate(time.Second)
// sub-0 oldest … sub-4 newest; sub-2 and sub-3 share a timestamp.
at := []time.Duration{0, time.Minute, 2 * time.Minute, 2 * time.Minute, 3 * time.Minute}
ids := make([]string, len(at))
for i := range at {
ids[i] = fmt.Sprintf("sub-%s-%d", tag, i)
if _, err := s.CreateSubmission(ctx, &submit.Submission{ID: ids[i], SubmittedBy: alice,
DisplayName: fmt.Sprintf("Pack %d of %s", i, tag), ContextRef: "s3://b/" + ids[i],
Status: submit.StatusPendingReview, CreatedAt: base.Add(at[i])}, 100); err != nil {
t.Fatalf("CreateSubmission(%d): %v", i, err)
}
}
bobs := "sub-" + tag + "-bob"
if _, err := s.CreateSubmission(ctx, &submit.Submission{ID: bobs, SubmittedBy: bob,
DisplayName: "Bob's pack " + tag, ContextRef: "s3://b/" + bobs,
Status: submit.StatusPendingReview, CreatedAt: base}, 100); err != nil {
t.Fatalf("CreateSubmission(bob): %v", err)
}
if won, err := s.ApproveSubmission(ctx, ids[1], "admin", "registry/x", "", mustNow()); err != nil || !won {
t.Fatalf("approve = %v, %v", won, err)
}
if won, err := s.RejectSubmission(ctx, ids[0], "admin", "no", mustNow()); err != nil || !won {
t.Fatalf("reject = %v, %v", won, err)
}
idsOf := func(p submit.Page) string {
var out []string
for _, sub := range p.Submissions {
out = append(out, sub.ID)
}
return strings.Join(out, ",")
}
page := func(opts submit.ListOpts) submit.Page {
t.Helper()
p, err := s.PageSubmissions(ctx, opts)
if err != nil {
t.Fatalf("PageSubmissions(%+v): %v", opts, err)
}
return p
}
p := page(submit.ListOpts{SubmittedBy: alice, Limit: 2})
if want := ids[4] + "," + ids[3]; idsOf(p) != want || p.Total != 5 {
t.Fatalf("first page = %s of %d, want %s of 5", idsOf(p), p.Total, want)
}
if p.Counts[submit.StatusPendingReview] != 3 || p.Counts[submit.StatusApproved] != 1 ||
p.Counts[submit.StatusRejected] != 1 || len(p.Counts) != 3 {
t.Fatalf("counts = %v, want 3 pending, 1 approved, 1 rejected", p.Counts)
}
p = page(submit.ListOpts{SubmittedBy: alice, Limit: 2, Offset: 2})
if want := ids[2] + "," + ids[1]; idsOf(p) != want {
t.Fatalf("second page = %s, want %s (the tie broken by id)", idsOf(p), want)
}
p = page(submit.ListOpts{SubmittedBy: alice, Status: submit.StatusApproved, Limit: 10})
if idsOf(p) != ids[1] || p.Total != 1 || p.Counts[submit.StatusPendingReview] != 3 {
t.Fatalf("approved = %s of %d with counts %v, want %s of 1 and the scope's counts", idsOf(p), p.Total, p.Counts, ids[1])
}
p = page(submit.ListOpts{SubmittedBy: alice, Query: strings.ToUpper("pack 2 of " + tag), Limit: 10})
if idsOf(p) != ids[2] || p.Total != 1 {
t.Fatalf("name query = %s of %d, want %s", idsOf(p), p.Total, ids[2])
}
p = page(submit.ListOpts{SubmittedBy: alice, Query: tag + "-3", Limit: 10})
if idsOf(p) != ids[3] {
t.Fatalf("id query = %s, want %s", idsOf(p), ids[3])
}
p = page(submit.ListOpts{SubmittedBy: bob, Limit: 10})
if idsOf(p) != bobs || p.Total != 1 || p.Counts[submit.StatusPendingReview] != 1 || len(p.Counts) != 1 {
t.Fatalf("bob's scope = %s of %d with %v, want only %s", idsOf(p), p.Total, p.Counts, bobs)
}
// The admin queue spans submitters; the tag keeps the shared schema out.
p = page(submit.ListOpts{Query: tag, Limit: 10})
if p.Total != 6 || !containsSubmission(p.Submissions, bobs) || !containsSubmission(p.Submissions, ids[0]) {
t.Fatalf("admin queue = %s of %d, want all 6", idsOf(p), p.Total)
}
p = page(submit.ListOpts{Query: "pgint-bob-" + tag, Limit: 10})
if idsOf(p) != bobs {
t.Fatalf("submitter query = %s, want %s", idsOf(p), bobs)
}
}
// The backups list pages newest first inside the caller's scope, narrows to one
// server on request, counts the matches, and never lists a backup that is gone.
func TestBackupListPaging(t *testing.T) {
ctx := context.Background()
tag := suffix(t)
s1, s2 := "pg-a-"+tag, "pg-b-"+tag
uid, other := "owner-"+tag, "other-"+tag
base := mustNow().Add(-time.Hour).Truncate(time.Second)
seed := func(id, server, owner, status string, at time.Duration) {
t.Helper()
if _, err := db.ExecContext(ctx,
`INSERT INTO world_backups (id, server_name, former_owner, backup_ref, size_bytes, reason, status, created_at, expires_at)
VALUES ($1, $2, $3, $4, 1, 'manual', $5, $6, $7)`,
id, server, owner, "/archives/"+id, status, base.Add(at), base.Add(90*24*time.Hour)); err != nil {
t.Fatalf("seed %s: %v", id, err)
}
}
b := func(n string) string { return "bk-" + tag + "-" + n }
seed(b("1"), s1, uid, "present", 0)
seed(b("2"), s1, other, "present", time.Minute)
seed(b("3"), s1, uid, "present", time.Minute) // ties with 2
seed(b("4"), s1, uid, "deleted", 2*time.Minute)
seed(b("5"), s2, uid, "present", 3*time.Minute)
idsOf := func(vs []api.BackupView) string {
var out []string
for _, v := range vs {
out = append(out, v.ID)
}
return strings.Join(out, ",")
}
vs, total, err := repo.AllBackups(ctx, api.BackupListOpts{Server: s1, Limit: 2})
if err != nil || idsOf(vs) != b("3")+","+b("2") || total != 3 {
t.Fatalf("s1 first page = %s of %d (%v), want %s,%s of 3", idsOf(vs), total, err, b("3"), b("2"))
}
vs, total, err = repo.AllBackups(ctx, api.BackupListOpts{Server: s1, Limit: 2, Offset: 2})
if err != nil || idsOf(vs) != b("1") || total != 3 {
t.Fatalf("s1 second page = %s of %d (%v), want %s of 3", idsOf(vs), total, err, b("1"))
}
vs, total, err = repo.BackupsForUser(ctx, uid, api.BackupListOpts{Server: s1, Limit: 10})
if err != nil || idsOf(vs) != b("3")+","+b("1") || total != 2 {
t.Fatalf("owner's s1 = %s of %d (%v), want %s,%s of 2", idsOf(vs), total, err, b("3"), b("1"))
}
vs, total, err = repo.BackupsForUser(ctx, uid, api.BackupListOpts{Limit: 10})
if err != nil || idsOf(vs) != b("5")+","+b("3")+","+b("1") || total != 3 {
t.Fatalf("owner's worlds = %s of %d (%v), want %s,%s,%s", idsOf(vs), total, err, b("5"), b("3"), b("1"))
}
vs, total, err = repo.BackupsForUser(ctx, other, api.BackupListOpts{Server: s2, Limit: 10})
if err != nil || len(vs) != 0 || total != 0 {
t.Fatalf("other's s2 = %s of %d (%v), want nothing", idsOf(vs), total, err)
}
}
+4 -4
View File
@@ -1330,12 +1330,12 @@ func TestSubmitStoreContract(t *testing.T) {
if _, err := s.GetSubmission(ctx, "missing-"+suffix(t)); !errors.Is(err, submit.ErrNotFound) {
t.Fatalf("missing submission = %v, want ErrNotFound", err)
}
byUser, err := s.ListSubmissionsBy(ctx, u.ID)
byUser, err := s.PageSubmissions(ctx, submit.ListOpts{SubmittedBy: u.ID, Limit: submit.MaxListLimit})
if err != nil {
t.Fatalf("ListSubmissionsBy: %v", err)
t.Fatalf("PageSubmissions: %v", err)
}
if !containsSubmission(byUser, id) {
t.Fatal("ListSubmissionsBy must return the caller's submission")
if !containsSubmission(byUser.Submissions, id) {
t.Fatal("PageSubmissions must return the caller's submission")
}
if all, err := s.ListSubmissions(ctx); err != nil || !containsSubmission(all, id) {
t.Fatalf("ListSubmissions: (%v, %v), want the submission present", all, err)
+1 -1
View File
@@ -342,7 +342,7 @@ func TestBackupReadBack(t *testing.T) {
t.Fatalf("BackupByID(intact) = (%+v, %v); want not Corrupt", b, err)
}
views, err := repo.AllBackups(ctx)
views, _, err := repo.AllBackups(ctx, api.BackupListOpts{Server: name, Limit: api.MaxBackupListLimit})
if err != nil {
t.Fatalf("AllBackups: %v", err)
}
+6 -3
View File
@@ -87,8 +87,11 @@ func ControlPlaneRBAC(p Params) RBAC {
// follow). It also Gets the world PVC before backup/restore
// (internal/api.k8scluster.WorldVolumeExists) so a never-started or reaped
// world is refused up front instead of leaving a Job Pending on a missing
// claim. felis-api uses a DIRECT client, so it needs no list/watch beyond the
// explicit List calls — and the PVC grant is get-only, mirroring that.
// claim. felis-api reads the fleet (velocity's pull, the fleet page, the wake
// cap) from an informer cache of minecraftservers, hence watch on that one
// resource; everything else goes through a DIRECT client, so it needs no
// list/watch beyond the explicit List calls — and the PVC grant is get-only,
// mirroring that.
//
// The read-side grant is deliberately minimal: pods:list + pods/log:get, NOT
// pods:get — the streamer lists pods by the server label then reads the chosen
@@ -98,7 +101,7 @@ func ControlPlaneRBAC(p Params) RBAC {
func APIMinecraftRole(p Params) *rbacv1.Role {
p = p.withDefaults()
return role(p.MinecraftNamespace, "felis-api", ComponentAPI, []rbacv1.PolicyRule{
rule([]string{groupFelis}, []string{"minecraftservers"}, []string{"get", "list", "create", "patch"}),
rule([]string{groupFelis}, []string{"minecraftservers"}, []string{"get", "list", "watch", "create", "patch"}),
rule([]string{groupCore}, []string{"secrets"}, []string{"get"}),
// get-only: WorldVolumeExists does a single direct Get of the world PVC;
// nothing in felis-api lists or deletes PVCs.
+7 -1
View File
@@ -107,7 +107,8 @@ func TestAPIRole_CreatesJobsInBothNamespaces(t *testing.T) {
// what it must NOT have: no status writes, no delete.
func TestAPIRole_MinecraftPowersExact(t *testing.T) {
mc := roleByName(t, ControlPlaneRBAC(testParams()).Roles, "felis-api")
for _, v := range []string{"get", "list", "create", "patch"} {
// watch backs the informer cache the fleet reads come from.
for _, v := range []string{"get", "list", "watch", "create", "patch"} {
if !hasRule(mc, groupFelis, "minecraftservers", v) {
t.Errorf("felis-api must have minecraftservers:%s", v)
}
@@ -121,6 +122,11 @@ func TestAPIRole_MinecraftPowersExact(t *testing.T) {
if !hasRule(mc, groupCore, "secrets", "get") {
t.Error("felis-api must read RCON secrets (secrets:get) for console writes")
}
// The informer cache covers minecraftservers only; RCON secrets stay a direct
// Get by name, so no watch or list ever mirrors every secret into felis-api.
if hasRule(mc, groupCore, "secrets", "list") || hasRule(mc, groupCore, "secrets", "watch") {
t.Error("felis-api must NOT list or watch secrets (RCON reads are a direct Get by name)")
}
// WorldVolumeExists (backup/restore pre-gate) does a single direct PVC Get;
// nothing in felis-api lists or deletes claims.
if !hasRule(mc, groupCore, "persistentvolumeclaims", "get") {
@@ -0,0 +1,11 @@
-- The lists people page through read newest first with id breaking ties: the
-- admin submission queue across every user, and the backups list across every
-- server, one former owner's worlds, or one server's archives. Each index lets a
-- page read its own rows instead of sorting the whole table on every request.
CREATE INDEX image_submissions_created_idx ON image_submissions (created_at DESC, id DESC);
CREATE INDEX world_backups_present_idx ON world_backups (created_at DESC, id DESC)
WHERE status = 'present';
CREATE INDEX world_backups_server_present_idx ON world_backups (server_name, created_at DESC, id DESC)
WHERE status = 'present';
CREATE INDEX world_backups_former_owner_present_idx ON world_backups (former_owner, created_at DESC, id DESC)
WHERE status = 'present';
+33 -4
View File
@@ -73,10 +73,39 @@ func (s *PGStore) ListSubmissions(ctx context.Context) ([]Submission, error) {
return s.querySubmissions(ctx, q)
}
func (s *PGStore) ListSubmissionsBy(ctx context.Context, submittedBy string) ([]Submission, error) {
const q = `SELECT ` + submissionColumns + `
FROM image_submissions WHERE submitted_by = $1 ORDER BY created_at DESC`
return s.querySubmissions(ctx, q, submittedBy)
// PageSubmissions reads the scope's status counts, the filtered total and one
// page. id breaks created_at ties so a page boundary never repeats or skips a row.
func (s *PGStore) PageSubmissions(ctx context.Context, opts ListOpts) (Page, error) {
page := Page{Counts: map[Status]int{}}
rows, err := s.db.QueryContext(ctx, `SELECT status, count(*) FROM image_submissions
WHERE $1::text = '' OR submitted_by = $1::text GROUP BY status`, opts.SubmittedBy)
if err != nil {
return Page{}, err
}
defer rows.Close()
for rows.Next() {
var st string
var n int
if err := rows.Scan(&st, &n); err != nil {
return Page{}, err
}
page.Counts[Status(st)] = n
}
if err := rows.Err(); err != nil {
return Page{}, err
}
const match = ` WHERE ($1::text = '' OR submitted_by = $1::text)
AND ($2::text = '' OR status::text = $2::text)
AND ($3::text = '' OR strpos(lower(id), lower($3::text)) > 0
OR strpos(lower(submitted_by), lower($3::text)) > 0
OR strpos(lower(display_name), lower($3::text)) > 0)`
args := []any{opts.SubmittedBy, string(opts.Status), opts.Query}
if err := s.db.QueryRowContext(ctx, `SELECT count(*) FROM image_submissions`+match, args...).Scan(&page.Total); err != nil {
return Page{}, err
}
page.Submissions, err = s.querySubmissions(ctx, `SELECT `+submissionColumns+` FROM image_submissions`+match+`
ORDER BY created_at DESC, id DESC LIMIT $4 OFFSET $5`, append(args, opts.Limit, opts.Offset)...)
return page, err
}
// cas executes a single-statement compare-and-set — an UPDATE or DELETE whose
+60 -10
View File
@@ -210,6 +210,34 @@ type Submission struct {
ContextSHA256 string `json:"context_sha256,omitempty"`
}
// ListOpts selects a page of submissions, newest first. SubmittedBy scopes it to
// one user (the "my uploads" view); empty is the admin queue across all users.
// Status matches exactly; Query matches any part of the id, the submitter or the
// display name, ignoring case. Empty filters match everything.
type ListOpts struct {
SubmittedBy string
Status Status
Query string
Limit int
Offset int
}
// Page is one page of submissions plus what a list header shows: Total is how
// many match the filters in all (for the pager), Counts how many of the scope's
// submissions sit in each status regardless of Status and Query (for the summary
// cards, which must not shrink as the reviewer narrows the list).
type Page struct {
Submissions []Submission
Total int
Counts map[Status]int
}
// DefaultListLimit and MaxListLimit bound one page of submissions.
const (
DefaultListLimit = 20
MaxListLimit = 100
)
// Store is the business-layer persistence the Manager depends on. It is an
// interface so the Manager is tested against an in-memory fake; the Postgres
// implementation (PGStore) is integration-tested only.
@@ -222,10 +250,13 @@ type Store interface {
CreateSubmission(ctx context.Context, s *Submission, maxPending int) (int, error)
// GetSubmission loads one submission, or ErrNotFound.
GetSubmission(ctx context.Context, id string) (*Submission, error)
// ListSubmissions returns every submission, newest first (admin queue).
// ListSubmissions returns every submission, newest first — the full scan the
// storage budget and the rejected-blob reaper need. Lists shown to a person
// go through PageSubmissions.
ListSubmissions(ctx context.Context) ([]Submission, error)
// ListSubmissionsBy returns one user's submissions, newest first.
ListSubmissionsBy(ctx context.Context, submittedBy string) ([]Submission, error)
// PageSubmissions returns one page of submissions, newest first, with the
// match total and the scope's per-status counts (see Page).
PageSubmissions(ctx context.Context, opts ListOpts) (Page, error)
// ApproveSubmission atomically flips pending_review -> approved, recording the
// derived image_ref, the reviewer and reviewed_at. It reports whether THIS
// call won the transition: false means a concurrent review already moved the
@@ -951,17 +982,36 @@ func (m *Manager) deleteBlob(ctx context.Context, id string) error {
return nil
}
// List returns every submission, newest first (the admin review queue).
func (m *Manager) List(ctx context.Context) ([]Submission, error) {
return m.Store.ListSubmissions(ctx)
// List returns one page of every user's submissions, newest first (the admin
// review queue). opts.SubmittedBy is ignored: the queue is the whole lane.
func (m *Manager) List(ctx context.Context, opts ListOpts) (Page, error) {
opts.SubmittedBy = ""
return m.page(ctx, opts)
}
// ListBy returns one user's submissions, newest first (the "my uploads" view).
func (m *Manager) ListBy(ctx context.Context, submittedBy string) ([]Submission, error) {
// ListBy returns one page of one user's submissions, newest first (the "my
// uploads" view). The scope is the submittedBy argument, never opts.
func (m *Manager) ListBy(ctx context.Context, submittedBy string, opts ListOpts) (Page, error) {
if strings.TrimSpace(submittedBy) == "" {
return nil, invalidf("submitter identity is required")
return Page{}, invalidf("submitter identity is required")
}
return m.Store.ListSubmissionsBy(ctx, submittedBy)
opts.SubmittedBy = submittedBy
return m.page(ctx, opts)
}
func (m *Manager) page(ctx context.Context, opts ListOpts) (Page, error) {
switch opts.Status {
case "", StatusPendingReview, StatusApproved, StatusRejected:
default:
return Page{}, invalidf("unknown status %q", opts.Status)
}
opts.Query = strings.TrimSpace(opts.Query)
if opts.Limit <= 0 {
opts.Limit = DefaultListLimit
}
opts.Limit = min(opts.Limit, MaxListLimit)
opts.Offset = max(opts.Offset, 0)
return m.Store.PageSubmissions(ctx, opts)
}
// Compile-time proof that the production build subsystem satisfies Builds.
+49 -9
View File
@@ -104,7 +104,8 @@ type fakeStore struct {
linkErr error
deleteErr error
linked []string // "id=buildID" recorder
linked []string // "id=buildID" recorder
pageOpts ListOpts // the last PageSubmissions call
// beforeApprove runs between the Manager's read and its CAS, the window a
// concurrent re-upload lands in.
@@ -151,14 +152,19 @@ func (f *fakeStore) ListSubmissions(_ context.Context) ([]Submission, error) {
return out, nil
}
func (f *fakeStore) ListSubmissionsBy(_ context.Context, by string) ([]Submission, error) {
var out []Submission
// PageSubmissions records the options the Manager settled on and scopes by
// submitter; the SQL filters and paging are pgint's to prove.
func (f *fakeStore) PageSubmissions(_ context.Context, opts ListOpts) (Page, error) {
f.pageOpts = opts
page := Page{Counts: map[Status]int{}}
for _, s := range f.subs {
if s.SubmittedBy == by {
out = append(out, *s)
if opts.SubmittedBy == "" || s.SubmittedBy == opts.SubmittedBy {
page.Submissions = append(page.Submissions, *s)
page.Counts[s.Status]++
}
}
return out, nil
page.Total = len(page.Submissions)
return page, nil
}
func (f *fakeStore) ApproveSubmission(_ context.Context, id, reviewedBy, imageRef, digest string, at time.Time) (bool, error) {
@@ -618,18 +624,52 @@ func TestListBy(t *testing.T) {
if _, err := m.Create(context.Background(), CreateRequest{DisplayName: "B", SubmittedBy: "user-2"}); err != nil {
t.Fatal(err)
}
mine, err := m.ListBy(context.Background(), "user-1")
// The scope is the argument: a SubmittedBy smuggled into opts cannot widen it.
mine, err := m.ListBy(context.Background(), "user-1", ListOpts{SubmittedBy: "user-2"})
if err != nil {
t.Fatalf("ListBy: %v", err)
}
if len(mine) != 1 || mine[0].SubmittedBy != "user-1" {
if len(mine.Submissions) != 1 || mine.Submissions[0].SubmittedBy != "user-1" || mine.Total != 1 {
t.Fatalf("ListBy scoped wrong: %+v", mine)
}
if _, err := m.ListBy(context.Background(), ""); !errors.Is(err, ErrInvalid) {
if _, err := m.ListBy(context.Background(), "", ListOpts{}); !errors.Is(err, ErrInvalid) {
t.Fatalf("empty submitter err = %v, want ErrInvalid", err)
}
}
// List is the whole lane whatever opts names, and both lists settle the page the
// same way: a missing limit takes the default, an oversized one the cap, a
// negative offset zero, and the query loses its padding.
func TestListPagingOptions(t *testing.T) {
m, st, _ := newManager()
for _, by := range []string{"user-1", "user-2"} {
if _, err := m.Create(context.Background(), CreateRequest{DisplayName: "P", SubmittedBy: by}); err != nil {
t.Fatal(err)
}
}
all, err := m.List(context.Background(), ListOpts{SubmittedBy: "user-1", Query: " pack ", Offset: -5})
if err != nil {
t.Fatalf("List: %v", err)
}
if all.Total != 2 {
t.Fatalf("List total = %d, want 2 (the admin queue ignores a submitter in opts)", all.Total)
}
want := ListOpts{Query: "pack", Limit: 20, Offset: 0}
if st.pageOpts != want {
t.Fatalf("List settled %+v, want %+v", st.pageOpts, want)
}
if _, err := m.ListBy(context.Background(), "user-2", ListOpts{Limit: 5000, Offset: 40, Status: StatusApproved}); err != nil {
t.Fatalf("ListBy: %v", err)
}
want = ListOpts{SubmittedBy: "user-2", Status: StatusApproved, Limit: 100, Offset: 40}
if st.pageOpts != want {
t.Fatalf("ListBy settled %+v, want %+v", st.pageOpts, want)
}
if _, err := m.List(context.Background(), ListOpts{Status: "building"}); !errors.Is(err, ErrInvalid) {
t.Fatalf("unknown status err = %v, want ErrInvalid", err)
}
}
func TestUploadContextStoresUnderDerivedID(t *testing.T) {
m, _, _ := newManager()
fb := newFakeBlobs()