at-container-registry/pkg/hold/labeler/subscriber_test.go

package labeler

import (
	"context"
	"sync"
	"testing"
	"time"

	comatproto "github.com/bluesky-social/indigo/api/atproto"
)

// stubPurger captures purge calls so we can assert what the subscriber routed.
type stubPurger struct {
	mu               sync.Mutex
	manifestCalls    []string
	userLevelCalls   []string
	purgeManifestErr error
	purgeUserError   error
	manifestOutcome  PurgeOutcome
	userLevelOutcome PurgeOutcome
}

func (s *stubPurger) PurgeManifestRecords(_ context.Context, uri string) (PurgeOutcome, error) {
	s.mu.Lock()
	s.manifestCalls = append(s.manifestCalls, uri)
	s.mu.Unlock()
	return s.manifestOutcome, s.purgeManifestErr
}

func (s *stubPurger) PurgeUserManifests(_ context.Context, did string) (PurgeOutcome, error) {
	s.mu.Lock()
	s.userLevelCalls = append(s.userLevelCalls, did)
	s.mu.Unlock()
	return s.userLevelOutcome, s.purgeUserError
}

func (s *stubPurger) snapshot() (manifests, users []string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return append([]string(nil), s.manifestCalls...), append([]string(nil), s.userLevelCalls...)
}

func TestApplyLabelManifestTakedownPurges(t *testing.T) {
	cache := newTestCache(t)
	purger := &stubPurger{}
	// Use a did:web identifier directly so we control exactly what Src must
	// match — derived URL is https://labeler.example.com.
	sub := NewSubscriber("did:web:labeler.example.com", cache, purger)

	uri := "at://did:plc:alice/io.atcr.manifest/abc"
	cts := time.Now().UTC().Format(time.RFC3339)

	sub.applyLabel(1, &comatproto.LabelDefs_Label{
		Src: "did:web:labeler.example.com",
		Uri: uri,
		Val: TakedownLabelValue,
		Cts: cts,
	})

	manifests, users := purger.snapshot()
	if len(manifests) != 1 || manifests[0] != uri {
		t.Fatalf("manifest purges = %v, want [%q]", manifests, uri)
	}
	if len(users) != 0 {
		t.Fatalf("expected no user-level purges, got %v", users)
	}
	if _, ok := cache.IsTakenDown(uri); !ok {
		t.Fatalf("cache should record the takedown")
	}
}

func TestApplyLabelUserLevelTakedownPurgesAllManifests(t *testing.T) {
	cache := newTestCache(t)
	purger := &stubPurger{}
	sub := NewSubscriber("did:web:labeler.example.com", cache, purger)

	uri := "at://did:plc:alice"
	sub.applyLabel(1, &comatproto.LabelDefs_Label{
		Src: "did:web:labeler.example.com",
		Uri: uri,
		Val: TakedownLabelValue,
		Cts: time.Now().UTC().Format(time.RFC3339),
	})

	manifests, users := purger.snapshot()
	if len(users) != 1 || users[0] != "did:plc:alice" {
		t.Fatalf("user-level purges = %v, want [did:plc:alice]", users)
	}
	if len(manifests) != 0 {
		t.Fatalf("expected no per-manifest purges, got %v", manifests)
	}
	if _, ok := cache.IsTakenDown("at://did:plc:alice/io.atcr.manifest/anything"); !ok {
		t.Fatalf("user-level entry should mask any manifest URI for that DID")
	}
}

func TestApplyLabelNegationDropsCacheNoPurge(t *testing.T) {
	cache := newTestCache(t)
	uri := "at://did:plc:alice/io.atcr.manifest/abc"
	if err := cache.Set(uri, "did:web:labeler.example.com", time.Now()); err != nil {
		t.Fatal(err)
	}
	purger := &stubPurger{}
	sub := NewSubscriber("did:web:labeler.example.com", cache, purger)

	sub.applyLabel(2, &comatproto.LabelDefs_Label{
		Src: "did:web:labeler.example.com",
		Uri: uri,
		Val: TakedownLabelValue,
		Neg: new(true),
		Cts: time.Now().UTC().Format(time.RFC3339),
	})

	if _, ok := cache.IsTakenDown(uri); ok {
		t.Fatalf("negation should drop the takedown from cache")
	}
	manifests, _ := purger.snapshot()
	if len(manifests) != 0 {
		t.Fatalf("negation must not trigger purge, got %v", manifests)
	}
}

func TestApplyLabelIgnoresUntrustedSource(t *testing.T) {
	cache := newTestCache(t)
	purger := &stubPurger{}
	// Subscriber is configured for did:web:operator; a label whose Src is a
	// different DID must be ignored (today's single-DID trust model).
	sub := NewSubscriber("did:web:operator", cache, purger)

	sub.applyLabel(1, &comatproto.LabelDefs_Label{
		Src: "did:web:rogue",
		Uri: "at://did:plc:alice/io.atcr.manifest/abc",
		Val: TakedownLabelValue,
		Cts: time.Now().UTC().Format(time.RFC3339),
	})

	manifests, users := purger.snapshot()
	if len(manifests) != 0 || len(users) != 0 {
		t.Fatalf("untrusted source must not trigger purge: manifests=%v users=%v", manifests, users)
	}
}

func TestSubscriberStoresDID(t *testing.T) {
	tests := []string{
		"did:web:labeler.atcr.io",
		"did:web:172.28.0.4%3A5002",
		"did:plc:4zul2zfigjltl24ti24xj3hy",
	}
	for _, did := range tests {
		t.Run(did, func(t *testing.T) {
			sub := NewSubscriber(did, nil, nil)
			if sub.labelerDID != did {
				t.Errorf("labelerDID = %q, want %q", sub.labelerDID, did)
			}
		})
	}
}

func TestApplyLabelIgnoresNonTakedownValues(t *testing.T) {
	cache := newTestCache(t)
	purger := &stubPurger{}
	sub := NewSubscriber("did:web:labeler.example.com", cache, purger)

	sub.applyLabel(1, &comatproto.LabelDefs_Label{
		Src: "did:web:labeler.example.com",
		Uri: "at://did:plc:alice/io.atcr.manifest/abc",
		Val: "spam",
		Cts: time.Now().UTC().Format(time.RFC3339),
	})

	manifests, users := purger.snapshot()
	if len(manifests) != 0 || len(users) != 0 {
		t.Fatalf("non-takedown labels must not trigger purge: manifests=%v users=%v", manifests, users)
	}
}

func TestClassifyURI(t *testing.T) {
	tests := []struct {
		uri  string
		kind uriKind
		did  string
	}{
		{"at://did:plc:alice/io.atcr.manifest/abc", uriKindManifest, "did:plc:alice"},
		{"at://did:plc:alice", uriKindUser, "did:plc:alice"},
		{"at://did:plc:alice/io.atcr.repo/myimage", uriKindOther, "did:plc:alice"},
		{"https://example.com", uriKindOther, ""},
		{"", uriKindOther, ""},
	}
	for _, tt := range tests {
		t.Run(tt.uri, func(t *testing.T) {
			got := classifyURI(tt.uri)
			if got.Kind != tt.kind {
				t.Fatalf("kind = %s, want %s", got.Kind, tt.kind)
			}
			if got.DID != tt.did {
				t.Fatalf("did = %s, want %s", got.DID, tt.did)
			}
		})
	}
}