diff --git a/deploy/README.md b/deploy/README.md index 19247d0..c2b2c76 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -243,6 +243,26 @@ docker pull atcr.io/yourhandle/test:latest docker logs -f atcr-appview ``` +#### Enable debug logging + +Toggle debug logging at runtime without restarting the container: + +```bash +# Enable debug logging (auto-reverts after 30 minutes) +docker kill -s SIGUSR1 atcr-appview +docker kill -s SIGUSR1 atcr-hold + +# Manually disable before timeout +docker kill -s SIGUSR1 atcr-appview +``` + +When toggled, you'll see: +``` +level=INFO msg="Log level changed" from=INFO to=DEBUG trigger=SIGUSR1 auto_revert_in=30m0s +``` + +**Note:** Despite the command name, `docker kill -s SIGUSR1` does NOT stop the container. It sends a user-defined signal that the application handles to toggle debug mode. + #### Restart services ```bash diff --git a/lexicons/io/atcr/hold/stats.json b/lexicons/io/atcr/hold/stats.json new file mode 100644 index 0000000..6e9a3f7 --- /dev/null +++ b/lexicons/io/atcr/hold/stats.json @@ -0,0 +1,52 @@ +{ + "lexicon": 1, + "id": "io.atcr.hold.stats", + "defs": { + "main": { + "type": "record", + "key": "any", + "description": "Repository statistics stored in the hold's embedded PDS. Tracks pull/push counts per owner+repository combination. Record key is deterministic: base32(sha256(ownerDID + \"/\" + repository)[:16]).", + "record": { + "type": "object", + "required": ["ownerDid", "repository", "pullCount", "pushCount", "updatedAt"], + "properties": { + "ownerDid": { + "type": "string", + "format": "did", + "description": "DID of the image owner (e.g., did:plc:xyz123)" + }, + "repository": { + "type": "string", + "description": "Repository name (e.g., myapp)", + "maxLength": 256 + }, + "pullCount": { + "type": "integer", + "minimum": 0, + "description": "Number of manifest downloads" + }, + "pushCount": { + "type": "integer", + "minimum": 0, + "description": "Number of manifest uploads" + }, + "lastPull": { + "type": "string", + "format": "datetime", + "description": "RFC3339 timestamp of last pull" + }, + "lastPush": { + "type": "string", + "format": "datetime", + "description": "RFC3339 timestamp of last push" + }, + "updatedAt": { + "type": "string", + "format": "datetime", + "description": "RFC3339 timestamp of when this record was last updated" + } + } + } + } + } +} diff --git a/pkg/atproto/lexicon.go b/pkg/atproto/lexicon.go index 5bc6946..c3cfd2b 100644 --- a/pkg/atproto/lexicon.go +++ b/pkg/atproto/lexicon.go @@ -665,6 +665,16 @@ func StatsRecordKey(ownerDID, repository string) string { return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(hash[:16])) } +// CrewRecordKey generates a deterministic rkey from member DID +// Uses same pattern as StatsRecordKey for consistency +// This enables O(1) crew membership lookups via getRecord instead of O(n) pagination +func CrewRecordKey(memberDID string) string { + hash := sha256.Sum256([]byte(memberDID)) + // Use first 16 bytes (128 bits) for collision resistance + // Encode with base32 (alphanumeric, lowercase, no padding) for ATProto rkey compatibility + return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(hash[:16])) +} + // TangledProfileRecord represents a Tangled profile for the hold // Collection: sh.tangled.actor.profile (singleton record at rkey "self") // Stored in the hold's embedded PDS diff --git a/pkg/auth/hold_remote.go b/pkg/auth/hold_remote.go index 93c3276..96f6b1d 100644 --- a/pkg/auth/hold_remote.go +++ b/pkg/auth/hold_remote.go @@ -324,75 +324,63 @@ func (a *RemoteHoldAuthorizer) IsCrewMember(ctx context.Context, holdDID, userDI } // isCrewMemberNoCache queries XRPC without caching (internal helper) -// Handles pagination to check all crew records, not just the first page +// Uses O(1) lookup via getRecord with hash-based rkey instead of pagination func (a *RemoteHoldAuthorizer) isCrewMemberNoCache(ctx context.Context, holdDID, userDID string) (bool, error) { // Resolve DID to URL holdURL := atproto.ResolveHoldURL(holdDID) - // Paginate through all crew records - cursor := "" - for { - // Build XRPC request URL with pagination - // GET /xrpc/com.atproto.repo.listRecords?repo={did}&collection=io.atcr.hold.crew&limit=100 - xrpcURL := fmt.Sprintf("%s%s?repo=%s&collection=%s&limit=100", - holdURL, atproto.RepoListRecords, url.QueryEscape(holdDID), url.QueryEscape(atproto.CrewCollection)) - if cursor != "" { - xrpcURL += "&cursor=" + url.QueryEscape(cursor) - } + // Generate deterministic rkey from member DID (hash-based) + rkey := atproto.CrewRecordKey(userDID) - req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil) - if err != nil { - return false, err - } + // Build XRPC request URL for direct record lookup + // GET /xrpc/com.atproto.repo.getRecord?repo={did}&collection=io.atcr.hold.crew&rkey={hash} + xrpcURL := fmt.Sprintf("%s%s?repo=%s&collection=%s&rkey=%s", + holdURL, atproto.RepoGetRecord, url.QueryEscape(holdDID), url.QueryEscape(atproto.CrewCollection), url.QueryEscape(rkey)) - resp, err := a.httpClient.Do(req) - if err != nil { - return false, fmt.Errorf("XRPC request failed: %w", err) - } - - if resp.StatusCode != http.StatusOK { - body, _ := io.ReadAll(resp.Body) - resp.Body.Close() - return false, fmt.Errorf("XRPC request failed: status %d: %s", resp.StatusCode, string(body)) - } - - // Parse response - var xrpcResp struct { - Cursor string `json:"cursor"` - Records []struct { - URI string `json:"uri"` - CID string `json:"cid"` - Value struct { - Type string `json:"$type"` - Member string `json:"member"` - Role string `json:"role"` - Permissions []string `json:"permissions"` - AddedAt string `json:"addedAt"` - } `json:"value"` - } `json:"records"` - } - - if err := json.NewDecoder(resp.Body).Decode(&xrpcResp); err != nil { - resp.Body.Close() - return false, fmt.Errorf("failed to decode XRPC response: %w", err) - } - resp.Body.Close() - - // Check if userDID is in this page of crew records - for _, record := range xrpcResp.Records { - if record.Value.Member == userDID { - // TODO: Check expiration if set - return true, nil - } - } - - // Check if there are more pages - if xrpcResp.Cursor == "" || len(xrpcResp.Records) == 0 { - break - } - cursor = xrpcResp.Cursor + req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil) + if err != nil { + return false, err } + resp, err := a.httpClient.Do(req) + if err != nil { + return false, fmt.Errorf("XRPC request failed: %w", err) + } + defer resp.Body.Close() + + // 404 means not a crew member (record doesn't exist) + if resp.StatusCode == http.StatusNotFound { + return false, nil + } + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return false, fmt.Errorf("XRPC request failed: status %d: %s", resp.StatusCode, string(body)) + } + + // Parse response to verify the member DID matches + var xrpcResp struct { + URI string `json:"uri"` + CID string `json:"cid"` + Value struct { + Type string `json:"$type"` + Member string `json:"member"` + Role string `json:"role"` + Permissions []string `json:"permissions"` + AddedAt string `json:"addedAt"` + } `json:"value"` + } + + if err := json.NewDecoder(resp.Body).Decode(&xrpcResp); err != nil { + return false, fmt.Errorf("failed to decode XRPC response: %w", err) + } + + // Verify the member DID matches (sanity check) + if xrpcResp.Value.Member == userDID { + return true, nil + } + + // Hash collision or invalid record - treat as not a member return false, nil } diff --git a/pkg/hold/admin/static/css/admin.css b/pkg/hold/admin/static/css/admin.css index 7e2f1c8..f046a4d 100644 --- a/pkg/hold/admin/static/css/admin.css +++ b/pkg/hold/admin/static/css/admin.css @@ -76,7 +76,7 @@ body { /* Container */ .container { - max-width: 1200px; + max-width: 1600px; margin: 0 auto; padding: 2rem; } @@ -302,6 +302,11 @@ body { color: var(--gray-500); } +.date-cell { + white-space: nowrap; + color: var(--gray-500); +} + /* Badges */ .badge { display: inline-block; diff --git a/pkg/hold/admin/templates/pages/crew.html b/pkg/hold/admin/templates/pages/crew.html index 4f49ed7..14a13a6 100644 --- a/pkg/hold/admin/templates/pages/crew.html +++ b/pkg/hold/admin/templates/pages/crew.html @@ -34,6 +34,7 @@ Permissions Tier Usage + Added Actions @@ -56,6 +57,7 @@ {{.UsagePercent}}% + {{formatTime .AddedAt}} diff --git a/pkg/hold/pds/crew.go b/pkg/hold/pds/crew.go index c240c9d..a5c33f4 100644 --- a/pkg/hold/pds/crew.go +++ b/pkg/hold/pds/crew.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "log/slog" "strings" "time" @@ -14,6 +15,8 @@ import ( ) // AddCrewMember adds a new crew member to the hold and commits to carstore +// Uses deterministic rkey based on member DID hash for O(1) lookups and automatic deduplication +// If the member already exists, updates their record (upsert behavior) func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, permissions []string) (cid.Cid, error) { crewRecord := &atproto.CrewRecord{ Type: atproto.CrewCollection, @@ -23,10 +26,12 @@ func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, per AddedAt: time.Now().Format(time.RFC3339), } - // Use repomgr for crew operations - auto-generated rkey is fine - _, recordCID, err := p.repomgr.CreateRecord(ctx, p.uid, atproto.CrewCollection, crewRecord) + // Use deterministic rkey based on member DID hash + // UpsertRecord handles create-or-update automatically + rkey := atproto.CrewRecordKey(memberDID) + _, recordCID, _, err := p.repomgr.UpsertRecord(ctx, p.uid, atproto.CrewCollection, rkey, crewRecord) if err != nil { - return cid.Undef, fmt.Errorf("failed to create crew record: %w", err) + return cid.Undef, fmt.Errorf("failed to upsert crew record: %w", err) } return recordCID, nil @@ -49,6 +54,13 @@ func (p *HoldPDS) GetCrewMember(ctx context.Context, rkey string) (cid.Cid, *atp return recordCID, crewRecord, nil } +// GetCrewMemberByDID retrieves a crew member by their DID using O(1) lookup +// Uses deterministic rkey based on member DID hash +func (p *HoldPDS) GetCrewMemberByDID(ctx context.Context, memberDID string) (cid.Cid, *atproto.CrewRecord, error) { + rkey := atproto.CrewRecordKey(memberDID) + return p.GetCrewMember(ctx, rkey) +} + // CrewMemberWithKey pairs a crew record with its rkey and CID type CrewMemberWithKey struct { Rkey string @@ -138,7 +150,7 @@ func (p *HoldPDS) ListCrewMembers(ctx context.Context) ([]*CrewMemberWithKey, er return crew, nil } -// RemoveCrewMember removes a crew member +// RemoveCrewMember removes a crew member by rkey func (p *HoldPDS) RemoveCrewMember(ctx context.Context, rkey string) error { // Use repomgr.DeleteRecord - it will automatically commit! // This fixes the bug where deletions weren't being committed @@ -150,53 +162,133 @@ func (p *HoldPDS) RemoveCrewMember(ctx context.Context, rkey string) error { return nil } +// RemoveCrewMemberByDID removes a crew member by their DID using O(1) lookup +func (p *HoldPDS) RemoveCrewMemberByDID(ctx context.Context, memberDID string) error { + rkey := atproto.CrewRecordKey(memberDID) + return p.RemoveCrewMember(ctx, rkey) +} + // UpdateCrewMemberTier updates a crew member's tier -// Since ATProto records are immutable, this finds the member's record by DID, -// deletes it, and recreates it with the new tier value. +// Uses O(1) lookup via hash-based rkey and PutRecord for atomic upsert func (p *HoldPDS) UpdateCrewMemberTier(ctx context.Context, memberDID, tier string) error { - // Find the crew member's record by iterating over crew records - members, err := p.ListCrewMembers(ctx) + // O(1) lookup using hash-based rkey + _, existing, err := p.GetCrewMemberByDID(ctx, memberDID) if err != nil { - return fmt.Errorf("failed to list crew members: %w", err) - } - - // Find the member with matching DID - var targetMember *CrewMemberWithKey - for _, m := range members { - if m.Record.Member == memberDID { - targetMember = m - break - } - } - - if targetMember == nil { - return fmt.Errorf("crew member not found: %s", memberDID) + return fmt.Errorf("crew member not found: %w", err) } // If tier is already the same, no update needed - if targetMember.Record.Tier == tier { + if existing.Tier == tier { return nil } - // Delete the old record - if err := p.RemoveCrewMember(ctx, targetMember.Rkey); err != nil { - return fmt.Errorf("failed to remove old crew record: %w", err) - } - - // Create new record with updated tier + // Create updated record (PutRecord handles upsert with same rkey) newRecord := &atproto.CrewRecord{ Type: atproto.CrewCollection, - Member: targetMember.Record.Member, - Role: targetMember.Record.Role, - Permissions: targetMember.Record.Permissions, + Member: existing.Member, + Role: existing.Role, + Permissions: existing.Permissions, Tier: tier, - AddedAt: targetMember.Record.AddedAt, // Preserve original add time + AddedAt: existing.AddedAt, // Preserve original add time } - _, _, err = p.repomgr.CreateRecord(ctx, p.uid, atproto.CrewCollection, newRecord) + rkey := atproto.CrewRecordKey(memberDID) + _, _, err = p.repomgr.PutRecord(ctx, p.uid, atproto.CrewCollection, rkey, newRecord) if err != nil { - return fmt.Errorf("failed to create updated crew record: %w", err) + return fmt.Errorf("failed to update crew record: %w", err) } return nil } + +// TODO(crew-migration): Remove this migration code after all holds have been upgraded (added 2026-01-06) +// This migrates TID-based crew records to hash-based rkeys for O(1) lookups + +// MigrateCrewRecordsToHashRkeys migrates old TID-based crew records to hash-based rkeys +// This is idempotent - records that already have hash-based rkeys are skipped +// Returns the number of records migrated +func (p *HoldPDS) MigrateCrewRecordsToHashRkeys(ctx context.Context) (int, error) { + // List all crew members (includes both TID and hash-based rkeys) + members, err := p.ListCrewMembers(ctx) + if err != nil { + return 0, fmt.Errorf("failed to list crew members: %w", err) + } + + slog.Info("Starting crew record migration", "totalRecords", len(members)) + + migrated := 0 + duplicatesDeleted := 0 + alreadyHashBased := 0 + seen := make(map[string]bool) // Track seen member DIDs to handle duplicates + + for _, m := range members { + memberDID := m.Record.Member + expectedRkey := atproto.CrewRecordKey(memberDID) + + // Skip if already using hash-based rkey + if m.Rkey == expectedRkey { + seen[memberDID] = true + alreadyHashBased++ + continue + } + + // This is a TID-based record that needs migration + slog.Info("Migrating crew record to hash-based rkey", + "memberDID", memberDID, + "oldRkey", m.Rkey, + "newRkey", expectedRkey) + + // Check if we already have a hash-based record for this DID (duplicate handling) + if seen[memberDID] { + // Already migrated this DID, just delete the old TID record + slog.Info("Deleting duplicate TID-based crew record", + "memberDID", memberDID, + "rkey", m.Rkey) + if err := p.RemoveCrewMember(ctx, m.Rkey); err != nil { + slog.Warn("Failed to delete duplicate crew record", + "rkey", m.Rkey, + "error", err) + } else { + duplicatesDeleted++ + } + continue + } + + // Create new record with hash-based rkey (PutRecord handles upsert) + newRecord := &atproto.CrewRecord{ + Type: atproto.CrewCollection, + Member: m.Record.Member, + Role: m.Record.Role, + Permissions: m.Record.Permissions, + Tier: m.Record.Tier, + AddedAt: m.Record.AddedAt, + } + + _, _, err := p.repomgr.PutRecord(ctx, p.uid, atproto.CrewCollection, expectedRkey, newRecord) + if err != nil { + slog.Error("Failed to create hash-based crew record", + "memberDID", memberDID, + "error", err) + continue + } + + // Delete the old TID-based record + if err := p.RemoveCrewMember(ctx, m.Rkey); err != nil { + slog.Warn("Failed to delete old TID-based crew record", + "rkey", m.Rkey, + "error", err) + // Continue anyway - the new record is created + } + + seen[memberDID] = true + migrated++ + } + + slog.Info("Crew record migration complete", + "migrated", migrated, + "duplicatesDeleted", duplicatesDeleted, + "alreadyHashBased", alreadyHashBased, + "totalRecords", len(members)) + + return migrated, nil +} diff --git a/pkg/hold/pds/records.go b/pkg/hold/pds/records.go index 5d1191c..5a55e0e 100644 --- a/pkg/hold/pds/records.go +++ b/pkg/hold/pds/records.go @@ -1,12 +1,14 @@ package pds import ( + "bytes" "context" "database/sql" "fmt" "log/slog" "strings" + "atcr.io/pkg/atproto" "github.com/bluesky-social/indigo/repo" "github.com/ipfs/go-cid" _ "github.com/mattn/go-sqlite3" @@ -24,6 +26,7 @@ type Record struct { Collection string Rkey string Cid string + Did string // Associated DID (member for crew, userDid for layers, ownerDid for stats) } const recordsSchema = ` @@ -31,18 +34,47 @@ CREATE TABLE IF NOT EXISTS records ( collection TEXT NOT NULL, rkey TEXT NOT NULL, cid TEXT NOT NULL, + did TEXT, PRIMARY KEY (collection, rkey) ); CREATE INDEX IF NOT EXISTS idx_records_collection_rkey ON records(collection, rkey); +CREATE INDEX IF NOT EXISTS idx_records_collection_did ON records(collection, did); ` +// Schema version for migration detection +const recordsSchemaVersion = 2 + // NewRecordsIndex creates or opens a records index +// If the schema is outdated (missing did column), drops and rebuilds the table func NewRecordsIndex(dbPath string) (*RecordsIndex, error) { db, err := sql.Open("sqlite3", dbPath) if err != nil { return nil, fmt.Errorf("failed to open records database: %w", err) } + // Check if table exists and has the did column + needsRebuild := false + var tableName string + err = db.QueryRow(`SELECT name FROM sqlite_master WHERE type='table' AND name='records'`).Scan(&tableName) + if err == nil { + // Table exists, check for did column + var colCount int + err = db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('records') WHERE name='did'`).Scan(&colCount) + if err != nil || colCount == 0 { + needsRebuild = true + slog.Info("Records index schema outdated, rebuilding with did column") + } + } + + if needsRebuild { + // Drop old table + _, err = db.Exec(`DROP TABLE IF EXISTS records`) + if err != nil { + db.Close() + return nil, fmt.Errorf("failed to drop old records table: %w", err) + } + } + // Create schema _, err = db.Exec(recordsSchema) if err != nil { @@ -62,11 +94,12 @@ func (ri *RecordsIndex) Close() error { } // IndexRecord adds or updates a record in the index -func (ri *RecordsIndex) IndexRecord(collection, rkey, cidStr string) error { +// did parameter is optional - pass empty string if not applicable +func (ri *RecordsIndex) IndexRecord(collection, rkey, cidStr, did string) error { _, err := ri.db.Exec(` - INSERT OR REPLACE INTO records (collection, rkey, cid) - VALUES (?, ?, ?) - `, collection, rkey, cidStr) + INSERT OR REPLACE INTO records (collection, rkey, cid, did) + VALUES (?, ?, ?, ?) + `, collection, rkey, cidStr, sql.NullString{String: did, Valid: did != ""}) return err } @@ -90,7 +123,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string, // Oldest first (ascending order) if cursor != "" { query = ` - SELECT collection, rkey, cid FROM records + SELECT collection, rkey, cid, COALESCE(did, '') FROM records WHERE collection = ? AND rkey > ? ORDER BY rkey ASC LIMIT ? @@ -98,7 +131,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string, args = []any{collection, cursor, limit + 1} } else { query = ` - SELECT collection, rkey, cid FROM records + SELECT collection, rkey, cid, COALESCE(did, '') FROM records WHERE collection = ? ORDER BY rkey ASC LIMIT ? @@ -109,7 +142,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string, // Newest first (descending order) - default if cursor != "" { query = ` - SELECT collection, rkey, cid FROM records + SELECT collection, rkey, cid, COALESCE(did, '') FROM records WHERE collection = ? AND rkey < ? ORDER BY rkey DESC LIMIT ? @@ -117,7 +150,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string, args = []any{collection, cursor, limit + 1} } else { query = ` - SELECT collection, rkey, cid FROM records + SELECT collection, rkey, cid, COALESCE(did, '') FROM records WHERE collection = ? ORDER BY rkey DESC LIMIT ? @@ -135,7 +168,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string, var records []Record for rows.Next() { var rec Record - if err := rows.Scan(&rec.Collection, &rec.Rkey, &rec.Cid); err != nil { + if err := rows.Scan(&rec.Collection, &rec.Rkey, &rec.Cid, &rec.Did); err != nil { return nil, "", fmt.Errorf("failed to scan record: %w", err) } records = append(records, rec) @@ -156,6 +189,58 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string, return records, nextCursor, nil } +// ListRecordsByDID returns records for a collection filtered by DID with pagination support +func (ri *RecordsIndex) ListRecordsByDID(collection, did string, limit int, cursor string) ([]Record, string, error) { + var query string + var args []any + + if cursor != "" { + query = ` + SELECT collection, rkey, cid, COALESCE(did, '') FROM records + WHERE collection = ? AND did = ? AND rkey < ? + ORDER BY rkey DESC + LIMIT ? + ` + args = []any{collection, did, cursor, limit + 1} + } else { + query = ` + SELECT collection, rkey, cid, COALESCE(did, '') FROM records + WHERE collection = ? AND did = ? + ORDER BY rkey DESC + LIMIT ? + ` + args = []any{collection, did, limit + 1} + } + + rows, err := ri.db.Query(query, args...) + if err != nil { + return nil, "", fmt.Errorf("failed to query records: %w", err) + } + defer rows.Close() + + var records []Record + for rows.Next() { + var rec Record + if err := rows.Scan(&rec.Collection, &rec.Rkey, &rec.Cid, &rec.Did); err != nil { + return nil, "", fmt.Errorf("failed to scan record: %w", err) + } + records = append(records, rec) + } + + if err := rows.Err(); err != nil { + return nil, "", fmt.Errorf("error iterating records: %w", err) + } + + // Determine next cursor + var nextCursor string + if len(records) > limit { + nextCursor = records[limit-1].Rkey + records = records[:limit] + } + + return records, nextCursor, nil +} + // Count returns the number of records in a collection func (ri *RecordsIndex) Count(collection string) (int, error) { var count int @@ -174,6 +259,7 @@ func (ri *RecordsIndex) TotalCount() (int, error) { // BackfillFromRepo populates the records index from an existing MST repo // Compares MST count with index count - only backfills if they differ +// Extracts DID from record content for crew, layer, and stats records func (ri *RecordsIndex) BackfillFromRepo(ctx context.Context, repoHandle *repo.Repo) error { // Count records in MST mstCount := 0 @@ -207,8 +293,8 @@ func (ri *RecordsIndex) BackfillFromRepo(ctx context.Context, repoHandle *repo.R defer tx.Rollback() stmt, err := tx.Prepare(` - INSERT OR REPLACE INTO records (collection, rkey, cid) - VALUES (?, ?, ?) + INSERT OR REPLACE INTO records (collection, rkey, cid, did) + VALUES (?, ?, ?, ?) `) if err != nil { return fmt.Errorf("failed to prepare statement: %w", err) @@ -224,7 +310,14 @@ func (ri *RecordsIndex) BackfillFromRepo(ctx context.Context, repoHandle *repo.R } collection, rkey := parts[0], parts[1] - _, err := stmt.Exec(collection, rkey, c.String()) + // Extract DID from record content based on collection type + var did string + _, recBytes, err := repoHandle.GetRecordBytes(ctx, key) + if err == nil && recBytes != nil { + did = extractDIDFromRecord(collection, *recBytes) + } + + _, err = stmt.Exec(collection, rkey, c.String(), sql.NullString{String: did, Valid: did != ""}) if err != nil { return fmt.Errorf("failed to index record %s: %w", key, err) } @@ -249,3 +342,25 @@ func (ri *RecordsIndex) BackfillFromRepo(ctx context.Context, repoHandle *repo.R slog.Info("Backfill complete", "records", recordCount) return nil } + +// extractDIDFromRecord extracts the associated DID from a record based on its collection type +func extractDIDFromRecord(collection string, recBytes []byte) string { + switch collection { + case atproto.CrewCollection: + var rec atproto.CrewRecord + if err := rec.UnmarshalCBOR(bytes.NewReader(recBytes)); err == nil { + return rec.Member + } + case atproto.LayerCollection: + var rec atproto.LayerRecord + if err := rec.UnmarshalCBOR(bytes.NewReader(recBytes)); err == nil { + return rec.UserDID + } + case atproto.StatsCollection: + var rec atproto.StatsRecord + if err := rec.UnmarshalCBOR(bytes.NewReader(recBytes)); err == nil { + return rec.OwnerDID + } + } + return "" +} diff --git a/pkg/hold/pds/records_test.go b/pkg/hold/pds/records_test.go index 619aabc..ede2e93 100644 --- a/pkg/hold/pds/records_test.go +++ b/pkg/hold/pds/records_test.go @@ -50,7 +50,7 @@ func TestRecordsIndex_IndexRecord(t *testing.T) { defer ri.Close() // Index a record - err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123") + err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123", "") if err != nil { t.Fatalf("IndexRecord() error = %v", err) } @@ -75,13 +75,13 @@ func TestRecordsIndex_IndexRecord_Upsert(t *testing.T) { defer ri.Close() // Index a record - err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123") + err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123", "") if err != nil { t.Fatalf("IndexRecord() first call error = %v", err) } // Update the same record with new CID - err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei456") + err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei456", "") if err != nil { t.Fatalf("IndexRecord() second call error = %v", err) } @@ -118,7 +118,7 @@ func TestRecordsIndex_DeleteRecord(t *testing.T) { defer ri.Close() // Index a record - err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123") + err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123", "") if err != nil { t.Fatalf("IndexRecord() error = %v", err) } @@ -217,7 +217,7 @@ func TestRecordsIndex_ListRecords_Basic(t *testing.T) { {"ccc", "cid3"}, } for _, r := range records { - if err := ri.IndexRecord("io.atcr.hold.crew", r.rkey, r.cid); err != nil { + if err := ri.IndexRecord("io.atcr.hold.crew", r.rkey, r.cid, ""); err != nil { t.Fatalf("IndexRecord() error = %v", err) } } @@ -248,7 +248,7 @@ func TestRecordsIndex_ListRecords_DefaultOrder(t *testing.T) { // Add records with different rkeys (TIDs are lexicographically ordered by time) rkeys := []string{"3m3aaaaaaaaa", "3m3bbbbbbbbb", "3m3ccccccccc"} for _, rkey := range rkeys { - if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey); err != nil { + if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey, ""); err != nil { t.Fatalf("IndexRecord() error = %v", err) } } @@ -286,7 +286,7 @@ func TestRecordsIndex_ListRecords_ReverseOrder(t *testing.T) { // Add records rkeys := []string{"3m3aaaaaaaaa", "3m3bbbbbbbbb", "3m3ccccccccc"} for _, rkey := range rkeys { - if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey); err != nil { + if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey, ""); err != nil { t.Fatalf("IndexRecord() error = %v", err) } } @@ -324,7 +324,7 @@ func TestRecordsIndex_ListRecords_Limit(t *testing.T) { // Add 5 records for i := range 5 { rkey := string(rune('a' + i)) - if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey); err != nil { + if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey, ""); err != nil { t.Fatalf("IndexRecord() error = %v", err) } } @@ -355,7 +355,7 @@ func TestRecordsIndex_ListRecords_Cursor(t *testing.T) { // Add 5 records rkeys := []string{"a", "b", "c", "d", "e"} for _, rkey := range rkeys { - if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey); err != nil { + if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey, ""); err != nil { t.Fatalf("IndexRecord() error = %v", err) } } @@ -430,7 +430,7 @@ func TestRecordsIndex_ListRecords_CursorReverse(t *testing.T) { // Add 5 records rkeys := []string{"a", "b", "c", "d", "e"} for _, rkey := range rkeys { - if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey); err != nil { + if err := ri.IndexRecord("io.atcr.hold.crew", rkey, "cid-"+rkey, ""); err != nil { t.Fatalf("IndexRecord() error = %v", err) } } @@ -474,10 +474,10 @@ func TestRecordsIndex_Count(t *testing.T) { // Add records to two collections for i := range 3 { - ri.IndexRecord("io.atcr.hold.crew", string(rune('a'+i)), "cid1") + ri.IndexRecord("io.atcr.hold.crew", string(rune('a'+i)), "cid1", "") } for i := range 5 { - ri.IndexRecord("io.atcr.hold.captain", string(rune('a'+i)), "cid2") + ri.IndexRecord("io.atcr.hold.captain", string(rune('a'+i)), "cid2", "") } // Count crew @@ -527,10 +527,10 @@ func TestRecordsIndex_TotalCount(t *testing.T) { defer ri.Close() // Add records to multiple collections - ri.IndexRecord("io.atcr.hold.crew", "a", "cid1") - ri.IndexRecord("io.atcr.hold.crew", "b", "cid2") - ri.IndexRecord("io.atcr.hold.captain", "self", "cid3") - ri.IndexRecord("io.atcr.manifest", "abc123", "cid4") + ri.IndexRecord("io.atcr.hold.crew", "a", "cid1", "") + ri.IndexRecord("io.atcr.hold.crew", "b", "cid2", "") + ri.IndexRecord("io.atcr.hold.captain", "self", "cid3", "") + ri.IndexRecord("io.atcr.manifest", "abc123", "cid4", "") count, err := ri.TotalCount() if err != nil { @@ -581,9 +581,9 @@ func TestRecordsIndex_MultipleCollections(t *testing.T) { defer ri.Close() // Add records to different collections with same rkeys - ri.IndexRecord("io.atcr.hold.crew", "abc", "cid-crew") - ri.IndexRecord("io.atcr.hold.captain", "abc", "cid-captain") - ri.IndexRecord("io.atcr.manifest", "abc", "cid-manifest") + ri.IndexRecord("io.atcr.hold.crew", "abc", "cid-crew", "") + ri.IndexRecord("io.atcr.hold.captain", "abc", "cid-captain", "") + ri.IndexRecord("io.atcr.manifest", "abc", "cid-manifest", "") // Listing should only return records from requested collection records, _, err := ri.ListRecords("io.atcr.hold.crew", 10, "", false) diff --git a/pkg/hold/pds/repomgr.go b/pkg/hold/pds/repomgr.go index 77e5a60..fd43b40 100644 --- a/pkg/hold/pds/repomgr.go +++ b/pkg/hold/pds/repomgr.go @@ -382,6 +382,94 @@ func (rm *RepoManager) PutRecord(ctx context.Context, user models.Uid, collectio return rpath, cc, nil } +// UpsertRecord creates or updates a record with an explicit rkey. +// If the record doesn't exist, it creates it. If it exists, it updates it. +// Returns the collection path (e.g., "io.atcr.captain/self"), CID, and whether it was created (true) or updated (false). +func (rm *RepoManager) UpsertRecord(ctx context.Context, user models.Uid, collection, rkey string, rec cbg.CBORMarshaler) (string, cid.Cid, bool, error) { + ctx, span := otel.Tracer("repoman").Start(ctx, "UpsertRecord") + defer span.End() + + unlock := rm.lockUser(ctx, user) + defer unlock() + + rev, err := rm.cs.GetUserRepoRev(ctx, user) + if err != nil { + return "", cid.Undef, false, err + } + + ds, err := rm.cs.NewDeltaSession(ctx, user, &rev) + if err != nil { + return "", cid.Undef, false, err + } + + head := ds.BaseCid() + r, err := repo.OpenRepo(ctx, ds, head) + if err != nil { + return "", cid.Undef, false, err + } + + rpath := collection + "/" + rkey + + // Check if record exists + _, _, err = r.GetRecordBytes(ctx, rpath) + recordExists := err == nil + + var cc cid.Cid + var evtKind EventKind + if recordExists { + // Update existing record + cc, err = r.UpdateRecord(ctx, rpath, rec) + evtKind = EvtKindUpdateRecord + } else { + // Create new record + cc, err = r.PutRecord(ctx, rpath, rec) + evtKind = EvtKindCreateRecord + } + if err != nil { + return "", cid.Undef, false, err + } + + nroot, nrev, err := r.Commit(ctx, rm.kmgr.SignForUser) + if err != nil { + return "", cid.Undef, false, err + } + + rslice, err := ds.CloseWithRoot(ctx, nroot, nrev) + if err != nil { + return "", cid.Undef, false, fmt.Errorf("close with root: %w", err) + } + + var oldroot *cid.Cid + if head.Defined() { + oldroot = &head + } + + if rm.events != nil { + op := RepoOp{ + Kind: evtKind, + Collection: collection, + Rkey: rkey, + RecCid: &cc, + } + + if rm.hydrateRecords { + op.Record = rec + } + + rm.events(ctx, &RepoEvent{ + User: user, + OldRoot: oldroot, + NewRoot: nroot, + Rev: nrev, + Since: &rev, + Ops: []RepoOp{op}, + RepoSlice: rslice, + }) + } + + return rpath, cc, !recordExists, nil +} + func (rm *RepoManager) DeleteRecord(ctx context.Context, user models.Uid, collection, rkey string) error { ctx, span := otel.Tracer("repoman").Start(ctx, "DeleteRecord") defer span.End() diff --git a/pkg/hold/pds/server.go b/pkg/hold/pds/server.go index 19f54e8..5f6e6e1 100644 --- a/pkg/hold/pds/server.go +++ b/pkg/hold/pds/server.go @@ -225,6 +225,14 @@ func (p *HoldPDS) Bootstrap(ctx context.Context, storageDriver driver.StorageDri } } + // TODO(crew-migration): Remove this call after all holds have been upgraded (added 2026-01-06) + // Migrate TID-based crew records to hash-based rkeys for O(1) lookups + if migrated, err := p.MigrateCrewRecordsToHashRkeys(ctx); err != nil { + slog.Warn("Crew record migration failed", "error", err) + } else if migrated > 0 { + slog.Info("Migrated crew records to hash-based rkeys", "count", migrated) + } + // Create Bluesky profile record (idempotent - check if exists first) // This runs even if captain exists (for existing holds being upgraded) // Skip if no storage driver (e.g., in tests) @@ -319,7 +327,9 @@ func (p *HoldPDS) CreateRecordsIndexEventHandler(broadcasterHandler func(context if op.RecCid != nil { cidStr = op.RecCid.String() } - if err := p.recordsIndex.IndexRecord(op.Collection, op.Rkey, cidStr); err != nil { + // Extract DID from record based on collection type + did := extractDIDFromOp(op) + if err := p.recordsIndex.IndexRecord(op.Collection, op.Rkey, cidStr, did); err != nil { slog.Warn("Failed to index record", "collection", op.Collection, "rkey", op.Rkey, "error", err) } case EvtKindDeleteRecord: @@ -338,6 +348,28 @@ func (p *HoldPDS) CreateRecordsIndexEventHandler(broadcasterHandler func(context } } +// extractDIDFromOp extracts the associated DID from a repo operation based on collection type +func extractDIDFromOp(op RepoOp) string { + if op.Record == nil { + return "" + } + switch op.Collection { + case atproto.CrewCollection: + if rec, ok := op.Record.(*atproto.CrewRecord); ok { + return rec.Member + } + case atproto.LayerCollection: + if rec, ok := op.Record.(*atproto.LayerRecord); ok { + return rec.UserDID + } + case atproto.StatsCollection: + if rec, ok := op.Record.(*atproto.StatsRecord); ok { + return rec.OwnerDID + } + } + return "" +} + // BackfillRecordsIndex populates the records index from existing MST data func (p *HoldPDS) BackfillRecordsIndex(ctx context.Context) error { if p.recordsIndex == nil { diff --git a/pkg/hold/pds/server_test.go b/pkg/hold/pds/server_test.go index 8d0f556..6fd6851 100644 --- a/pkg/hold/pds/server_test.go +++ b/pkg/hold/pds/server_test.go @@ -524,16 +524,15 @@ func TestBootstrap_CrewWithoutCaptain(t *testing.T) { } // Verify crew wasn't duplicated (Bootstrap adds owner as crew, but they already exist) + // With hash-based rkeys, AddCrewMember uses PutRecord which upserts - no duplicates possible crewAfter, err := pds.ListCrewMembers(ctx) if err != nil { t.Fatalf("ListCrewMembers failed after bootstrap: %v", err) } - // Should have 2 crew members now: original + one added by bootstrap - // (Bootstrap doesn't check for duplicates currently) - if len(crewAfter) != 2 { - t.Logf("Note: Bootstrap added owner as crew even though they already existed") - t.Logf("Crew count after bootstrap: %d", len(crewAfter)) + // Should still have 1 crew member (hash-based rkey ensures upsert, not duplicate) + if len(crewAfter) != 1 { + t.Errorf("Expected 1 crew member after bootstrap (upsert), got %d", len(crewAfter)) } } diff --git a/pkg/logging/logger.go b/pkg/logging/logger.go index 389e5a9..38571c9 100644 --- a/pkg/logging/logger.go +++ b/pkg/logging/logger.go @@ -1,18 +1,39 @@ // Package logging provides centralized structured logging using slog // with configurable log levels. Call InitLogger() from main() to configure. +// +// Dynamic debug logging: +// Send SIGUSR1 to toggle debug mode at runtime (auto-reverts after 30 minutes). +// Example: docker kill -s SIGUSR1 package logging import ( "io" "log/slog" "os" + "os/signal" "strings" + "sync" + "sync/atomic" + "syscall" + "time" +) + +const debugTimeout = 30 * time.Minute + +var ( + levelVar *slog.LevelVar + originalLevel slog.Level + debugEnabled atomic.Bool + revertTimer *time.Timer + revertMu sync.Mutex ) // InitLogger initializes the global slog default logger with the specified log level. // Valid levels: debug, info, warn, error (case-insensitive) // If level is empty or invalid, defaults to INFO. // Call this from main() at startup. +// +// Also starts a signal handler for SIGUSR1 to toggle debug mode at runtime. func InitLogger(level string) { var logLevel slog.Level @@ -29,12 +50,96 @@ func InitLogger(level string) { logLevel = slog.LevelInfo } + // Store original level for toggle-back and use LevelVar for dynamic changes + originalLevel = logLevel + levelVar = new(slog.LevelVar) + levelVar.Set(logLevel) + opts := &slog.HandlerOptions{ - Level: logLevel, + Level: levelVar, } handler := slog.NewTextHandler(os.Stdout, opts) slog.SetDefault(slog.New(handler)) + + // Start signal handler for dynamic debug toggle + go handleDebugSignal() +} + +func handleDebugSignal() { + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGUSR1) + + for range sigChan { + ToggleDebug() + } +} + +// ToggleDebug toggles between the original log level and DEBUG. +// When enabling debug, starts a 30-minute timer that auto-reverts. +// Typically called via SIGUSR1 signal. +func ToggleDebug() { + revertMu.Lock() + defer revertMu.Unlock() + + wasDebug := debugEnabled.Swap(!debugEnabled.Load()) + + // Cancel any existing revert timer + if revertTimer != nil { + revertTimer.Stop() + revertTimer = nil + } + + if wasDebug { + // Turning debug OFF + levelVar.Set(originalLevel) + slog.Info("Log level changed", + "from", "DEBUG", + "to", levelToString(originalLevel), + "trigger", "SIGUSR1") + } else { + // Turning debug ON - start auto-revert timer + levelVar.Set(slog.LevelDebug) + revertTimer = time.AfterFunc(debugTimeout, autoRevert) + slog.Info("Log level changed", + "from", levelToString(originalLevel), + "to", "DEBUG", + "trigger", "SIGUSR1", + "auto_revert_in", debugTimeout) + } +} + +func autoRevert() { + revertMu.Lock() + defer revertMu.Unlock() + + if !debugEnabled.Load() { + return // Already reverted manually + } + + debugEnabled.Store(false) + levelVar.Set(originalLevel) + revertTimer = nil + + slog.Info("Log level changed", + "from", "DEBUG", + "to", levelToString(originalLevel), + "trigger", "auto-revert") +} + +func levelToString(l slog.Level) string { + switch l { + case slog.LevelDebug: + return "DEBUG" + case slog.LevelInfo: + return "INFO" + case slog.LevelWarn: + return "WARN" + case slog.LevelError: + return "ERROR" + default: + return l.String() + } } // SetupTestLogger configures logging for tests to reduce noise.