mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-09-20 09:14:16 +00:00
add ability to toggle debug. refactor hold pds logic to allow crew record lookups by rkey rather than a list
This commit is contained in:
@@ -243,6 +243,26 @@ docker pull atcr.io/yourhandle/test:latest
|
|||||||
docker logs -f atcr-appview
|
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
|
#### Restart services
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|||||||
@@ -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"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -665,6 +665,16 @@ func StatsRecordKey(ownerDID, repository string) string {
|
|||||||
return strings.ToLower(base32.StdEncoding.WithPadding(base32.NoPadding).EncodeToString(hash[:16]))
|
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
|
// TangledProfileRecord represents a Tangled profile for the hold
|
||||||
// Collection: sh.tangled.actor.profile (singleton record at rkey "self")
|
// Collection: sh.tangled.actor.profile (singleton record at rkey "self")
|
||||||
// Stored in the hold's embedded PDS
|
// Stored in the hold's embedded PDS
|
||||||
|
|||||||
+18
-30
@@ -324,21 +324,18 @@ func (a *RemoteHoldAuthorizer) IsCrewMember(ctx context.Context, holdDID, userDI
|
|||||||
}
|
}
|
||||||
|
|
||||||
// isCrewMemberNoCache queries XRPC without caching (internal helper)
|
// 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) {
|
func (a *RemoteHoldAuthorizer) isCrewMemberNoCache(ctx context.Context, holdDID, userDID string) (bool, error) {
|
||||||
// Resolve DID to URL
|
// Resolve DID to URL
|
||||||
holdURL := atproto.ResolveHoldURL(holdDID)
|
holdURL := atproto.ResolveHoldURL(holdDID)
|
||||||
|
|
||||||
// Paginate through all crew records
|
// Generate deterministic rkey from member DID (hash-based)
|
||||||
cursor := ""
|
rkey := atproto.CrewRecordKey(userDID)
|
||||||
for {
|
|
||||||
// Build XRPC request URL with pagination
|
// Build XRPC request URL for direct record lookup
|
||||||
// GET /xrpc/com.atproto.repo.listRecords?repo={did}&collection=io.atcr.hold.crew&limit=100
|
// GET /xrpc/com.atproto.repo.getRecord?repo={did}&collection=io.atcr.hold.crew&rkey={hash}
|
||||||
xrpcURL := fmt.Sprintf("%s%s?repo=%s&collection=%s&limit=100",
|
xrpcURL := fmt.Sprintf("%s%s?repo=%s&collection=%s&rkey=%s",
|
||||||
holdURL, atproto.RepoListRecords, url.QueryEscape(holdDID), url.QueryEscape(atproto.CrewCollection))
|
holdURL, atproto.RepoGetRecord, url.QueryEscape(holdDID), url.QueryEscape(atproto.CrewCollection), url.QueryEscape(rkey))
|
||||||
if cursor != "" {
|
|
||||||
xrpcURL += "&cursor=" + url.QueryEscape(cursor)
|
|
||||||
}
|
|
||||||
|
|
||||||
req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil)
|
req, err := http.NewRequestWithContext(ctx, "GET", xrpcURL, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -349,17 +346,20 @@ func (a *RemoteHoldAuthorizer) isCrewMemberNoCache(ctx context.Context, holdDID,
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return false, fmt.Errorf("XRPC request failed: %w", err)
|
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 {
|
if resp.StatusCode != http.StatusOK {
|
||||||
body, _ := io.ReadAll(resp.Body)
|
body, _ := io.ReadAll(resp.Body)
|
||||||
resp.Body.Close()
|
|
||||||
return false, fmt.Errorf("XRPC request failed: status %d: %s", resp.StatusCode, string(body))
|
return false, fmt.Errorf("XRPC request failed: status %d: %s", resp.StatusCode, string(body))
|
||||||
}
|
}
|
||||||
|
|
||||||
// Parse response
|
// Parse response to verify the member DID matches
|
||||||
var xrpcResp struct {
|
var xrpcResp struct {
|
||||||
Cursor string `json:"cursor"`
|
|
||||||
Records []struct {
|
|
||||||
URI string `json:"uri"`
|
URI string `json:"uri"`
|
||||||
CID string `json:"cid"`
|
CID string `json:"cid"`
|
||||||
Value struct {
|
Value struct {
|
||||||
@@ -369,30 +369,18 @@ func (a *RemoteHoldAuthorizer) isCrewMemberNoCache(ctx context.Context, holdDID,
|
|||||||
Permissions []string `json:"permissions"`
|
Permissions []string `json:"permissions"`
|
||||||
AddedAt string `json:"addedAt"`
|
AddedAt string `json:"addedAt"`
|
||||||
} `json:"value"`
|
} `json:"value"`
|
||||||
} `json:"records"`
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := json.NewDecoder(resp.Body).Decode(&xrpcResp); err != nil {
|
if err := json.NewDecoder(resp.Body).Decode(&xrpcResp); err != nil {
|
||||||
resp.Body.Close()
|
|
||||||
return false, fmt.Errorf("failed to decode XRPC response: %w", err)
|
return false, fmt.Errorf("failed to decode XRPC response: %w", err)
|
||||||
}
|
}
|
||||||
resp.Body.Close()
|
|
||||||
|
|
||||||
// Check if userDID is in this page of crew records
|
// Verify the member DID matches (sanity check)
|
||||||
for _, record := range xrpcResp.Records {
|
if xrpcResp.Value.Member == userDID {
|
||||||
if record.Value.Member == userDID {
|
|
||||||
// TODO: Check expiration if set
|
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
// Check if there are more pages
|
|
||||||
if xrpcResp.Cursor == "" || len(xrpcResp.Records) == 0 {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
cursor = xrpcResp.Cursor
|
|
||||||
}
|
|
||||||
|
|
||||||
|
// Hash collision or invalid record - treat as not a member
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -76,7 +76,7 @@ body {
|
|||||||
|
|
||||||
/* Container */
|
/* Container */
|
||||||
.container {
|
.container {
|
||||||
max-width: 1200px;
|
max-width: 1600px;
|
||||||
margin: 0 auto;
|
margin: 0 auto;
|
||||||
padding: 2rem;
|
padding: 2rem;
|
||||||
}
|
}
|
||||||
@@ -302,6 +302,11 @@ body {
|
|||||||
color: var(--gray-500);
|
color: var(--gray-500);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
.date-cell {
|
||||||
|
white-space: nowrap;
|
||||||
|
color: var(--gray-500);
|
||||||
|
}
|
||||||
|
|
||||||
/* Badges */
|
/* Badges */
|
||||||
.badge {
|
.badge {
|
||||||
display: inline-block;
|
display: inline-block;
|
||||||
|
|||||||
@@ -34,6 +34,7 @@
|
|||||||
<th>Permissions</th>
|
<th>Permissions</th>
|
||||||
<th>Tier</th>
|
<th>Tier</th>
|
||||||
<th>Usage</th>
|
<th>Usage</th>
|
||||||
|
<th>Added</th>
|
||||||
<th class="actions-header">Actions</th>
|
<th class="actions-header">Actions</th>
|
||||||
</tr>
|
</tr>
|
||||||
</thead>
|
</thead>
|
||||||
@@ -56,6 +57,7 @@
|
|||||||
<small>{{.UsagePercent}}%</small>
|
<small>{{.UsagePercent}}%</small>
|
||||||
</div>
|
</div>
|
||||||
</td>
|
</td>
|
||||||
|
<td class="date-cell">{{formatTime .AddedAt}}</td>
|
||||||
<td class="actions">
|
<td class="actions">
|
||||||
<a href="/admin/crew/{{.RKey}}" class="btn btn-icon" title="Edit">
|
<a href="/admin/crew/{{.RKey}}" class="btn btn-icon" title="Edit">
|
||||||
<i data-lucide="pencil"></i>
|
<i data-lucide="pencil"></i>
|
||||||
|
|||||||
+127
-35
@@ -5,6 +5,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -14,6 +15,8 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// AddCrewMember adds a new crew member to the hold and commits to carstore
|
// 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) {
|
func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, permissions []string) (cid.Cid, error) {
|
||||||
crewRecord := &atproto.CrewRecord{
|
crewRecord := &atproto.CrewRecord{
|
||||||
Type: atproto.CrewCollection,
|
Type: atproto.CrewCollection,
|
||||||
@@ -23,10 +26,12 @@ func (p *HoldPDS) AddCrewMember(ctx context.Context, memberDID, role string, per
|
|||||||
AddedAt: time.Now().Format(time.RFC3339),
|
AddedAt: time.Now().Format(time.RFC3339),
|
||||||
}
|
}
|
||||||
|
|
||||||
// Use repomgr for crew operations - auto-generated rkey is fine
|
// Use deterministic rkey based on member DID hash
|
||||||
_, recordCID, err := p.repomgr.CreateRecord(ctx, p.uid, atproto.CrewCollection, crewRecord)
|
// 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 {
|
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
|
return recordCID, nil
|
||||||
@@ -49,6 +54,13 @@ func (p *HoldPDS) GetCrewMember(ctx context.Context, rkey string) (cid.Cid, *atp
|
|||||||
return recordCID, crewRecord, nil
|
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
|
// CrewMemberWithKey pairs a crew record with its rkey and CID
|
||||||
type CrewMemberWithKey struct {
|
type CrewMemberWithKey struct {
|
||||||
Rkey string
|
Rkey string
|
||||||
@@ -138,7 +150,7 @@ func (p *HoldPDS) ListCrewMembers(ctx context.Context) ([]*CrewMemberWithKey, er
|
|||||||
return crew, nil
|
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 {
|
func (p *HoldPDS) RemoveCrewMember(ctx context.Context, rkey string) error {
|
||||||
// Use repomgr.DeleteRecord - it will automatically commit!
|
// Use repomgr.DeleteRecord - it will automatically commit!
|
||||||
// This fixes the bug where deletions weren't being committed
|
// 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
|
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
|
// UpdateCrewMemberTier updates a crew member's tier
|
||||||
// Since ATProto records are immutable, this finds the member's record by DID,
|
// Uses O(1) lookup via hash-based rkey and PutRecord for atomic upsert
|
||||||
// deletes it, and recreates it with the new tier value.
|
|
||||||
func (p *HoldPDS) UpdateCrewMemberTier(ctx context.Context, memberDID, tier string) error {
|
func (p *HoldPDS) UpdateCrewMemberTier(ctx context.Context, memberDID, tier string) error {
|
||||||
// Find the crew member's record by iterating over crew records
|
// O(1) lookup using hash-based rkey
|
||||||
members, err := p.ListCrewMembers(ctx)
|
_, existing, err := p.GetCrewMemberByDID(ctx, memberDID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to list crew members: %w", err)
|
return fmt.Errorf("crew member not found: %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)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// If tier is already the same, no update needed
|
// If tier is already the same, no update needed
|
||||||
if targetMember.Record.Tier == tier {
|
if existing.Tier == tier {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Delete the old record
|
// Create updated record (PutRecord handles upsert with same rkey)
|
||||||
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
|
|
||||||
newRecord := &atproto.CrewRecord{
|
newRecord := &atproto.CrewRecord{
|
||||||
Type: atproto.CrewCollection,
|
Type: atproto.CrewCollection,
|
||||||
Member: targetMember.Record.Member,
|
Member: existing.Member,
|
||||||
Role: targetMember.Record.Role,
|
Role: existing.Role,
|
||||||
Permissions: targetMember.Record.Permissions,
|
Permissions: existing.Permissions,
|
||||||
Tier: tier,
|
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 {
|
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
|
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
|
||||||
|
}
|
||||||
|
|||||||
+127
-12
@@ -1,12 +1,14 @@
|
|||||||
package pds
|
package pds
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
|
"atcr.io/pkg/atproto"
|
||||||
"github.com/bluesky-social/indigo/repo"
|
"github.com/bluesky-social/indigo/repo"
|
||||||
"github.com/ipfs/go-cid"
|
"github.com/ipfs/go-cid"
|
||||||
_ "github.com/mattn/go-sqlite3"
|
_ "github.com/mattn/go-sqlite3"
|
||||||
@@ -24,6 +26,7 @@ type Record struct {
|
|||||||
Collection string
|
Collection string
|
||||||
Rkey string
|
Rkey string
|
||||||
Cid string
|
Cid string
|
||||||
|
Did string // Associated DID (member for crew, userDid for layers, ownerDid for stats)
|
||||||
}
|
}
|
||||||
|
|
||||||
const recordsSchema = `
|
const recordsSchema = `
|
||||||
@@ -31,18 +34,47 @@ CREATE TABLE IF NOT EXISTS records (
|
|||||||
collection TEXT NOT NULL,
|
collection TEXT NOT NULL,
|
||||||
rkey TEXT NOT NULL,
|
rkey TEXT NOT NULL,
|
||||||
cid TEXT NOT NULL,
|
cid TEXT NOT NULL,
|
||||||
|
did TEXT,
|
||||||
PRIMARY KEY (collection, rkey)
|
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_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
|
// 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) {
|
func NewRecordsIndex(dbPath string) (*RecordsIndex, error) {
|
||||||
db, err := sql.Open("sqlite3", dbPath)
|
db, err := sql.Open("sqlite3", dbPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to open records database: %w", err)
|
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
|
// Create schema
|
||||||
_, err = db.Exec(recordsSchema)
|
_, err = db.Exec(recordsSchema)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -62,11 +94,12 @@ func (ri *RecordsIndex) Close() error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// IndexRecord adds or updates a record in the index
|
// 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(`
|
_, err := ri.db.Exec(`
|
||||||
INSERT OR REPLACE INTO records (collection, rkey, cid)
|
INSERT OR REPLACE INTO records (collection, rkey, cid, did)
|
||||||
VALUES (?, ?, ?)
|
VALUES (?, ?, ?, ?)
|
||||||
`, collection, rkey, cidStr)
|
`, collection, rkey, cidStr, sql.NullString{String: did, Valid: did != ""})
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -90,7 +123,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string,
|
|||||||
// Oldest first (ascending order)
|
// Oldest first (ascending order)
|
||||||
if cursor != "" {
|
if cursor != "" {
|
||||||
query = `
|
query = `
|
||||||
SELECT collection, rkey, cid FROM records
|
SELECT collection, rkey, cid, COALESCE(did, '') FROM records
|
||||||
WHERE collection = ? AND rkey > ?
|
WHERE collection = ? AND rkey > ?
|
||||||
ORDER BY rkey ASC
|
ORDER BY rkey ASC
|
||||||
LIMIT ?
|
LIMIT ?
|
||||||
@@ -98,7 +131,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string,
|
|||||||
args = []any{collection, cursor, limit + 1}
|
args = []any{collection, cursor, limit + 1}
|
||||||
} else {
|
} else {
|
||||||
query = `
|
query = `
|
||||||
SELECT collection, rkey, cid FROM records
|
SELECT collection, rkey, cid, COALESCE(did, '') FROM records
|
||||||
WHERE collection = ?
|
WHERE collection = ?
|
||||||
ORDER BY rkey ASC
|
ORDER BY rkey ASC
|
||||||
LIMIT ?
|
LIMIT ?
|
||||||
@@ -109,7 +142,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string,
|
|||||||
// Newest first (descending order) - default
|
// Newest first (descending order) - default
|
||||||
if cursor != "" {
|
if cursor != "" {
|
||||||
query = `
|
query = `
|
||||||
SELECT collection, rkey, cid FROM records
|
SELECT collection, rkey, cid, COALESCE(did, '') FROM records
|
||||||
WHERE collection = ? AND rkey < ?
|
WHERE collection = ? AND rkey < ?
|
||||||
ORDER BY rkey DESC
|
ORDER BY rkey DESC
|
||||||
LIMIT ?
|
LIMIT ?
|
||||||
@@ -117,7 +150,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string,
|
|||||||
args = []any{collection, cursor, limit + 1}
|
args = []any{collection, cursor, limit + 1}
|
||||||
} else {
|
} else {
|
||||||
query = `
|
query = `
|
||||||
SELECT collection, rkey, cid FROM records
|
SELECT collection, rkey, cid, COALESCE(did, '') FROM records
|
||||||
WHERE collection = ?
|
WHERE collection = ?
|
||||||
ORDER BY rkey DESC
|
ORDER BY rkey DESC
|
||||||
LIMIT ?
|
LIMIT ?
|
||||||
@@ -135,7 +168,7 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string,
|
|||||||
var records []Record
|
var records []Record
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var rec Record
|
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)
|
return nil, "", fmt.Errorf("failed to scan record: %w", err)
|
||||||
}
|
}
|
||||||
records = append(records, rec)
|
records = append(records, rec)
|
||||||
@@ -156,6 +189,58 @@ func (ri *RecordsIndex) ListRecords(collection string, limit int, cursor string,
|
|||||||
return records, nextCursor, nil
|
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
|
// Count returns the number of records in a collection
|
||||||
func (ri *RecordsIndex) Count(collection string) (int, error) {
|
func (ri *RecordsIndex) Count(collection string) (int, error) {
|
||||||
var count int
|
var count int
|
||||||
@@ -174,6 +259,7 @@ func (ri *RecordsIndex) TotalCount() (int, error) {
|
|||||||
|
|
||||||
// BackfillFromRepo populates the records index from an existing MST repo
|
// BackfillFromRepo populates the records index from an existing MST repo
|
||||||
// Compares MST count with index count - only backfills if they differ
|
// 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 {
|
func (ri *RecordsIndex) BackfillFromRepo(ctx context.Context, repoHandle *repo.Repo) error {
|
||||||
// Count records in MST
|
// Count records in MST
|
||||||
mstCount := 0
|
mstCount := 0
|
||||||
@@ -207,8 +293,8 @@ func (ri *RecordsIndex) BackfillFromRepo(ctx context.Context, repoHandle *repo.R
|
|||||||
defer tx.Rollback()
|
defer tx.Rollback()
|
||||||
|
|
||||||
stmt, err := tx.Prepare(`
|
stmt, err := tx.Prepare(`
|
||||||
INSERT OR REPLACE INTO records (collection, rkey, cid)
|
INSERT OR REPLACE INTO records (collection, rkey, cid, did)
|
||||||
VALUES (?, ?, ?)
|
VALUES (?, ?, ?, ?)
|
||||||
`)
|
`)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to prepare statement: %w", err)
|
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]
|
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 {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to index record %s: %w", key, err)
|
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)
|
slog.Info("Backfill complete", "records", recordCount)
|
||||||
return nil
|
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 ""
|
||||||
|
}
|
||||||
|
|||||||
@@ -50,7 +50,7 @@ func TestRecordsIndex_IndexRecord(t *testing.T) {
|
|||||||
defer ri.Close()
|
defer ri.Close()
|
||||||
|
|
||||||
// Index a record
|
// Index a record
|
||||||
err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123")
|
err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123", "")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("IndexRecord() error = %v", err)
|
t.Fatalf("IndexRecord() error = %v", err)
|
||||||
}
|
}
|
||||||
@@ -75,13 +75,13 @@ func TestRecordsIndex_IndexRecord_Upsert(t *testing.T) {
|
|||||||
defer ri.Close()
|
defer ri.Close()
|
||||||
|
|
||||||
// Index a record
|
// Index a record
|
||||||
err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123")
|
err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123", "")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("IndexRecord() first call error = %v", err)
|
t.Fatalf("IndexRecord() first call error = %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Update the same record with new CID
|
// 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 {
|
if err != nil {
|
||||||
t.Fatalf("IndexRecord() second call error = %v", err)
|
t.Fatalf("IndexRecord() second call error = %v", err)
|
||||||
}
|
}
|
||||||
@@ -118,7 +118,7 @@ func TestRecordsIndex_DeleteRecord(t *testing.T) {
|
|||||||
defer ri.Close()
|
defer ri.Close()
|
||||||
|
|
||||||
// Index a record
|
// Index a record
|
||||||
err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123")
|
err = ri.IndexRecord("io.atcr.hold.crew", "abc123", "bafyrei123", "")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("IndexRecord() error = %v", err)
|
t.Fatalf("IndexRecord() error = %v", err)
|
||||||
}
|
}
|
||||||
@@ -217,7 +217,7 @@ func TestRecordsIndex_ListRecords_Basic(t *testing.T) {
|
|||||||
{"ccc", "cid3"},
|
{"ccc", "cid3"},
|
||||||
}
|
}
|
||||||
for _, r := range records {
|
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)
|
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)
|
// Add records with different rkeys (TIDs are lexicographically ordered by time)
|
||||||
rkeys := []string{"3m3aaaaaaaaa", "3m3bbbbbbbbb", "3m3ccccccccc"}
|
rkeys := []string{"3m3aaaaaaaaa", "3m3bbbbbbbbb", "3m3ccccccccc"}
|
||||||
for _, rkey := range rkeys {
|
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)
|
t.Fatalf("IndexRecord() error = %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -286,7 +286,7 @@ func TestRecordsIndex_ListRecords_ReverseOrder(t *testing.T) {
|
|||||||
// Add records
|
// Add records
|
||||||
rkeys := []string{"3m3aaaaaaaaa", "3m3bbbbbbbbb", "3m3ccccccccc"}
|
rkeys := []string{"3m3aaaaaaaaa", "3m3bbbbbbbbb", "3m3ccccccccc"}
|
||||||
for _, rkey := range rkeys {
|
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)
|
t.Fatalf("IndexRecord() error = %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -324,7 +324,7 @@ func TestRecordsIndex_ListRecords_Limit(t *testing.T) {
|
|||||||
// Add 5 records
|
// Add 5 records
|
||||||
for i := range 5 {
|
for i := range 5 {
|
||||||
rkey := string(rune('a' + i))
|
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)
|
t.Fatalf("IndexRecord() error = %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -355,7 +355,7 @@ func TestRecordsIndex_ListRecords_Cursor(t *testing.T) {
|
|||||||
// Add 5 records
|
// Add 5 records
|
||||||
rkeys := []string{"a", "b", "c", "d", "e"}
|
rkeys := []string{"a", "b", "c", "d", "e"}
|
||||||
for _, rkey := range rkeys {
|
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)
|
t.Fatalf("IndexRecord() error = %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -430,7 +430,7 @@ func TestRecordsIndex_ListRecords_CursorReverse(t *testing.T) {
|
|||||||
// Add 5 records
|
// Add 5 records
|
||||||
rkeys := []string{"a", "b", "c", "d", "e"}
|
rkeys := []string{"a", "b", "c", "d", "e"}
|
||||||
for _, rkey := range rkeys {
|
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)
|
t.Fatalf("IndexRecord() error = %v", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -474,10 +474,10 @@ func TestRecordsIndex_Count(t *testing.T) {
|
|||||||
|
|
||||||
// Add records to two collections
|
// Add records to two collections
|
||||||
for i := range 3 {
|
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 {
|
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
|
// Count crew
|
||||||
@@ -527,10 +527,10 @@ func TestRecordsIndex_TotalCount(t *testing.T) {
|
|||||||
defer ri.Close()
|
defer ri.Close()
|
||||||
|
|
||||||
// Add records to multiple collections
|
// Add records to multiple collections
|
||||||
ri.IndexRecord("io.atcr.hold.crew", "a", "cid1")
|
ri.IndexRecord("io.atcr.hold.crew", "a", "cid1", "")
|
||||||
ri.IndexRecord("io.atcr.hold.crew", "b", "cid2")
|
ri.IndexRecord("io.atcr.hold.crew", "b", "cid2", "")
|
||||||
ri.IndexRecord("io.atcr.hold.captain", "self", "cid3")
|
ri.IndexRecord("io.atcr.hold.captain", "self", "cid3", "")
|
||||||
ri.IndexRecord("io.atcr.manifest", "abc123", "cid4")
|
ri.IndexRecord("io.atcr.manifest", "abc123", "cid4", "")
|
||||||
|
|
||||||
count, err := ri.TotalCount()
|
count, err := ri.TotalCount()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -581,9 +581,9 @@ func TestRecordsIndex_MultipleCollections(t *testing.T) {
|
|||||||
defer ri.Close()
|
defer ri.Close()
|
||||||
|
|
||||||
// Add records to different collections with same rkeys
|
// Add records to different collections with same rkeys
|
||||||
ri.IndexRecord("io.atcr.hold.crew", "abc", "cid-crew")
|
ri.IndexRecord("io.atcr.hold.crew", "abc", "cid-crew", "")
|
||||||
ri.IndexRecord("io.atcr.hold.captain", "abc", "cid-captain")
|
ri.IndexRecord("io.atcr.hold.captain", "abc", "cid-captain", "")
|
||||||
ri.IndexRecord("io.atcr.manifest", "abc", "cid-manifest")
|
ri.IndexRecord("io.atcr.manifest", "abc", "cid-manifest", "")
|
||||||
|
|
||||||
// Listing should only return records from requested collection
|
// Listing should only return records from requested collection
|
||||||
records, _, err := ri.ListRecords("io.atcr.hold.crew", 10, "", false)
|
records, _, err := ri.ListRecords("io.atcr.hold.crew", 10, "", false)
|
||||||
|
|||||||
@@ -382,6 +382,94 @@ func (rm *RepoManager) PutRecord(ctx context.Context, user models.Uid, collectio
|
|||||||
return rpath, cc, nil
|
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 {
|
func (rm *RepoManager) DeleteRecord(ctx context.Context, user models.Uid, collection, rkey string) error {
|
||||||
ctx, span := otel.Tracer("repoman").Start(ctx, "DeleteRecord")
|
ctx, span := otel.Tracer("repoman").Start(ctx, "DeleteRecord")
|
||||||
defer span.End()
|
defer span.End()
|
||||||
|
|||||||
+33
-1
@@ -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)
|
// Create Bluesky profile record (idempotent - check if exists first)
|
||||||
// This runs even if captain exists (for existing holds being upgraded)
|
// This runs even if captain exists (for existing holds being upgraded)
|
||||||
// Skip if no storage driver (e.g., in tests)
|
// Skip if no storage driver (e.g., in tests)
|
||||||
@@ -319,7 +327,9 @@ func (p *HoldPDS) CreateRecordsIndexEventHandler(broadcasterHandler func(context
|
|||||||
if op.RecCid != nil {
|
if op.RecCid != nil {
|
||||||
cidStr = op.RecCid.String()
|
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)
|
slog.Warn("Failed to index record", "collection", op.Collection, "rkey", op.Rkey, "error", err)
|
||||||
}
|
}
|
||||||
case EvtKindDeleteRecord:
|
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
|
// BackfillRecordsIndex populates the records index from existing MST data
|
||||||
func (p *HoldPDS) BackfillRecordsIndex(ctx context.Context) error {
|
func (p *HoldPDS) BackfillRecordsIndex(ctx context.Context) error {
|
||||||
if p.recordsIndex == nil {
|
if p.recordsIndex == nil {
|
||||||
|
|||||||
@@ -524,16 +524,15 @@ func TestBootstrap_CrewWithoutCaptain(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Verify crew wasn't duplicated (Bootstrap adds owner as crew, but they already exist)
|
// 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)
|
crewAfter, err := pds.ListCrewMembers(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("ListCrewMembers failed after bootstrap: %v", err)
|
t.Fatalf("ListCrewMembers failed after bootstrap: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Should have 2 crew members now: original + one added by bootstrap
|
// Should still have 1 crew member (hash-based rkey ensures upsert, not duplicate)
|
||||||
// (Bootstrap doesn't check for duplicates currently)
|
if len(crewAfter) != 1 {
|
||||||
if len(crewAfter) != 2 {
|
t.Errorf("Expected 1 crew member after bootstrap (upsert), got %d", len(crewAfter))
|
||||||
t.Logf("Note: Bootstrap added owner as crew even though they already existed")
|
|
||||||
t.Logf("Crew count after bootstrap: %d", len(crewAfter))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+106
-1
@@ -1,18 +1,39 @@
|
|||||||
// Package logging provides centralized structured logging using slog
|
// Package logging provides centralized structured logging using slog
|
||||||
// with configurable log levels. Call InitLogger() from main() to configure.
|
// 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 <container>
|
||||||
package logging
|
package logging
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"io"
|
"io"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"os"
|
"os"
|
||||||
|
"os/signal"
|
||||||
"strings"
|
"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.
|
// InitLogger initializes the global slog default logger with the specified log level.
|
||||||
// Valid levels: debug, info, warn, error (case-insensitive)
|
// Valid levels: debug, info, warn, error (case-insensitive)
|
||||||
// If level is empty or invalid, defaults to INFO.
|
// If level is empty or invalid, defaults to INFO.
|
||||||
// Call this from main() at startup.
|
// Call this from main() at startup.
|
||||||
|
//
|
||||||
|
// Also starts a signal handler for SIGUSR1 to toggle debug mode at runtime.
|
||||||
func InitLogger(level string) {
|
func InitLogger(level string) {
|
||||||
var logLevel slog.Level
|
var logLevel slog.Level
|
||||||
|
|
||||||
@@ -29,12 +50,96 @@ func InitLogger(level string) {
|
|||||||
logLevel = slog.LevelInfo
|
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{
|
opts := &slog.HandlerOptions{
|
||||||
Level: logLevel,
|
Level: levelVar,
|
||||||
}
|
}
|
||||||
|
|
||||||
handler := slog.NewTextHandler(os.Stdout, opts)
|
handler := slog.NewTextHandler(os.Stdout, opts)
|
||||||
slog.SetDefault(slog.New(handler))
|
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.
|
// SetupTestLogger configures logging for tests to reduce noise.
|
||||||
|
|||||||
Reference in New Issue
Block a user