mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-09-26 12:14:17 +00:00
begin getRepo and subscribeRepos
This commit is contained in:
Evan Jarrett
@@ -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
|
||||
}
|
||||
@@ -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 == "" {
|
||||
|
||||
+109
-7
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user