From 963786f7ccf49e061089d30cec943eb3f5b7bf7a Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Thu, 16 Oct 2025 20:16:59 -0500 Subject: [PATCH] begin getRepo and subscribeRepos --- CLAUDE.md | 18 +++ cmd/hold/main.go | 11 +- deploy/request-crawl.sh | 55 +++++++++ pkg/hold/pds/events.go | 252 ++++++++++++++++++++++++++++++++++++++++ pkg/hold/pds/server.go | 5 + pkg/hold/pds/xrpc.go | 116 ++++++++++++++++-- 6 files changed, 448 insertions(+), 9 deletions(-) create mode 100755 deploy/request-crawl.sh create mode 100644 pkg/hold/pds/events.go diff --git a/CLAUDE.md b/CLAUDE.md index 07b0978..815df78 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -53,6 +53,11 @@ export STORAGE_ROOT_DIR=/tmp/atcr-hold export HOLD_OWNER=did:plc:your-did-here ./bin/atcr-hold # Hold starts immediately with embedded PDS + +# Request Bluesky relay crawl (makes your PDS discoverable) +./deploy/request-crawl.sh hold01.atcr.io +# Or specify a different relay: +./deploy/request-crawl.sh hold01.atcr.io https://custom-relay.example.com/xrpc/com.atproto.sync.requestCrawl ``` ## Architecture Overview @@ -371,6 +376,19 @@ Key insight: "Private" gates anonymous access, not authenticated access. This re - `POST /register` - Manual registration endpoint - `GET /health` - Health check +**Embedded PDS Endpoints:** + +Each hold service includes an embedded PDS (Personal Data Server) that stores captain + crew records: + +- `GET /xrpc/com.atproto.sync.getRepo?did={did}` - Download full repository as CAR file +- `GET /xrpc/com.atproto.sync.getRepo?did={did}&since={rev}` - Download repository diff since revision +- `GET /xrpc/com.atproto.sync.subscribeRepos` - WebSocket firehose for real-time events +- `GET /xrpc/com.atproto.sync.listRepos` - List all repositories (single-user PDS) +- `GET /.well-known/did.json` - DID document (did:web resolution) +- Standard ATProto repo endpoints (getRecord, listRecords, etc.) + +The `subscribeRepos` endpoint broadcasts #commit events whenever crew membership changes, allowing AppViews to monitor hold access control in real-time. + **Configuration:** Environment variables (see `.env.example`) - `HOLD_PUBLIC_URL` - Public URL of hold service (required) - `STORAGE_DRIVER` - Storage driver type (s3, filesystem) diff --git a/cmd/hold/main.go b/cmd/hold/main.go index cb52327..7e7b378 100644 --- a/cmd/hold/main.go +++ b/cmd/hold/main.go @@ -27,6 +27,7 @@ func main() { // This must happen before creating HoldService since service needs PDS for authorization var holdPDS *pds.HoldPDS var xrpcHandler *pds.XRPCHandler + var broadcaster *pds.EventBroadcaster if cfg.Database.Path != "" { // Generate did:web from public URL holdDID := pds.GenerateDIDFromURL(cfg.Server.PublicURL) @@ -44,7 +45,13 @@ func main() { log.Fatalf("Failed to bootstrap PDS: %v", err) } - log.Printf("Embedded PDS initialized successfully") + // Create event broadcaster for subscribeRepos firehose + broadcaster = pds.NewEventBroadcaster(holdDID, 100) // Keep 100 events for backfill + + // Wire up repo event handler to broadcaster + holdPDS.RepomgrRef().SetEventHandler(broadcaster.SetRepoEventHandler(), true) + + log.Printf("Embedded PDS initialized successfully with firehose enabled") } else { log.Fatalf("Database path is required for embedded PDS authorization") } @@ -59,7 +66,7 @@ func main() { if holdPDS != nil { holdDID := holdPDS.DID() blobStore := hold.NewHoldServiceBlobStore(service, holdDID) - xrpcHandler = pds.NewXRPCHandler(holdPDS, cfg.Server.PublicURL, blobStore) + xrpcHandler = pds.NewXRPCHandler(holdPDS, cfg.Server.PublicURL, blobStore, broadcaster) } // Setup HTTP routes diff --git a/deploy/request-crawl.sh b/deploy/request-crawl.sh new file mode 100755 index 0000000..071334e --- /dev/null +++ b/deploy/request-crawl.sh @@ -0,0 +1,55 @@ +#!/bin/bash +# +# Request crawl for a PDS from the Bluesky relay +# +# Usage: ./request-crawl.sh [relay-url] +# Example: ./request-crawl.sh hold01.atcr.io +# + +set -e + +DEFAULT_RELAY="https://bsky.network/xrpc/com.atproto.sync.requestCrawl" + +# Parse arguments +HOSTNAME="${1:-}" +RELAY_URL="${2:-$DEFAULT_RELAY}" + +# Validate hostname +if [ -z "$HOSTNAME" ]; then + echo "Error: hostname is required" >&2 + echo "" >&2 + echo "Usage: $0 [relay-url]" >&2 + echo "Example: $0 hold01.atcr.io" >&2 + echo "" >&2 + echo "Options:" >&2 + echo " hostname Hostname of the PDS to request crawl for (required)" >&2 + echo " relay-url Relay URL to send crawl request to (default: $DEFAULT_RELAY)" >&2 + exit 1 +fi + +# Log what we're doing +echo "Requesting crawl for hostname: $HOSTNAME" +echo "Sending to relay: $RELAY_URL" + +# Make the request +RESPONSE=$(curl -s -w "\n%{http_code}" -X POST "$RELAY_URL" \ + -H "Content-Type: application/json" \ + -d "{\"hostname\":\"$HOSTNAME\"}") + +# Split response and status code +HTTP_BODY=$(echo "$RESPONSE" | head -n -1) +HTTP_CODE=$(echo "$RESPONSE" | tail -n 1) + +# Check response +if [ "$HTTP_CODE" -ge 200 ] && [ "$HTTP_CODE" -lt 300 ]; then + echo "✅ Success! Crawl requested for $HOSTNAME" + if [ -n "$HTTP_BODY" ]; then + echo "Response: $HTTP_BODY" + fi +else + echo "❌ Failed with status $HTTP_CODE" >&2 + if [ -n "$HTTP_BODY" ]; then + echo "Response: $HTTP_BODY" >&2 + fi + exit 1 +fi diff --git a/pkg/hold/pds/events.go b/pkg/hold/pds/events.go new file mode 100644 index 0000000..a873c33 --- /dev/null +++ b/pkg/hold/pds/events.go @@ -0,0 +1,252 @@ +package pds + +import ( + "context" + "encoding/json" + "log" + "sync" + "time" + + atproto "github.com/bluesky-social/indigo/api/atproto" + lexutil "github.com/bluesky-social/indigo/lex/util" + "github.com/gorilla/websocket" +) + +// EventBroadcaster manages WebSocket connections and broadcasts repo events +type EventBroadcaster struct { + mu sync.RWMutex + subscribers map[*Subscriber]bool + eventSeq int64 + eventHistory []HistoricalEvent // Ring buffer for cursor backfill + maxHistory int + holdDID string // DID of the hold for setting repo field +} + +// Subscriber represents a WebSocket client subscribed to the firehose +type Subscriber struct { + conn *websocket.Conn + send chan *RepoCommitEvent + cursor int64 // Last sequence number this subscriber has seen +} + +// HistoricalEvent stores past events for cursor-based backfill +type HistoricalEvent struct { + Seq int64 + Event *RepoCommitEvent +} + +// RepoCommitEvent represents a #commit event in subscribeRepos +type RepoCommitEvent struct { + Seq int64 `json:"seq" cborgen:"seq"` + Repo string `json:"repo" cborgen:"repo"` + Commit string `json:"commit" cborgen:"commit"` // CID string + Rev string `json:"rev" cborgen:"rev"` + Since *string `json:"since,omitempty" cborgen:"since,omitempty"` + Blocks []byte `json:"blocks" cborgen:"blocks"` // CAR slice bytes + Ops []*atproto.SyncSubscribeRepos_RepoOp `json:"ops" cborgen:"ops"` + Time string `json:"time" cborgen:"time"` + Type string `json:"$type" cborgen:"$type"` // Always "#commit" +} + +// NewEventBroadcaster creates a new event broadcaster +func NewEventBroadcaster(holdDID string, maxHistory int) *EventBroadcaster { + if maxHistory <= 0 { + maxHistory = 100 // Default to keeping 100 events + } + + return &EventBroadcaster{ + subscribers: make(map[*Subscriber]bool), + eventSeq: 0, + eventHistory: make([]HistoricalEvent, 0, maxHistory), + maxHistory: maxHistory, + holdDID: holdDID, + } +} + +// Subscribe adds a new WebSocket subscriber +func (b *EventBroadcaster) Subscribe(conn *websocket.Conn, cursor int64) *Subscriber { + sub := &Subscriber{ + conn: conn, + send: make(chan *RepoCommitEvent, 10), // Buffer 10 events + cursor: cursor, + } + + b.mu.Lock() + b.subscribers[sub] = true + currentSeq := b.eventSeq + b.mu.Unlock() + + // Send historical events if cursor is provided and < current seq + if cursor > 0 && cursor < currentSeq { + go b.backfillSubscriber(sub, cursor) + } + + // Start goroutine to handle sending events to this subscriber + go b.handleSubscriber(sub) + + return sub +} + +// Unsubscribe removes a WebSocket subscriber +func (b *EventBroadcaster) Unsubscribe(sub *Subscriber) { + b.mu.Lock() + defer b.mu.Unlock() + + if _, ok := b.subscribers[sub]; ok { + delete(b.subscribers, sub) + close(sub.send) + } +} + +// Broadcast sends an event to all subscribers +func (b *EventBroadcaster) Broadcast(ctx context.Context, event *RepoEvent) { + b.mu.Lock() + defer b.mu.Unlock() + + // Increment sequence + b.eventSeq++ + seq := b.eventSeq + + // Convert RepoEvent to RepoCommitEvent + commitEvent := b.convertToCommitEvent(event, seq) + + // Store in history for backfill + b.addToHistory(seq, commitEvent) + + // Broadcast to all subscribers + for sub := range b.subscribers { + select { + case sub.send <- commitEvent: + // Sent successfully + default: + // Subscriber's buffer is full, skip (they'll get disconnected for being too slow) + log.Printf("Warning: subscriber buffer full, skipping event seq=%d", seq) + } + } +} + +// convertToCommitEvent converts a RepoEvent to a RepoCommitEvent +func (b *EventBroadcaster) convertToCommitEvent(event *RepoEvent, seq int64) *RepoCommitEvent { + // Convert RepoOps to atproto.SyncSubscribeRepos_RepoOp + ops := make([]*atproto.SyncSubscribeRepos_RepoOp, len(event.Ops)) + for i, op := range event.Ops { + action := string(op.Kind) // "create", "update", "delete" + path := op.Collection + "/" + op.Rkey + + // Convert CID to LexLink if present + var cidLink *lexutil.LexLink + if op.RecCid != nil { + link := lexutil.LexLink(*op.RecCid) + cidLink = &link + } + + ops[i] = &atproto.SyncSubscribeRepos_RepoOp{ + Action: action, + Path: path, + Cid: cidLink, + } + } + + // Event.NewRoot is a cid.Cid, convert to string + commitCID := event.NewRoot.String() + + return &RepoCommitEvent{ + Seq: seq, + Repo: b.holdDID, // Set to hold's DID + Commit: commitCID, + Rev: event.Rev, + Since: event.Since, + Blocks: event.RepoSlice, // CAR slice bytes + Ops: ops, + Time: time.Now().Format(time.RFC3339), + Type: "#commit", + } +} + +// addToHistory adds an event to the history ring buffer +func (b *EventBroadcaster) addToHistory(seq int64, event *RepoCommitEvent) { + he := HistoricalEvent{ + Seq: seq, + Event: event, + } + + // Simple ring buffer: keep last N events + if len(b.eventHistory) >= b.maxHistory { + // Remove oldest event + b.eventHistory = b.eventHistory[1:] + } + b.eventHistory = append(b.eventHistory, he) +} + +// backfillSubscriber sends historical events to a subscriber +func (b *EventBroadcaster) backfillSubscriber(sub *Subscriber, cursor int64) { + b.mu.RLock() + defer b.mu.RUnlock() + + for _, he := range b.eventHistory { + if he.Seq > cursor { + select { + case sub.send <- he.Event: + // Sent + case <-time.After(5 * time.Second): + // Timeout, subscriber too slow + log.Printf("Backfill timeout for subscriber at seq=%d", he.Seq) + return + } + } + } +} + +// handleSubscriber handles sending events to a subscriber over WebSocket +func (b *EventBroadcaster) handleSubscriber(sub *Subscriber) { + defer func() { + b.Unsubscribe(sub) + sub.conn.Close() + }() + + for event := range sub.send { + // Encode as CBOR + cborBytes, err := encodeCBOR(event) + if err != nil { + log.Printf("Failed to encode event as CBOR: %v", err) + continue + } + + // Write CBOR message to WebSocket + err = sub.conn.WriteMessage(websocket.BinaryMessage, cborBytes) + if err != nil { + log.Printf("Failed to write to websocket: %v", err) + return + } + + // Update cursor + sub.cursor = event.Seq + } +} + +// encodeCBOR encodes an event as CBOR +func encodeCBOR(event *RepoCommitEvent) ([]byte, error) { + // For now, use JSON encoding wrapped in CBOR envelope + // In production, you'd use proper CBOR encoding + // The atproto spec requires DAG-CBOR with specific header + + // Simple approach: encode as JSON for MVP + // Real implementation needs proper CBOR-gen types + return json.Marshal(event) +} + +// SetRepoEventHandler creates a callback to be registered with RepoManager +func (b *EventBroadcaster) SetRepoEventHandler() func(context.Context, *RepoEvent) { + return func(ctx context.Context, event *RepoEvent) { + // Broadcast the event to all subscribers + // The holdDID is already set in the broadcaster + b.Broadcast(ctx, event) + } +} + +// GetCurrentSeq returns the current event sequence number +func (b *EventBroadcaster) GetCurrentSeq() int64 { + b.mu.RLock() + defer b.mu.RUnlock() + return b.eventSeq +} diff --git a/pkg/hold/pds/server.go b/pkg/hold/pds/server.go index 92f134f..c3b267f 100644 --- a/pkg/hold/pds/server.go +++ b/pkg/hold/pds/server.go @@ -102,6 +102,11 @@ func (p *HoldPDS) SigningKey() *atcrypto.PrivateKeyK256 { return p.signingKey } +// RepomgrRef returns a reference to the RepoManager for event handler setup +func (p *HoldPDS) RepomgrRef() *RepoManager { + return p.repomgr +} + // Bootstrap initializes the hold with the captain record and owner as first crew member func (p *HoldPDS) Bootstrap(ctx context.Context, ownerDID string, public bool, allowAllCrew bool) error { if ownerDID == "" { diff --git a/pkg/hold/pds/xrpc.go b/pkg/hold/pds/xrpc.go index ae4cf27..96b0d2d 100644 --- a/pkg/hold/pds/xrpc.go +++ b/pkg/hold/pds/xrpc.go @@ -5,11 +5,13 @@ import ( "encoding/json" "fmt" "net/http" + "strconv" "strings" "atcr.io/pkg/atproto" lexutil "github.com/bluesky-social/indigo/lex/util" "github.com/bluesky-social/indigo/repo" + "github.com/gorilla/websocket" "github.com/ipfs/go-cid" "github.com/ipld/go-car" carutil "github.com/ipld/go-car/util" @@ -19,9 +21,10 @@ import ( // XRPCHandler handles XRPC requests for the embedded PDS type XRPCHandler struct { - pds *HoldPDS - publicURL string - blobStore BlobStore + pds *HoldPDS + publicURL string + blobStore BlobStore + broadcaster *EventBroadcaster } // BlobStore interface wraps the existing hold service storage operations @@ -33,11 +36,12 @@ type BlobStore interface { } // NewXRPCHandler creates a new XRPC handler -func NewXRPCHandler(pds *HoldPDS, publicURL string, blobStore BlobStore) *XRPCHandler { +func NewXRPCHandler(pds *HoldPDS, publicURL string, blobStore BlobStore, broadcaster *EventBroadcaster) *XRPCHandler { return &XRPCHandler{ - pds: pds, - publicURL: publicURL, - blobStore: blobStore, + pds: pds, + publicURL: publicURL, + blobStore: blobStore, + broadcaster: broadcaster, } } @@ -72,6 +76,8 @@ func (h *XRPCHandler) RegisterHandlers(mux *http.ServeMux) { // Sync endpoints mux.HandleFunc("/xrpc/com.atproto.sync.listRepos", corsMiddleware(h.HandleListRepos)) mux.HandleFunc("/xrpc/com.atproto.sync.getRecord", corsMiddleware(h.HandleSyncGetRecord)) + mux.HandleFunc("/xrpc/com.atproto.sync.getRepo", corsMiddleware(h.HandleGetRepo)) + mux.HandleFunc("/xrpc/com.atproto.sync.subscribeRepos", corsMiddleware(h.HandleSubscribeRepos)) // Blob endpoints (wrap existing presigned URL logic) mux.HandleFunc("/xrpc/com.atproto.repo.uploadBlob", corsMiddleware(h.HandleUploadBlob)) @@ -440,6 +446,102 @@ func (h *XRPCHandler) HandleSyncGetRecord(w http.ResponseWriter, r *http.Request w.Write(buf.Bytes()) } +// HandleGetRepo returns the full repository as a CAR file +// This is the critical endpoint for relay crawling and Bluesky discovery +func (h *XRPCHandler) HandleGetRepo(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + // Get required 'did' parameter + did := r.URL.Query().Get("did") + if did == "" { + http.Error(w, "missing required parameter: did", http.StatusBadRequest) + return + } + + // Validate DID matches this PDS + if did != h.pds.DID() { + http.Error(w, "repo not found", http.StatusNotFound) + return + } + + // Get optional 'since' parameter for diff export + since := r.URL.Query().Get("since") + + // Set CAR content type + w.Header().Set("Content-Type", "application/vnd.ipld.car") + + // Stream the repository CAR file directly to the response + // ReadRepo handles full export or diff based on 'since' parameter + err := h.pds.repomgr.ReadRepo(r.Context(), h.pds.uid, since, w) + if err != nil { + // Error already written to response by ReadRepo streaming + // Log it but don't try to write another HTTP error + fmt.Printf("Error streaming repo CAR: %v\n", err) + return + } +} + +// WebSocket upgrader +var upgrader = websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { + // Allow all origins for MVP (ATProto firehose is public) + return true + }, +} + +// HandleSubscribeRepos handles WebSocket connections for the firehose +// This is the real-time event stream for repo changes +func (h *XRPCHandler) HandleSubscribeRepos(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + + // Check if broadcaster is configured + if h.broadcaster == nil { + http.Error(w, "firehose not enabled", http.StatusNotImplemented) + return + } + + // Get optional cursor parameter for backfill + var cursor int64 = 0 + if cursorStr := r.URL.Query().Get("cursor"); cursorStr != "" { + var err error + cursor, err = strconv.ParseInt(cursorStr, 10, 64) + if err != nil { + http.Error(w, "invalid cursor parameter", http.StatusBadRequest) + return + } + } + + // Upgrade to WebSocket + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + fmt.Printf("WebSocket upgrade failed: %v\n", err) + return + } + + // Subscribe to events + sub := h.broadcaster.Subscribe(conn, cursor) + + // The broadcaster's handleSubscriber goroutine will manage this connection + // We just need to keep reading to detect client disconnects + go func() { + defer h.broadcaster.Unsubscribe(sub) + for { + // Read messages from client (mostly just to detect disconnect) + _, _, err := conn.ReadMessage() + if err != nil { + // Client disconnected + break + } + } + }() +} + // HandleUploadBlob wraps existing presigned upload URL logic func (h *XRPCHandler) HandleUploadBlob(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost {