From 7b4a2e22a2ed141d7f8e252506ba5b323b8cd479 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sat, 2 May 2026 22:59:32 -0500 Subject: [PATCH] fix more db query issues, improve labeler usage --- pkg/appview/db/annotations.go | 30 ++++ pkg/appview/db/queries.go | 244 ++++++++++++++++--------- pkg/appview/db/queries_test.go | 160 +++++++++++++++++ pkg/labeler/db.go | 313 ++++++++++++++++++++++----------- pkg/labeler/db_test.go | 176 ++++++++++++------ pkg/labeler/takedown.go | 267 +++++++++++++++++++++------- 6 files changed, 882 insertions(+), 308 deletions(-) diff --git a/pkg/appview/db/annotations.go b/pkg/appview/db/annotations.go index 0d110d1..86fc636 100644 --- a/pkg/appview/db/annotations.go +++ b/pkg/appview/db/annotations.go @@ -29,6 +29,36 @@ func GetRepositoryAnnotations(db DBTX, did, repository string) (map[string]strin return annotations, rows.Err() } +// GetRepositoryAnnotationsByDID retrieves all annotations for every +// repository owned by a DID, grouped as map[repository]map[key]value. +// Used by bulk-fetch paths to avoid issuing one query per repository. +func GetRepositoryAnnotationsByDID(db DBTX, did string) (map[string]map[string]string, error) { + rows, err := db.Query(` + SELECT repository, key, value + FROM repository_annotations + WHERE did = ? + `, did) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make(map[string]map[string]string) + for rows.Next() { + var repo, key, value string + if err := rows.Scan(&repo, &key, &value); err != nil { + return nil, err + } + m, ok := out[repo] + if !ok { + m = make(map[string]string) + out[repo] = m + } + m[key] = value + } + return out, rows.Err() +} + // UpsertRepositoryAnnotations upserts annotations for a repository. // Stale keys not present in the new map are deleted. // Unchanged values are skipped to avoid unnecessary writes. diff --git a/pkg/appview/db/queries.go b/pkg/appview/db/queries.go index c10c12b..03d821f 100644 --- a/pkg/appview/db/queries.go +++ b/pkg/appview/db/queries.go @@ -241,10 +241,15 @@ func SearchRepositories(db DBTX, query string, limit, offset int, currentUserDID // GetUserRepositories fetches all repositories for a user. // viewerDID scopes results to repositories whose manifests live on holds the // viewer can access (empty viewerDID = anonymous → public + self-service only). +// +// Implementation: one summary query for the accessible repository set, then +// four bulk queries (tags, manifests, annotations, repo_pages) all keyed by +// did. Results are grouped in Go and assembled per repo. Total: 5 queries +// regardless of how many repos the user owns. func GetUserRepositories(db DBTX, did string, viewerDID string) ([]Repository, error) { - // Get repository summary. - // Both tags and manifests are filtered via join onto manifests.hold_endpoint - // so repositories where every row lives on an inaccessible hold drop out. + // Step 1: summary query. Both tags and manifests are filtered via join + // onto manifests.hold_endpoint so repositories where every row lives on + // an inaccessible hold drop out. rows, err := db.Query(` SELECT repository, @@ -264,97 +269,80 @@ func GetUserRepositories(db DBTX, did string, viewerDID string) ([]Repository, e GROUP BY repository ORDER BY last_push DESC `, did, viewerDID, viewerDID, did, viewerDID, viewerDID) - if err != nil { return nil, err } - defer rows.Close() - var repos []Repository + type repoSummary struct { + Name string + TagCount int + ManifestCount int + LastPushStr string + } + var summaries []repoSummary for rows.Next() { - var r Repository - var lastPushStr string - if err := rows.Scan(&r.Name, &r.TagCount, &r.ManifestCount, &lastPushStr); err != nil { + var s repoSummary + if err := rows.Scan(&s.Name, &s.TagCount, &s.ManifestCount, &s.LastPushStr); err != nil { + rows.Close() return nil, err } + summaries = append(summaries, s) + } + rows.Close() - // Parse the timestamp string into time.Time - if lastPushStr != "" { - // Try multiple timestamp formats - formats := []string{ - time.RFC3339Nano, // 2006-01-02T15:04:05.999999999Z07:00 - "2006-01-02 15:04:05.999999999-07:00", // SQLite with microseconds and timezone - "2006-01-02 15:04:05.999999999", // SQLite with microseconds - time.RFC3339, // 2006-01-02T15:04:05Z07:00 - "2006-01-02 15:04:05", // SQLite default - } + if len(summaries) == 0 { + return nil, nil + } - for _, format := range formats { - if t, err := time.Parse(format, lastPushStr); err == nil { - r.LastPush = t - break - } - } - } - - // Get tags for this repo - tagRows, err := db.Query(` - SELECT id, tag, digest, created_at - FROM tags - WHERE did = ? AND repository = ? - ORDER BY created_at DESC - `, did, r.Name) - - if err != nil { - return nil, err - } - - for tagRows.Next() { - var t Tag - t.DID = did - t.Repository = r.Name - if err := tagRows.Scan(&t.ID, &t.Tag, &t.Digest, &t.CreatedAt); err != nil { - tagRows.Close() - return nil, err - } - r.Tags = append(r.Tags, t) - } - tagRows.Close() - - // Get manifests for this repo - manifestRows, err := db.Query(` - SELECT id, digest, hold_endpoint, schema_version, media_type, - config_digest, config_size, artifact_type, created_at - FROM manifests - WHERE did = ? AND repository = ? - ORDER BY created_at DESC - `, did, r.Name) - - if err != nil { - return nil, err - } - - for manifestRows.Next() { - var m Manifest - m.DID = did - m.Repository = r.Name - - if err := manifestRows.Scan(&m.ID, &m.Digest, &m.HoldEndpoint, &m.SchemaVersion, - &m.MediaType, &m.ConfigDigest, &m.ConfigSize, &m.ArtifactType, &m.CreatedAt); err != nil { - manifestRows.Close() - return nil, err - } - - r.Manifests = append(r.Manifests, m) - } - manifestRows.Close() - - // Fetch repository-level annotations from annotations table - annotations, err := GetRepositoryAnnotations(db, did, r.Name) - if err != nil { - return nil, err + // Build the set of accessible repo names for filtering bulk-fetched rows + // against repos that the viewer can't see (rows for repos owned by `did` + // but stored on inaccessible holds). + accessible := make(map[string]bool, len(summaries)) + for _, s := range summaries { + accessible[s.Name] = true + } + + // Step 2: bulk-fetch tags for all repos owned by did, grouped by repo. + tagsByRepo, err := bulkTagsByRepo(db, did, accessible) + if err != nil { + return nil, err + } + + // Step 3: bulk-fetch manifests, grouped by repo. + manifestsByRepo, err := bulkManifestsByRepo(db, did, accessible) + if err != nil { + return nil, err + } + + // Step 4: bulk-fetch annotations, grouped by repo. + annotationsByRepo, err := GetRepositoryAnnotationsByDID(db, did) + if err != nil { + return nil, err + } + + // Step 5: bulk-fetch repo pages (existing helper), keyed by repo. + pages, err := GetRepoPagesByDID(db, did) + if err != nil { + return nil, err + } + pagesByRepo := make(map[string]*RepoPage, len(pages)) + for i := range pages { + pagesByRepo[pages[i].Repository] = &pages[i] + } + + // Assemble results in summary order (preserves last_push DESC). + repos := make([]Repository, 0, len(summaries)) + for _, s := range summaries { + r := Repository{ + Name: s.Name, + TagCount: s.TagCount, + ManifestCount: s.ManifestCount, + LastPush: parseRepoTimestamp(s.LastPushStr), + Tags: tagsByRepo[s.Name], + Manifests: manifestsByRepo[s.Name], } + annotations := annotationsByRepo[s.Name] r.Title = annotations["org.opencontainers.image.title"] r.Description = annotations["org.opencontainers.image.description"] r.SourceURL = annotations["org.opencontainers.image.source"] @@ -363,10 +351,9 @@ func GetUserRepositories(db DBTX, did string, viewerDID string) ([]Repository, e r.IconURL = annotations["io.atcr.icon"] r.ReadmeURL = annotations["io.atcr.readme"] - // Check for repo page avatar (overrides annotation icon) - repoPage, err := GetRepoPage(db, did, r.Name) - if err == nil && repoPage != nil && repoPage.AvatarCID != "" { - r.IconURL = BlobCDNURL(did, repoPage.AvatarCID) + // Repo page avatar overrides annotation icon when present. + if page, ok := pagesByRepo[s.Name]; ok && page.AvatarCID != "" { + r.IconURL = BlobCDNURL(did, page.AvatarCID) } repos = append(repos, r) @@ -375,6 +362,89 @@ func GetUserRepositories(db DBTX, did string, viewerDID string) ([]Repository, e return repos, nil } +// bulkTagsByRepo fetches every tag owned by did and groups by repository, +// dropping repos not in the accessible set. Result preserves created_at DESC +// ordering within each repo. +func bulkTagsByRepo(db DBTX, did string, accessible map[string]bool) (map[string][]Tag, error) { + rows, err := db.Query(` + SELECT id, repository, tag, digest, created_at + FROM tags + WHERE did = ? + ORDER BY repository, created_at DESC + `, did) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make(map[string][]Tag) + for rows.Next() { + var t Tag + t.DID = did + if err := rows.Scan(&t.ID, &t.Repository, &t.Tag, &t.Digest, &t.CreatedAt); err != nil { + return nil, err + } + if !accessible[t.Repository] { + continue + } + out[t.Repository] = append(out[t.Repository], t) + } + return out, rows.Err() +} + +// bulkManifestsByRepo fetches every manifest owned by did and groups by +// repository, dropping repos not in the accessible set. Result preserves +// created_at DESC ordering within each repo. +func bulkManifestsByRepo(db DBTX, did string, accessible map[string]bool) (map[string][]Manifest, error) { + rows, err := db.Query(` + SELECT id, repository, digest, hold_endpoint, schema_version, media_type, + config_digest, config_size, artifact_type, created_at + FROM manifests + WHERE did = ? + ORDER BY repository, created_at DESC + `, did) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make(map[string][]Manifest) + for rows.Next() { + var m Manifest + m.DID = did + if err := rows.Scan(&m.ID, &m.Repository, &m.Digest, &m.HoldEndpoint, &m.SchemaVersion, + &m.MediaType, &m.ConfigDigest, &m.ConfigSize, &m.ArtifactType, &m.CreatedAt); err != nil { + return nil, err + } + if !accessible[m.Repository] { + continue + } + out[m.Repository] = append(out[m.Repository], m) + } + return out, rows.Err() +} + +// parseRepoTimestamp tolerates the several timestamp formats SQLite/libsql +// can return for MAX(created_at) depending on driver and schema history. +func parseRepoTimestamp(s string) time.Time { + if s == "" { + return time.Time{} + } + formats := []string{ + time.RFC3339Nano, // 2006-01-02T15:04:05.999999999Z07:00 + "2006-01-02 15:04:05.999999999-07:00", // SQLite with microseconds and timezone + "2006-01-02 15:04:05.999999999", // SQLite with microseconds + time.RFC3339, // 2006-01-02T15:04:05Z07:00 + "2006-01-02 15:04:05", // SQLite default + } + for _, format := range formats { + if t, err := time.Parse(format, s); err == nil { + return t + } + } + return time.Time{} +} + // GetRepositoryMetadata retrieves metadata for a repository from annotations table // Returns a map of annotation key -> value for easy access in templates and handlers func GetRepositoryMetadata(db DBTX, did string, repository string) (map[string]string, error) { diff --git a/pkg/appview/db/queries_test.go b/pkg/appview/db/queries_test.go index 7d3d129..29e2e91 100644 --- a/pkg/appview/db/queries_test.go +++ b/pkg/appview/db/queries_test.go @@ -1607,3 +1607,163 @@ func TestGetUserRepositories_HoldAccessFilter(t *testing.T) { t.Errorf("crew viewer: expected both repos, got %d: %v", len(repos), repos) } } + +// TestGetUserRepositories_BulkGrouping verifies that the bulk-fetch +// implementation correctly groups tags, manifests, annotations, and repo-page +// avatars per repository — and that ordering (last_push DESC for repos, +// created_at DESC for tags/manifests within a repo) is preserved. +// +// Regression guard for the previous N+1 implementation, which issued one +// query per repo and per relation. +func TestGetUserRepositories_BulkGrouping(t *testing.T) { + db, err := InitDB("file:TestGetUserRepositories_BulkGrouping?mode=memory&cache=shared", LibsqlConfig{}) + if err != nil { + t.Fatalf("init db: %v", err) + } + defer db.Close() + + user := &User{DID: "did:plc:owner", Handle: "owner.test", PDSEndpoint: "https://pds.example", LastSeen: time.Now()} + if err := UpsertUser(db, user); err != nil { + t.Fatalf("upsert user: %v", err) + } + if err := UpsertCaptainRecord(db, &HoldCaptainRecord{ + HoldDID: "did:web:hold.example", OwnerDID: "did:plc:holdowner", Public: true, + }); err != nil { + t.Fatalf("seed captain: %v", err) + } + + now := time.Now().UTC().Truncate(time.Second) + mediaType := "application/vnd.oci.image.manifest.v1+json" + + // repoA: two manifests (oldest then newer) and two tags. last_push = now+10s. + manifestA1, err := InsertManifest(db, &Manifest{ + DID: user.DID, Repository: "repoA", Digest: "sha256:a1", + HoldEndpoint: "did:web:hold.example", SchemaVersion: 2, MediaType: mediaType, + CreatedAt: now, + }) + if err != nil { + t.Fatalf("insert manifest a1: %v", err) + } + manifestA2, err := InsertManifest(db, &Manifest{ + DID: user.DID, Repository: "repoA", Digest: "sha256:a2", + HoldEndpoint: "did:web:hold.example", SchemaVersion: 2, MediaType: mediaType, + CreatedAt: now.Add(5 * time.Second), + }) + if err != nil { + t.Fatalf("insert manifest a2: %v", err) + } + if err := UpsertTag(db, &Tag{DID: user.DID, Repository: "repoA", Tag: "v1", Digest: "sha256:a1", CreatedAt: now.Add(8 * time.Second)}); err != nil { + t.Fatalf("upsert tag v1: %v", err) + } + if err := UpsertTag(db, &Tag{DID: user.DID, Repository: "repoA", Tag: "v2", Digest: "sha256:a2", CreatedAt: now.Add(10 * time.Second)}); err != nil { + t.Fatalf("upsert tag v2: %v", err) + } + + // repoB: one manifest, one tag. last_push = now+1s (older than repoA → repoA sorts first). + if _, err := InsertManifest(db, &Manifest{ + DID: user.DID, Repository: "repoB", Digest: "sha256:b1", + HoldEndpoint: "did:web:hold.example", SchemaVersion: 2, MediaType: mediaType, + CreatedAt: now.Add(1 * time.Second), + }); err != nil { + t.Fatalf("insert manifest b1: %v", err) + } + if err := UpsertTag(db, &Tag{DID: user.DID, Repository: "repoB", Tag: "latest", Digest: "sha256:b1", CreatedAt: now.Add(1 * time.Second)}); err != nil { + t.Fatalf("upsert tag b latest: %v", err) + } + + // Annotations only on repoA, plus a repo-page avatar on repoB to exercise the icon override. + if err := UpsertRepositoryAnnotations(db, user.DID, "repoA", map[string]string{ + "org.opencontainers.image.title": "Repo A Title", + "org.opencontainers.image.description": "alpha", + "io.atcr.icon": "https://example.com/a.png", + }); err != nil { + t.Fatalf("upsert annotations: %v", err) + } + if err := UpsertRepoPage(db, user.DID, "repoB", "", "bafyrepob", false, now, now); err != nil { + t.Fatalf("upsert repo page: %v", err) + } + + repos, err := GetUserRepositories(db, user.DID, "") + if err != nil { + t.Fatalf("GetUserRepositories: %v", err) + } + + // Order: repoA first (newer last_push), then repoB. + if len(repos) != 2 { + t.Fatalf("expected 2 repos, got %d: %#v", len(repos), repos) + } + if repos[0].Name != "repoA" || repos[1].Name != "repoB" { + t.Fatalf("expected order [repoA, repoB] (last_push DESC), got [%s, %s]", repos[0].Name, repos[1].Name) + } + + // repoA grouping + a := repos[0] + if len(a.Tags) != 2 { + t.Errorf("repoA: expected 2 tags, got %d", len(a.Tags)) + } + // tags ordered created_at DESC → v2 first + if len(a.Tags) >= 2 && (a.Tags[0].Tag != "v2" || a.Tags[1].Tag != "v1") { + t.Errorf("repoA tags out of order, want [v2, v1] got [%s, %s]", a.Tags[0].Tag, a.Tags[1].Tag) + } + if len(a.Manifests) != 2 { + t.Errorf("repoA: expected 2 manifests, got %d", len(a.Manifests)) + } + // manifests ordered created_at DESC → a2 first + if len(a.Manifests) >= 2 && (a.Manifests[0].ID != manifestA2 || a.Manifests[1].ID != manifestA1) { + t.Errorf("repoA manifests out of order, want [a2, a1] got [%d, %d]", a.Manifests[0].ID, a.Manifests[1].ID) + } + if a.Title != "Repo A Title" || a.Description != "alpha" { + t.Errorf("repoA annotations not applied: title=%q desc=%q", a.Title, a.Description) + } + if a.IconURL != "https://example.com/a.png" { + t.Errorf("repoA icon: expected annotation URL, got %q", a.IconURL) + } + + // repoB grouping + page-avatar override + b := repos[1] + if len(b.Tags) != 1 || b.Tags[0].Tag != "latest" { + t.Errorf("repoB tags: %#v", b.Tags) + } + if len(b.Manifests) != 1 || b.Manifests[0].Digest != "sha256:b1" { + t.Errorf("repoB manifests: %#v", b.Manifests) + } + if b.IconURL == "" { + t.Errorf("repoB icon should be derived from repo-page avatar CID, got empty") + } + + // Cross-repo isolation: tags/manifests for repoB must not leak into repoA and vice versa. + for _, tag := range a.Tags { + if tag.Repository != "repoA" { + t.Errorf("repoA tag has wrong repository: %#v", tag) + } + } + for _, m := range b.Manifests { + if m.Repository != "repoB" { + t.Errorf("repoB manifest has wrong repository: %#v", m) + } + } +} + +// TestGetUserRepositories_Empty verifies the bulk-fetch path short-circuits +// cleanly when the summary query returns no rows (no extra queries issued, +// nil slice returned). +func TestGetUserRepositories_Empty(t *testing.T) { + db, err := InitDB("file:TestGetUserRepositories_Empty?mode=memory&cache=shared", LibsqlConfig{}) + if err != nil { + t.Fatalf("init db: %v", err) + } + defer db.Close() + + user := &User{DID: "did:plc:nobody", Handle: "nobody.test", PDSEndpoint: "https://pds.example", LastSeen: time.Now()} + if err := UpsertUser(db, user); err != nil { + t.Fatalf("upsert user: %v", err) + } + + repos, err := GetUserRepositories(db, user.DID, "") + if err != nil { + t.Fatalf("GetUserRepositories empty: %v", err) + } + if repos != nil { + t.Errorf("expected nil slice for user with no repos, got %#v", repos) + } +} diff --git a/pkg/labeler/db.go b/pkg/labeler/db.go index f169320..b5a4652 100644 --- a/pkg/labeler/db.go +++ b/pkg/labeler/db.go @@ -19,6 +19,20 @@ import ( const LabelVersion int64 = labeling.ATPROTO_LABEL_VERSION const schema = ` +CREATE TABLE IF NOT EXISTS takedowns ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + input TEXT NOT NULL, + subject_did TEXT NOT NULL, + subject_repo TEXT NOT NULL DEFAULT '', + subject_handle TEXT NOT NULL DEFAULT '', + reason TEXT NOT NULL DEFAULT '', + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + created_by TEXT NOT NULL DEFAULT '', + reversed_at TIMESTAMP, + reversed_by TEXT NOT NULL DEFAULT '' +); +CREATE INDEX IF NOT EXISTS idx_takedowns_active ON takedowns(reversed_at, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_takedowns_subject ON takedowns(subject_did, subject_repo); CREATE TABLE IF NOT EXISTS labels ( id INTEGER PRIMARY KEY AUTOINCREMENT, src TEXT NOT NULL, @@ -31,16 +45,21 @@ CREATE TABLE IF NOT EXISTS labels ( ver INTEGER NOT NULL DEFAULT 1, sig BLOB NOT NULL, subject_did TEXT NOT NULL, - subject_repo TEXT NOT NULL DEFAULT '' + subject_repo TEXT NOT NULL DEFAULT '', + takedown_id INTEGER REFERENCES takedowns(id) ); CREATE INDEX IF NOT EXISTS idx_labels_subject ON labels(subject_did, subject_repo); CREATE INDEX IF NOT EXISTS idx_labels_cts ON labels(cts DESC); CREATE INDEX IF NOT EXISTS idx_labels_uri ON labels(uri); +CREATE INDEX IF NOT EXISTS idx_labels_takedown ON labels(takedown_id); ` // Label represents an ATProto label record stored locally. Its on-the-wire representation // is produced by ToLabeling() which round-trips through indigo's labeling package so the // signature stays valid byte-for-byte. +// +// TakedownID is a labeler-internal pointer to the takedown event that produced this +// label (positive or negation). It's never serialized into ATProto wire format. type Label struct { ID int64 Src string @@ -54,6 +73,23 @@ type Label struct { Sig []byte SubjectDID string SubjectRepo string + TakedownID *int64 +} + +// Takedown is a single operator-issued takedown action. Each Takedown owns one or more +// Label rows linked by takedown_id. Reversal sets reversed_at / reversed_by in place. +type Takedown struct { + ID int64 + Input string + SubjectDID string + SubjectRepo string + SubjectHandle string + Reason string + CreatedAt time.Time + CreatedBy string + ReversedAt *time.Time + ReversedBy string + LabelCount int } // LibsqlSync configures optional embedded-replica sync to a remote libSQL database. @@ -233,12 +269,16 @@ func CreateLabel(db *sql.DB, l *Label) (int64, error) { s := l.Exp.UTC().Format(time.RFC3339) expStr = &s } + var takedownID any + if l.TakedownID != nil { + takedownID = *l.TakedownID + } result, err := db.Exec( - `INSERT INTO labels (src, uri, cid, val, neg, cts, exp, ver, sig, subject_did, subject_repo) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + `INSERT INTO labels (src, uri, cid, val, neg, cts, exp, ver, sig, subject_did, subject_repo, takedown_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, l.Src, l.URI, nullableString(l.CID), l.Val, l.Neg, l.Cts.UTC().Format(time.RFC3339), expStr, l.Ver, l.Sig, - l.SubjectDID, l.SubjectRepo, + l.SubjectDID, l.SubjectRepo, takedownID, ) if err != nil { return 0, fmt.Errorf("failed to insert label: %w", err) @@ -261,7 +301,7 @@ func nullableString(s string) any { // GetLabelsSince returns labels with id > cursor, ordered by id ascending. func GetLabelsSince(db *sql.DB, cursor int64, limit int) ([]Label, error) { rows, err := db.Query( - `SELECT id, src, uri, COALESCE(cid, ''), val, neg, cts, exp, ver, sig, subject_did, subject_repo + `SELECT id, src, uri, COALESCE(cid, ''), val, neg, cts, exp, ver, sig, subject_did, subject_repo, takedown_id FROM labels WHERE id > ? ORDER BY id ASC LIMIT ?`, cursor, limit, ) @@ -284,34 +324,73 @@ func LatestSeq(db *sql.DB) (int64, error) { return seq.Int64, nil } -// ListActiveTakedowns returns active (non-negated) takedown labels. -func ListActiveTakedowns(db *sql.DB, limit, offset int) ([]Label, int, error) { - var total int - err := db.QueryRow( - `SELECT COUNT(*) FROM labels l1 - WHERE l1.val = '!takedown' AND l1.neg = 0 - AND NOT EXISTS ( - SELECT 1 FROM labels l2 - WHERE l2.src = l1.src AND l2.uri = l1.uri AND l2.val = l1.val - AND l2.neg = 1 AND l2.id > l1.id - ) - AND (l1.exp IS NULL OR l1.exp > CURRENT_TIMESTAMP)`, - ).Scan(&total) +// CreateTakedown inserts a takedown event row and returns its id. The id should then +// be stamped onto every label produced by this takedown (positive labels at issue time, +// negation labels at reversal time) so the audit trail stays linked. +func CreateTakedown(db *sql.DB, t *Takedown) (int64, error) { + if t.CreatedAt.IsZero() { + t.CreatedAt = time.Now().UTC() + } + result, err := db.Exec( + `INSERT INTO takedowns (input, subject_did, subject_repo, subject_handle, reason, created_at, created_by) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + t.Input, t.SubjectDID, t.SubjectRepo, t.SubjectHandle, t.Reason, + t.CreatedAt.UTC().Format(time.RFC3339), t.CreatedBy, + ) if err != nil { + return 0, fmt.Errorf("failed to insert takedown: %w", err) + } + id, err := result.LastInsertId() + if err != nil { + return 0, err + } + t.ID = id + return id, nil +} + +// GetTakedown loads a single takedown row by id. Returns sql.ErrNoRows when missing. +func GetTakedown(db *sql.DB, id int64) (*Takedown, error) { + row := db.QueryRow( + `SELECT t.id, t.input, t.subject_did, t.subject_repo, t.subject_handle, t.reason, + t.created_at, t.created_by, t.reversed_at, t.reversed_by, + (SELECT COUNT(*) FROM labels l WHERE l.takedown_id = t.id AND l.neg = 0) + FROM takedowns t WHERE t.id = ?`, + id, + ) + return scanTakedown(row.Scan) +} + +// TakedownFilter scopes ListTakedowns to active, reversed, or all rows. +type TakedownFilter int + +const ( + TakedownAll TakedownFilter = iota // every takedown row, regardless of reversal state + TakedownActive // only takedowns whose reversed_at is NULL + TakedownReversed // only takedowns whose reversed_at is set +) + +// ListTakedowns returns takedown events ordered by created_at DESC, scoped by filter. +// The total count reflects the same filter. +func ListTakedowns(db *sql.DB, filter TakedownFilter, limit, offset int) ([]Takedown, int, error) { + where := "" + switch filter { + case TakedownActive: + where = "WHERE reversed_at IS NULL" + case TakedownReversed: + where = "WHERE reversed_at IS NOT NULL" + } + + var total int + if err := db.QueryRow(`SELECT COUNT(*) FROM takedowns ` + where).Scan(&total); err != nil { return nil, 0, err } rows, err := db.Query( - `SELECT l1.id, l1.src, l1.uri, COALESCE(l1.cid, ''), l1.val, l1.neg, l1.cts, l1.exp, l1.ver, l1.sig, l1.subject_did, l1.subject_repo - FROM labels l1 - WHERE l1.val = '!takedown' AND l1.neg = 0 - AND NOT EXISTS ( - SELECT 1 FROM labels l2 - WHERE l2.src = l1.src AND l2.uri = l1.uri AND l2.val = l1.val - AND l2.neg = 1 AND l2.id > l1.id - ) - AND (l1.exp IS NULL OR l1.exp > CURRENT_TIMESTAMP) - ORDER BY l1.cts DESC LIMIT ? OFFSET ?`, + `SELECT t.id, t.input, t.subject_did, t.subject_repo, t.subject_handle, t.reason, + t.created_at, t.created_by, t.reversed_at, t.reversed_by, + (SELECT COUNT(*) FROM labels l WHERE l.takedown_id = t.id AND l.neg = 0) + FROM takedowns t `+where+` + ORDER BY t.created_at DESC LIMIT ? OFFSET ?`, limit, offset, ) if err != nil { @@ -319,18 +398,47 @@ func ListActiveTakedowns(db *sql.DB, limit, offset int) ([]Label, int, error) { } defer rows.Close() - labels, err := scanLabels(rows) - return labels, total, err + var out []Takedown + for rows.Next() { + t, err := scanTakedown(rows.Scan) + if err != nil { + return nil, 0, err + } + out = append(out, *t) + } + return out, total, rows.Err() } -// GetLabelsForRepo returns all labels for a specific DID + repository. -func GetLabelsForRepo(db *sql.DB, did, repo string) ([]Label, error) { +// MarkTakedownReversed sets the reversed_at / reversed_by fields on the takedown row. +// Refuses to overwrite an existing reversal. +func MarkTakedownReversed(db *sql.DB, id int64, by string, at time.Time) error { + if at.IsZero() { + at = time.Now().UTC() + } + res, err := db.Exec( + `UPDATE takedowns SET reversed_at = ?, reversed_by = ? + WHERE id = ? AND reversed_at IS NULL`, + at.UTC().Format(time.RFC3339), by, id, + ) + if err != nil { + return fmt.Errorf("failed to mark takedown reversed: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return err + } + if n == 0 { + return fmt.Errorf("takedown %d not found or already reversed", id) + } + return nil +} + +// GetLabelsByTakedown returns all labels (positive + negations) linked to a takedown. +func GetLabelsByTakedown(db *sql.DB, takedownID int64) ([]Label, error) { rows, err := db.Query( - `SELECT id, src, uri, COALESCE(cid, ''), val, neg, cts, exp, ver, sig, subject_did, subject_repo - FROM labels - WHERE subject_did = ? AND subject_repo = ? - ORDER BY cts DESC`, - did, repo, + `SELECT id, src, uri, COALESCE(cid, ''), val, neg, cts, exp, ver, sig, subject_did, subject_repo, takedown_id + FROM labels WHERE takedown_id = ? ORDER BY id ASC`, + takedownID, ) if err != nil { return nil, err @@ -339,75 +447,36 @@ func GetLabelsForRepo(db *sql.DB, did, repo string) ([]Label, error) { return scanLabels(rows) } -// newNegationLabel constructs an unsigned negation label awaiting Sign(). -func newNegationLabel(src, uri, val, did, repo string) *Label { - return &Label{ - Src: src, - URI: uri, - Val: val, - Neg: true, - Cts: time.Now().UTC(), - SubjectDID: did, - SubjectRepo: repo, - } -} - -// NegateRepoLabels signs+inserts negation labels for all active takedown labels on (DID, repo). -func NegateRepoLabels(db *sql.DB, key *atcrypto.PrivateKeyK256, src, did, repo string) ([]Label, error) { +// NegateTakedownLabels signs+inserts negation labels for every active (non-negated) +// label linked to the given takedown_id. Negations carry the same takedown_id so they +// remain part of the takedown's audit trail. +// +// The NOT EXISTS subquery skips URIs that already have a later neg=1 row (from a prior +// reversal call or from an external negation streamed in via subscribeLabels), so this +// function is idempotent and won't emit duplicate negations. +func NegateTakedownLabels(db *sql.DB, key *atcrypto.PrivateKeyK256, src string, takedownID int64) ([]Label, error) { rows, err := db.Query( - `SELECT uri FROM labels - WHERE subject_did = ? AND subject_repo = ? AND val = '!takedown' AND neg = 0`, - did, repo, + `SELECT l1.uri, l1.subject_did, l1.subject_repo FROM labels l1 + WHERE l1.takedown_id = ? AND l1.val = '!takedown' AND l1.neg = 0 + AND NOT EXISTS ( + SELECT 1 FROM labels l2 + WHERE l2.src = l1.src AND l2.uri = l1.uri AND l2.val = l1.val + AND l2.neg = 1 AND l2.id > l1.id + )`, + takedownID, ) if err != nil { return nil, err } - var uris []string - for rows.Next() { - var uri string - if err := rows.Scan(&uri); err != nil { - rows.Close() - return nil, err - } - uris = append(uris, uri) - } - rows.Close() - if err := rows.Err(); err != nil { - return nil, err - } - - out := make([]Label, 0, len(uris)) - for _, uri := range uris { - neg := newNegationLabel(src, uri, "!takedown", did, repo) - if err := neg.Sign(key); err != nil { - return out, err - } - if _, err := CreateLabel(db, neg); err != nil { - return out, err - } - out = append(out, *neg) - } - return out, nil -} - -// NegateUserLabels signs+inserts negation labels for all active takedown labels on a DID. -func NegateUserLabels(db *sql.DB, key *atcrypto.PrivateKeyK256, src, did string) ([]Label, error) { - rows, err := db.Query( - `SELECT uri, subject_repo FROM labels - WHERE subject_did = ? AND val = '!takedown' AND neg = 0`, - did, - ) - if err != nil { - return nil, err - } - type uriRepo struct { + type entry struct { uri string + did string repo string } - var entries []uriRepo + var entries []entry for rows.Next() { - var e uriRepo - if err := rows.Scan(&e.uri, &e.repo); err != nil { + var e entry + if err := rows.Scan(&e.uri, &e.did, &e.repo); err != nil { rows.Close() return nil, err } @@ -418,9 +487,19 @@ func NegateUserLabels(db *sql.DB, key *atcrypto.PrivateKeyK256, src, did string) return nil, err } + id := takedownID out := make([]Label, 0, len(entries)) for _, e := range entries { - neg := newNegationLabel(src, e.uri, "!takedown", did, e.repo) + neg := &Label{ + Src: src, + URI: e.uri, + Val: "!takedown", + Neg: true, + Cts: time.Now().UTC(), + SubjectDID: e.did, + SubjectRepo: e.repo, + TakedownID: &id, + } if err := neg.Sign(key); err != nil { return out, err } @@ -432,13 +511,39 @@ func NegateUserLabels(db *sql.DB, key *atcrypto.PrivateKeyK256, src, did string) return out, nil } +func scanTakedown(scan func(...any) error) (*Takedown, error) { + var ( + t Takedown + created string + revAt *string + ) + if err := scan( + &t.ID, &t.Input, &t.SubjectDID, &t.SubjectRepo, &t.SubjectHandle, &t.Reason, + &created, &t.CreatedBy, &revAt, &t.ReversedBy, &t.LabelCount, + ); err != nil { + return nil, err + } + if ts, err := time.Parse(time.RFC3339, created); err == nil { + t.CreatedAt = ts + } + if revAt != nil { + if ts, err := time.Parse(time.RFC3339, *revAt); err == nil { + t.ReversedAt = &ts + } + } + return &t, nil +} + func scanLabels(rows *sql.Rows) ([]Label, error) { var labels []Label for rows.Next() { - var l Label - var cts string - var exp *string - if err := rows.Scan(&l.ID, &l.Src, &l.URI, &l.CID, &l.Val, &l.Neg, &cts, &exp, &l.Ver, &l.Sig, &l.SubjectDID, &l.SubjectRepo); err != nil { + var ( + l Label + cts string + exp *string + tdID sql.NullInt64 + ) + if err := rows.Scan(&l.ID, &l.Src, &l.URI, &l.CID, &l.Val, &l.Neg, &cts, &exp, &l.Ver, &l.Sig, &l.SubjectDID, &l.SubjectRepo, &tdID); err != nil { return nil, err } if t, err := time.Parse(time.RFC3339, cts); err == nil { @@ -449,6 +554,10 @@ func scanLabels(rows *sql.Rows) ([]Label, error) { l.Exp = &t } } + if tdID.Valid { + id := tdID.Int64 + l.TakedownID = &id + } labels = append(labels, l) } return labels, rows.Err() diff --git a/pkg/labeler/db_test.go b/pkg/labeler/db_test.go index 8b2c727..9604a7e 100644 --- a/pkg/labeler/db_test.go +++ b/pkg/labeler/db_test.go @@ -151,7 +151,7 @@ func TestSignAndVerify(t *testing.T) { } } -func TestListActiveTakedowns(t *testing.T) { +func TestListTakedowns(t *testing.T) { dir := t.TempDir() db := openTestDB(t, filepath.Join(dir, "test.db")) key := newTestKey(t) @@ -159,36 +159,70 @@ func TestListActiveTakedowns(t *testing.T) { src := "did:plc:labeler-1" now := time.Now().UTC() + // Create three takedown events, each with a single summary label, so the + // label_count subquery has something to count. + var ids []int64 for i, repo := range []string{"repo1", "repo2", "repo3"} { + td := &Takedown{ + Input: "atcr.io/r/did:plc:abc/" + repo, + SubjectDID: "did:plc:abc", + SubjectRepo: repo, + CreatedAt: now.Add(time.Duration(i) * time.Minute), + } + id, err := CreateTakedown(db, td) + if err != nil { + t.Fatalf("CreateTakedown: %v", err) + } + ids = append(ids, id) signAndCreate(t, db, key, &Label{ Src: src, URI: "at://did:plc:abc/io.atcr.repo/" + repo, - Val: "!takedown", Cts: now.Add(time.Duration(i) * time.Minute), + Val: "!takedown", Cts: td.CreatedAt, SubjectDID: "did:plc:abc", SubjectRepo: repo, + TakedownID: &id, }) } - labels, total, err := ListActiveTakedowns(db, 10, 0) + tds, total, err := ListTakedowns(db, TakedownActive, 10, 0) if err != nil { t.Fatal(err) } - if total != 3 || len(labels) != 3 { - t.Errorf("expected 3 active takedowns, got total=%d returned=%d", total, len(labels)) + if total != 3 || len(tds) != 3 { + t.Errorf("expected 3 active takedowns, got total=%d returned=%d", total, len(tds)) + } + for _, td := range tds { + if td.LabelCount != 1 { + t.Errorf("takedown %d label_count = %d, want 1", td.ID, td.LabelCount) + } } - if _, err := NegateRepoLabels(db, key, src, "did:plc:abc", "repo2"); err != nil { + // Reverse the middle takedown. + if _, err := NegateTakedownLabels(db, key, src, ids[1]); err != nil { + t.Fatal(err) + } + if err := MarkTakedownReversed(db, ids[1], "did:plc:operator", time.Now().UTC()); err != nil { t.Fatal(err) } - _, total, err = ListActiveTakedowns(db, 10, 0) + _, activeTotal, err := ListTakedowns(db, TakedownActive, 10, 0) if err != nil { t.Fatal(err) } - if total != 2 { - t.Errorf("expected 2 active takedowns after negation, got %d", total) + if activeTotal != 2 { + t.Errorf("expected 2 active takedowns after reversal, got %d", activeTotal) + } + revs, revTotal, err := ListTakedowns(db, TakedownReversed, 10, 0) + if err != nil { + t.Fatal(err) + } + if revTotal != 1 || len(revs) != 1 { + t.Errorf("expected 1 reversed takedown, got total=%d returned=%d", revTotal, len(revs)) + } + if revs[0].ID != ids[1] || revs[0].ReversedAt == nil || revs[0].ReversedBy != "did:plc:operator" { + t.Errorf("reversed takedown row has wrong fields: %+v", revs[0]) } } -func TestNegateRepoLabels(t *testing.T) { +func TestNegateTakedownLabels(t *testing.T) { dir := t.TempDir() db := openTestDB(t, filepath.Join(dir, "test.db")) key := newTestKey(t) @@ -197,6 +231,14 @@ func TestNegateRepoLabels(t *testing.T) { now := time.Now().UTC() did := "did:plc:abc" + tdID, err := CreateTakedown(db, &Takedown{ + Input: "atcr.io/r/did:plc:abc/myimage", SubjectDID: did, SubjectRepo: "myimage", + CreatedAt: now, + }) + if err != nil { + t.Fatal(err) + } + uris := []string{ "at://did:plc:abc/io.atcr.manifest/sha256-111", "at://did:plc:abc/io.atcr.manifest/sha256-222", @@ -205,11 +247,11 @@ func TestNegateRepoLabels(t *testing.T) { for _, uri := range uris { signAndCreate(t, db, key, &Label{ Src: src, URI: uri, Val: "!takedown", Cts: now, - SubjectDID: did, SubjectRepo: "myimage", + SubjectDID: did, SubjectRepo: "myimage", TakedownID: &tdID, }) } - negs, err := NegateRepoLabels(db, key, src, did, "myimage") + negs, err := NegateTakedownLabels(db, key, src, tdID) if err != nil { t.Fatal(err) } @@ -217,47 +259,57 @@ func TestNegateRepoLabels(t *testing.T) { t.Errorf("expected %d negation labels, got %d", len(uris), len(negs)) } - _, total, err := ListActiveTakedowns(db, 10, 0) + // Negations must carry the same takedown_id so they're part of the audit trail. + all, err := GetLabelsByTakedown(db, tdID) if err != nil { t.Fatal(err) } - if total != 0 { - t.Errorf("expected 0 active takedowns after repo negation, got %d", total) + if len(all) != 2*len(uris) { + t.Errorf("expected %d labels (positive + negation) for takedown %d, got %d", 2*len(uris), tdID, len(all)) + } + var pos, neg int + for _, l := range all { + if l.TakedownID == nil || *l.TakedownID != tdID { + t.Errorf("label %d takedown_id = %v, want %d", l.ID, l.TakedownID, tdID) + } + if l.Neg { + neg++ + } else { + pos++ + } + } + if pos != len(uris) || neg != len(uris) { + t.Errorf("expected pos=%d neg=%d, got pos=%d neg=%d", len(uris), len(uris), pos, neg) + } + + // Calling negate again must be a no-op (no remaining positive labels to flip). + negs2, err := NegateTakedownLabels(db, key, src, tdID) + if err != nil { + t.Fatal(err) + } + if len(negs2) != 0 { + t.Errorf("expected 0 negations on second call, got %d", len(negs2)) } } -func TestNegateUserLabels(t *testing.T) { +func TestMarkTakedownReversed_RefusesDoubleReverse(t *testing.T) { dir := t.TempDir() db := openTestDB(t, filepath.Join(dir, "test.db")) - key := newTestKey(t) - src := "did:plc:labeler-1" - now := time.Now().UTC() - did := "did:plc:abc" - - signAndCreate(t, db, key, &Label{ - Src: src, URI: "at://did:plc:abc", Val: "!takedown", Cts: now, - SubjectDID: did, + id, err := CreateTakedown(db, &Takedown{ + Input: "did:plc:abc", SubjectDID: "did:plc:abc", }) - signAndCreate(t, db, key, &Label{ - Src: src, URI: "at://did:plc:abc/io.atcr.repo/repo1", Val: "!takedown", Cts: now, - SubjectDID: did, SubjectRepo: "repo1", - }) - - negs, err := NegateUserLabels(db, key, src, did) if err != nil { t.Fatal(err) } - if len(negs) != 2 { - t.Errorf("expected 2 negation labels, got %d", len(negs)) + if err := MarkTakedownReversed(db, id, "did:plc:op", time.Now().UTC()); err != nil { + t.Fatalf("first reverse: %v", err) } - - _, total, err := ListActiveTakedowns(db, 10, 0) - if err != nil { - t.Fatal(err) + if err := MarkTakedownReversed(db, id, "did:plc:op", time.Now().UTC()); err == nil { + t.Error("expected second reverse to fail (already reversed)") } - if total != 0 { - t.Errorf("expected 0 active takedowns after user negation, got %d", total) + if err := MarkTakedownReversed(db, 9999, "did:plc:op", time.Now().UTC()); err == nil { + t.Error("expected reverse on unknown id to fail") } } @@ -326,7 +378,7 @@ func TestLatestSeq(t *testing.T) { } } -func TestGetLabelsForRepo(t *testing.T) { +func TestGetLabelsByTakedown(t *testing.T) { dir := t.TempDir() db := openTestDB(t, filepath.Join(dir, "test.db")) key := newTestKey(t) @@ -334,40 +386,54 @@ func TestGetLabelsForRepo(t *testing.T) { src := "did:plc:labeler-1" now := time.Now().UTC() + tdA, err := CreateTakedown(db, &Takedown{Input: "did:plc:abc/repo1", SubjectDID: "did:plc:abc", SubjectRepo: "repo1", CreatedAt: now}) + if err != nil { + t.Fatal(err) + } + tdB, err := CreateTakedown(db, &Takedown{Input: "did:plc:def/repo1", SubjectDID: "did:plc:def", SubjectRepo: "repo1", CreatedAt: now}) + if err != nil { + t.Fatal(err) + } + signAndCreate(t, db, key, &Label{ Src: src, URI: "at://did:plc:abc/io.atcr.repo/repo1", - Val: "!takedown", Cts: now, SubjectDID: "did:plc:abc", SubjectRepo: "repo1", + Val: "!takedown", Cts: now, SubjectDID: "did:plc:abc", SubjectRepo: "repo1", TakedownID: &tdA, }) signAndCreate(t, db, key, &Label{ - Src: src, URI: "at://did:plc:abc/io.atcr.repo/repo2", - Val: "!takedown", Cts: now, SubjectDID: "did:plc:abc", SubjectRepo: "repo2", + Src: src, URI: "at://did:plc:abc/io.atcr.manifest/sha256-aaa", + Val: "!takedown", Cts: now, SubjectDID: "did:plc:abc", SubjectRepo: "repo1", TakedownID: &tdA, }) signAndCreate(t, db, key, &Label{ Src: src, URI: "at://did:plc:def/io.atcr.repo/repo1", - Val: "!takedown", Cts: now, SubjectDID: "did:plc:def", SubjectRepo: "repo1", + Val: "!takedown", Cts: now, SubjectDID: "did:plc:def", SubjectRepo: "repo1", TakedownID: &tdB, }) - labels, err := GetLabelsForRepo(db, "did:plc:abc", "repo1") + labels, err := GetLabelsByTakedown(db, tdA) + if err != nil { + t.Fatal(err) + } + if len(labels) != 2 { + t.Errorf("takedown A: expected 2 labels, got %d", len(labels)) + } + for _, l := range labels { + if l.TakedownID == nil || *l.TakedownID != tdA { + t.Errorf("label %d has wrong takedown_id: %v", l.ID, l.TakedownID) + } + } + + labels, err = GetLabelsByTakedown(db, tdB) if err != nil { t.Fatal(err) } if len(labels) != 1 { - t.Errorf("expected 1 label for did:plc:abc/repo1, got %d", len(labels)) + t.Errorf("takedown B: expected 1 label, got %d", len(labels)) } - labels, err = GetLabelsForRepo(db, "did:plc:def", "repo1") - if err != nil { - t.Fatal(err) - } - if len(labels) != 1 { - t.Errorf("expected 1 label for did:plc:def/repo1, got %d", len(labels)) - } - - labels, err = GetLabelsForRepo(db, "did:plc:xyz", "repo1") + labels, err = GetLabelsByTakedown(db, 9999) if err != nil { t.Fatal(err) } if len(labels) != 0 { - t.Errorf("expected 0 labels for unknown did, got %d", len(labels)) + t.Errorf("unknown takedown id: expected 0 labels, got %d", len(labels)) } } diff --git a/pkg/labeler/takedown.go b/pkg/labeler/takedown.go index c9718c4..968f264 100644 --- a/pkg/labeler/takedown.go +++ b/pkg/labeler/takedown.go @@ -8,6 +8,7 @@ import ( "log/slog" "net/http" "net/url" + "strconv" "strings" "time" @@ -19,6 +20,11 @@ type TakedownInput struct { DID string Handle string Repository string // empty = user-level takedown + // Operator-supplied context. Captured into the takedowns row so we can show + // who/why/what-was-typed on the dashboard. None are required. + RawInput string // exact string the operator submitted (URL, did, handle, AT URI) + Reason string // optional free-text note + CreatedBy string // operator DID from session, "" if unknown } // ParseTakedownInput parses various input formats into a TakedownInput. @@ -165,6 +171,7 @@ func resolveIdentifier(ctx context.Context, identifier string) (did, handle stri // TakedownResult contains the results of a takedown operation. type TakedownResult struct { + TakedownID int64 DID string Handle string Repository string @@ -172,11 +179,39 @@ type TakedownResult struct { UserLevel bool } -// ExecuteTakedown creates takedown labels for a repo or user. +// ExecuteTakedown creates a takedown event row and the labels that belong to it. +// Every label (the user-level label, the per-record labels discovered via PDS, and the +// repo-level summary label) carries the new takedown_id so reversal can target the +// exact set without re-querying by subject. func (s *Server) ExecuteTakedown(ctx context.Context, input *TakedownInput) (*TakedownResult, error) { src := s.did now := time.Now().UTC() + + td := &Takedown{ + Input: input.RawInput, + SubjectDID: input.DID, + SubjectRepo: input.Repository, + SubjectHandle: input.Handle, + Reason: input.Reason, + CreatedAt: now, + CreatedBy: input.CreatedBy, + } + if td.Input == "" { + // Fallback so the dashboard always has something to show, even if a caller + // (e.g. a future API) didn't pass the original string. + if input.Repository != "" { + td.Input = fmt.Sprintf("%s/%s", input.DID, input.Repository) + } else { + td.Input = input.DID + } + } + takedownID, err := CreateTakedown(s.db, td) + if err != nil { + return nil, fmt.Errorf("failed to create takedown event: %w", err) + } + result := &TakedownResult{ + TakedownID: takedownID, DID: input.DID, Handle: input.Handle, Repository: input.Repository, @@ -184,7 +219,7 @@ func (s *Server) ExecuteTakedown(ctx context.Context, input *TakedownInput) (*Ta } if input.Repository == "" { - // User-level takedown + // User-level takedown: a single label on at://. label := &Label{ Src: src, URI: "at://" + input.DID, @@ -192,6 +227,7 @@ func (s *Server) ExecuteTakedown(ctx context.Context, input *TakedownInput) (*Ta Cts: now, SubjectDID: input.DID, SubjectRepo: "", + TakedownID: &takedownID, } if err := label.Sign(s.signingKey); err != nil { return nil, fmt.Errorf("failed to sign user-level label: %w", err) @@ -201,19 +237,20 @@ func (s *Server) ExecuteTakedown(ctx context.Context, input *TakedownInput) (*Ta } s.hub.Broadcast(label) result.Labels = append(result.Labels, *label) - slog.Info("Created user-level takedown", "did", input.DID, "handle", input.Handle) + slog.Info("Created user-level takedown", "takedown_id", takedownID, "did", input.DID, "handle", input.Handle) return result, nil } - // Repo-level takedown: discover all records from PDS - labels, err := s.discoverAndLabelRecords(ctx, input.DID, input.Repository, src, now) + // Repo-level takedown: discover all records from PDS and label each. + labels, err := s.discoverAndLabelRecords(ctx, input.DID, input.Repository, src, now, takedownID) if err != nil { - // Even if PDS discovery fails, create a repo-level summary label + // Even if PDS discovery fails, create a repo-level summary label so reads + // against the well-known summary URI still see the takedown. slog.Warn("PDS discovery failed, creating summary label only", "error", err) } result.Labels = append(result.Labels, labels...) - // Always create a repo-level summary label for efficient filtering + // Always create a repo-level summary label for efficient filtering. summaryLabel := &Label{ Src: src, URI: fmt.Sprintf("at://%s/io.atcr.repo/%s", input.DID, input.Repository), @@ -221,6 +258,7 @@ func (s *Server) ExecuteTakedown(ctx context.Context, input *TakedownInput) (*Ta Cts: now, SubjectDID: input.DID, SubjectRepo: input.Repository, + TakedownID: &takedownID, } if err := summaryLabel.Sign(s.signingKey); err != nil { return nil, fmt.Errorf("failed to sign summary label: %w", err) @@ -232,6 +270,7 @@ func (s *Server) ExecuteTakedown(ctx context.Context, input *TakedownInput) (*Ta result.Labels = append(result.Labels, *summaryLabel) slog.Info("Created repo-level takedown", + "takedown_id", takedownID, "did", input.DID, "handle", input.Handle, "repository", input.Repository, @@ -241,9 +280,49 @@ func (s *Server) ExecuteTakedown(ctx context.Context, input *TakedownInput) (*Ta return result, nil } +// ReverseTakedown negates every active label belonging to the given takedown event and +// marks the event row as reversed. Refuses to act on a takedown that doesn't exist or +// has already been reversed. +func (s *Server) ReverseTakedown(ctx context.Context, takedownID int64, reversedBy string) (*TakedownResult, error) { + td, err := GetTakedown(s.db, takedownID) + if err != nil { + return nil, fmt.Errorf("failed to load takedown %d: %w", takedownID, err) + } + if td.ReversedAt != nil { + return nil, fmt.Errorf("takedown %d already reversed at %s", takedownID, td.ReversedAt.Format(time.RFC3339)) + } + + negs, err := NegateTakedownLabels(s.db, s.signingKey, s.did, takedownID) + if err != nil { + return nil, fmt.Errorf("failed to negate labels for takedown %d: %w", takedownID, err) + } + for i := range negs { + s.hub.Broadcast(&negs[i]) + } + if err := MarkTakedownReversed(s.db, takedownID, reversedBy, time.Now().UTC()); err != nil { + return nil, err + } + + slog.Info("Reversed takedown", + "takedown_id", takedownID, + "did", td.SubjectDID, + "repository", td.SubjectRepo, + "reversed_by", reversedBy, + "negations", len(negs), + ) + return &TakedownResult{ + TakedownID: takedownID, + DID: td.SubjectDID, + Handle: td.SubjectHandle, + Repository: td.SubjectRepo, + Labels: negs, + UserLevel: td.SubjectRepo == "", + }, nil +} + // discoverAndLabelRecords queries the user's PDS for all records in the given repo -// and creates takedown labels for each. -func (s *Server) discoverAndLabelRecords(ctx context.Context, did, repo, src string, now time.Time) ([]Label, error) { +// and creates takedown labels for each, all linked to takedownID. +func (s *Server) discoverAndLabelRecords(ctx context.Context, did, repo, src string, now time.Time, takedownID int64) ([]Label, error) { _, _, pdsEndpoint, err := atproto.ResolveIdentity(ctx, did) if err != nil { return nil, fmt.Errorf("failed to resolve DID: %w", err) @@ -282,6 +361,7 @@ func (s *Server) discoverAndLabelRecords(ctx context.Context, did, repo, src str Cts: now, SubjectDID: did, SubjectRepo: repo, + TakedownID: &takedownID, } if err := label.Sign(s.signingKey); err != nil { slog.Warn("Failed to sign label", "uri", uri, "error", err) @@ -314,11 +394,17 @@ func extractRepoField(value json.RawMessage, collection string) string { // Handlers func (s *Server) handleDashboard(w http.ResponseWriter, r *http.Request) { - labels, total, err := ListActiveTakedowns(s.db, 50, 0) + active, activeTotal, err := ListTakedowns(s.db, TakedownActive, 50, 0) if err != nil { http.Error(w, "Failed to list takedowns", http.StatusInternalServerError) return } + reversed, reversedTotal, err := ListTakedowns(s.db, TakedownReversed, 50, 0) + if err != nil { + http.Error(w, "Failed to list reversed takedowns", http.StatusInternalServerError) + return + } + csrf := "" if session := SessionFromContext(r.Context()); session != nil { csrf = session.CSRFToken @@ -329,16 +415,18 @@ func (s *Server) handleDashboard(w http.ResponseWriter, r *http.Request) { %s Labeler @@ -351,40 +439,89 @@ form{display:inline}

Active Takedowns (%d)

`, s.config.Server.ClientShortName, s.config.Server.ClientShortName, - total, + activeTotal, ) - if len(labels) == 0 { - fmt.Fprint(w, `

No active takedowns.

`) - } else { - fmt.Fprint(w, ``) - for _, l := range labels { - repoDisplay := l.SubjectRepo - if repoDisplay == "" { - repoDisplay = "all repos (user-level)" - } - fmt.Fprintf(w, ` - - - - - -`, - template.HTMLEscapeString(l.SubjectDID), - repoDisplay, - template.HTMLEscapeString(l.URI), - l.Cts.Format("2006-01-02 15:04"), - csrfInputHTML(csrf), - template.HTMLEscapeString(l.SubjectDID), - template.HTMLEscapeString(l.SubjectRepo), - ) - } - fmt.Fprint(w, `
SubjectRepositoryURICreatedAction
%s%s%s%s
%s
`) - } + renderTakedownRows(w, active, csrf, true) + + fmt.Fprintf(w, `

Reversed (%d)

`, reversedTotal) + renderTakedownRows(w, reversed, csrf, false) fmt.Fprint(w, ``) } +// renderTakedownRows writes either an active table (with a Reverse button) or a +// reversed-history table (with a reversed-at column instead). +func renderTakedownRows(w http.ResponseWriter, ts []Takedown, csrf string, withReverse bool) { + if len(ts) == 0 { + if withReverse { + fmt.Fprint(w, `

No active takedowns.

`) + } else { + fmt.Fprint(w, `

No reversed takedowns yet.

`) + } + return + } + fmt.Fprint(w, ``) + if withReverse { + fmt.Fprint(w, ``) + } else { + fmt.Fprint(w, ``) + } + fmt.Fprint(w, ``) + for _, t := range ts { + subject := template.HTMLEscapeString(t.SubjectDID) + if t.SubjectHandle != "" { + subject = fmt.Sprintf(`%s
%s`, + template.HTMLEscapeString(t.SubjectHandle), subject) + } + if t.SubjectRepo != "" { + subject += fmt.Sprintf(` / %s`, template.HTMLEscapeString(t.SubjectRepo)) + } else { + subject += ` (user-level)` + } + + reason := template.HTMLEscapeString(t.Reason) + if reason == "" { + reason = `—` + } + + var lastCol string + if withReverse { + lastCol = fmt.Sprintf( + `%s`, + csrfInputHTML(csrf), t.ID, + ) + } else { + rev := "" + if t.ReversedAt != nil { + rev = t.ReversedAt.Format("2006-01-02 15:04") + } + by := "" + if t.ReversedBy != "" { + by = fmt.Sprintf(`
by %s`, template.HTMLEscapeString(t.ReversedBy)) + } + lastCol = rev + by + } + + fmt.Fprintf(w, ` + + + + + + +`, + template.HTMLEscapeString(t.Input), + subject, + reason, + t.LabelCount, + t.CreatedAt.Format("2006-01-02 15:04"), + lastCol, + ) + } + fmt.Fprint(w, `
InputSubjectReasonLabelsCreatedActionReversed
%s%s%s%d%s%s
`) +} + func (s *Server) handleTakedownForm(w http.ResponseWriter, r *http.Request) { msg := r.URL.Query().Get("msg") errorMsg := r.URL.Query().Get("error") @@ -431,6 +568,10 @@ nav{display:flex;gap:16px;margin-bottom:24px}

Repo: /r/handle/repo (or full atcr.io URL). User-level: /u/handle, a bare handle, or a DID. AT URIs (at://...) also work.

+ + + +
@@ -443,12 +584,18 @@ func (s *Server) handleTakedownSubmit(w http.ResponseWriter, r *http.Request) { http.Redirect(w, r, "/takedown?error=Target+is+required", http.StatusFound) return } + reason := strings.TrimSpace(r.FormValue("reason")) input, err := ParseTakedownInput(r.Context(), target) if err != nil { http.Redirect(w, r, "/takedown?error="+strings.ReplaceAll(err.Error(), " ", "+"), http.StatusFound) return } + input.RawInput = target + input.Reason = reason + if session := SessionFromContext(r.Context()); session != nil { + input.CreatedBy = session.DID + } result, err := s.ExecuteTakedown(r.Context(), input) if err != nil { @@ -456,7 +603,7 @@ func (s *Server) handleTakedownSubmit(w http.ResponseWriter, r *http.Request) { return } - msg := fmt.Sprintf("Takedown issued: %d labels created for %s", len(result.Labels), result.DID) + msg := fmt.Sprintf("Takedown #%d issued: %d labels created for %s", result.TakedownID, len(result.Labels), result.DID) if result.Repository != "" { msg += "/" + result.Repository } @@ -464,35 +611,27 @@ func (s *Server) handleTakedownSubmit(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleReverse(w http.ResponseWriter, r *http.Request) { - did := strings.TrimSpace(r.FormValue("did")) - repo := strings.TrimSpace(r.FormValue("repo")) - - if did == "" { - http.Redirect(w, r, "/?error=DID+is+required", http.StatusFound) + idStr := strings.TrimSpace(r.FormValue("takedown_id")) + if idStr == "" { + http.Redirect(w, r, "/?error=Missing+takedown_id", http.StatusFound) + return + } + id, err := strconv.ParseInt(idStr, 10, 64) + if err != nil || id <= 0 { + http.Redirect(w, r, "/?error=Invalid+takedown_id", http.StatusFound) return } - src := s.did - var ( - negs []Label - err error - ) - if repo == "" { - negs, err = NegateUserLabels(s.db, s.signingKey, src, did) - } else { - negs, err = NegateRepoLabels(s.db, s.signingKey, src, did, repo) + reversedBy := "" + if session := SessionFromContext(r.Context()); session != nil { + reversedBy = session.DID } - if err != nil { - slog.Error("Failed to reverse takedown", "did", did, "repo", repo, "error", err) - http.Redirect(w, r, "/?error=Failed+to+reverse+takedown", http.StatusFound) + if _, err := s.ReverseTakedown(r.Context(), id, reversedBy); err != nil { + slog.Error("Failed to reverse takedown", "takedown_id", id, "error", err) + http.Redirect(w, r, "/?error="+strings.ReplaceAll(err.Error(), " ", "+"), http.StatusFound) return } - for i := range negs { - s.hub.Broadcast(&negs[i]) - } - - slog.Info("Reversed takedown", "did", did, "repo", repo, "negations", len(negs)) http.Redirect(w, r, "/", http.StatusFound) }