labelmerge initial commit · whey.party/red-dwarf-server@ae57860

+2 -1

.gitignore

··· 1 1 cmd/aturilist/badger_data 2 - cmd/backstream/temp 2 + cmd/backstream/temp 3 + cmd/labelmerge/badger_cache

+112

cmd/labelmerge/handlers.go

··· 1 + package main 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "strconv" 7 + "time" 8 + 9 + comatprototypes "github.com/bluesky-social/indigo/api/atproto" 10 + appreddwarflabelmerge "tangled.org/whey.party/red-dwarf-server/labelmerge/lex" 11 + ) 12 + 13 + func (s *Server) handleAppReddwarfLabelmergeQueryLabels( 14 + ctx context.Context, 15 + inputl []string, 16 + inputs []string, 17 + strict *bool, 18 + ) (*appreddwarflabelmerge.QueryLabels_Output, error) { 19 + 20 + // Build the query struct your service expects 21 + query := IncomingQuery{ 22 + Labelers: inputl, 23 + Subjects: inputs, 24 + Strict: strict != nil && *strict, 25 + } 26 + 27 + // enforce limits 28 + if len(query.Labelers) > MaxLabelersPerQuery || len(query.Subjects) > MaxSubjectsPerQuery { 29 + return &appreddwarflabelmerge.QueryLabels_Output{ 30 + Error: []*appreddwarflabelmerge.QueryLabels_Error{ 31 + { 32 + S: "too_many_labels_or_subjects", 33 + E: ptrString(fmt.Sprintf("Labelers: %d, Subjects: %d", len(query.Labelers), len(query.Subjects))), 34 + }, 35 + }, 36 + }, nil 37 + } 38 + 39 + // Setup context with timeout 40 + ctx, cancel := context.WithTimeout(ctx, IncomingQueryTimeout) 41 + defer cancel() 42 + 43 + promise := &QueryPromise{ 44 + ID: strconv.FormatInt(time.Now().UnixNano(), 10), 45 + Query: query, 46 + Response: make(chan FinalResponse, 1), 47 + Ctx: ctx, 48 + } 49 + 50 + s.Service.HandlePromise(promise) 51 + 52 + select { 53 + case result := <-promise.Response: 54 + // convert FinalResponse -> QueryLabels_Output 55 + // Flatten your map into a slice 56 + var flatLabels []*comatprototypes.LabelDefs_Label 57 + for _, submap := range result.Labels { // result.Labels is map[labelerDID]map[subjectURI][]*Label 58 + for _, labels := range submap { 59 + flatLabels = append(flatLabels, SliceToPtrSlice(labels)...) 60 + } 61 + } 62 + 63 + out := &appreddwarflabelmerge.QueryLabels_Output{ 64 + Labels: flatLabels, 65 + } 66 + 67 + if result.Error != nil { 68 + for _, msg := range result.Error.LabelerResolutionFailure { 69 + out.Error = append(out.Error, &appreddwarflabelmerge.QueryLabels_Error{ 70 + S: "labeler_resolution_failure", 71 + E: &msg, 72 + }) 73 + } 74 + for _, msg := range result.Error.QueryLabelsTooManyPages { 75 + out.Error = append(out.Error, &appreddwarflabelmerge.QueryLabels_Error{ 76 + S: "too_many_pages", 77 + E: &msg, 78 + }) 79 + } 80 + for _, msg := range result.Error.LabelerQueryFailure { 81 + out.Error = append(out.Error, &appreddwarflabelmerge.QueryLabels_Error{ 82 + S: "labeler_query_failure", 83 + E: &msg, 84 + }) 85 + } 86 + } 87 + 88 + // handle strict mode 89 + if query.Strict && out.Error != nil && len(out.Error) > 0 { 90 + return out, fmt.Errorf("strict mode failure: %v", out.Error) 91 + } 92 + 93 + return out, nil 94 + 95 + case <-ctx.Done(): 96 + timeoutMsg := "one of the queries didn't return in time, try again later" 97 + out := &appreddwarflabelmerge.QueryLabels_Output{ 98 + Error: []*appreddwarflabelmerge.QueryLabels_Error{ 99 + { 100 + S: "timeout", 101 + E: &timeoutMsg, 102 + }, 103 + }, 104 + } 105 + if query.Strict { 106 + return out, fmt.Errorf(timeoutMsg) 107 + } 108 + return out, nil 109 + } 110 + } 111 + 112 + func ptrString(s string) *string { return &s }

+980

cmd/labelmerge/main.go

··· 1 + package main 2 + 3 + import ( 4 + "bytes" 5 + "context" 6 + "encoding/json" 7 + "fmt" 8 + "log" 9 + "net/http" 10 + "net/url" 11 + "slices" 12 + "strconv" 13 + "strings" 14 + "sync" 15 + "time" 16 + 17 + comatprototypes "github.com/bluesky-social/indigo/api/atproto" 18 + "github.com/bluesky-social/indigo/atproto/identity" 19 + "github.com/bluesky-social/indigo/atproto/syntax" 20 + "github.com/gin-gonic/gin" 21 + "github.com/labstack/echo/v4" 22 + "github.com/labstack/echo/v4/middleware" 23 + lru "tangled.org/whey.party/red-dwarf-server/labelmerge/lru" 24 + ) 25 + 26 + // --- Constants --- 27 + const ( 28 + ServerPort = ":3879" 29 + MaxLabelersPerQuery = 20 30 + MaxSubjectsPerQuery = 100 31 + LRUTTL = 30 * time.Minute 32 + LabelerResolverTTL = 1 * time.Hour 33 + LabelerResolutionTimeout = 4 * time.Second 34 + BusQueueBatchInterval = 500 * time.Millisecond // T seconds, set to 0.5s 35 + MaxSubjectsPerExternalRequest = 50 36 + MaxQueryPages = 16 37 + IncomingQueryTimeout = 30 * time.Second 38 + ) 39 + 40 + // --- Core Data Structures (AT Protocol & Internal) --- 41 + 42 + // Label definition updated to include CID, Neg, and Exp fields, matching the full AT Protocol specification. 43 + // type Label struct { 44 + // Ver int `json:"ver"` 45 + // Src string `json:"src"` // DID of the actor who created this label. 46 + // URI string `json:"uri"` // AT URI of the record, repository (account), or other resource. 47 + // CID string `json:"cid,omitempty"` // Optionally, CID specifying the specific version. 48 + // Val string `json:"val"` // The short string name of the value or type of this label. 49 + // Neg bool `json:"neg,omitempty"` // If true, this is a negation label. 50 + // CTS string `json:"cts"` // Timestamp when this label was created. 51 + // Exp string `json:"exp,omitempty"` // Timestamp at which this label expires. 52 + // Sig struct { 53 + // Bytes string `json:"$bytes"` 54 + // } `json:"sig"` // Signature of dag-cbor encoded label. 55 + // } 56 + 57 + type Label = comatprototypes.LabelDefs_Label 58 + 59 + /* 60 + type LabelDefs_Label struct { 61 + // cid: Optionally, CID specifying the specific version of 'uri' resource this label applies to. 62 + Cid *string `json:"cid,omitempty" cborgen:"cid,omitempty"` 63 + // cts: Timestamp when this label was created. 64 + Cts string `json:"cts" cborgen:"cts"` 65 + // exp: Timestamp at which this label expires (no longer applies). 66 + Exp *string `json:"exp,omitempty" cborgen:"exp,omitempty"` 67 + // neg: If true, this is a negation label, overwriting a previous label. 68 + Neg *bool `json:"neg,omitempty" cborgen:"neg,omitempty"` 69 + // sig: Signature of dag-cbor encoded label. 70 + Sig lexutil.LexBytes `json:"sig,omitempty" cborgen:"sig,omitempty"` 71 + // src: DID of the actor who created this label. 72 + Src string `json:"src" cborgen:"src"` 73 + // uri: AT URI of the record, repository (account), or other resource that this label applies to. 74 + Uri string `json:"uri" cborgen:"uri"` 75 + // val: The short string name of the value or type of this label. 76 + Val string `json:"val" cborgen:"val"` 77 + // ver: The AT Protocol version of the label object. 78 + Ver *int64 `json:"ver,omitempty" cborgen:"ver,omitempty"` 79 + } 80 + LabelDefs_Label is a "label" in the com.atproto.label.defs schema. 81 + 82 + Metadata tag on an atproto resource (eg, repo or record). 83 + 84 + func (t *comatprototypes.LabelDefs_Label) MarshalCBOR(w io.Writer) error 85 + func (t *comatprototypes.LabelDefs_Label) UnmarshalCBOR(r io.Reader) (err error) 86 + */ 87 + 88 + type IncomingQuery struct { 89 + Subjects []string `json:"s"` 90 + Labelers []string `json:"l"` 91 + Strict bool `json:"strict"` 92 + } 93 + 94 + // Req is the internal unit of work sent to the Bus Queue (Subject -> Labeler[]) 95 + type Req struct { 96 + Subject string 97 + Labelers []string 98 + } 99 + 100 + // QueryError defines the structure for reporting errors in the response body. 101 + type QueryError struct { 102 + QueryLabelsTooManyPages []string `json:"queryLabelsTooManyPages,omitempty"` 103 + LabelerResolutionFailure []string `json:"labelerResolutionFailure,omitempty"` 104 + LabelerQueryFailure []string `json:"labelerQueryFailure,omitempty"` // NEW FAILURE TYPE 105 + } 106 + 107 + func (qe *QueryError) HasLabelerError(s string) bool { 108 + sets := [][]string{ 109 + //qe.QueryLabelsTooManyPages, // not a labeler error. it should be negligable maybe with the larger tolerance of 10 pages 110 + qe.LabelerResolutionFailure, 111 + qe.LabelerQueryFailure, 112 + } 113 + 114 + for _, set := range sets { 115 + if slices.Contains(set, s) { 116 + return true 117 + } 118 + } 119 + return false 120 + } 121 + 122 + // FinalResponse is the structure returned to the client. 123 + type FinalResponse struct { 124 + Labels Results `json:"labels"` // map[labelerDID][subjectURI] -> []Label 125 + Error *QueryError `json:"error,omitempty"` 126 + } 127 + 128 + type QueryPromise struct { 129 + ID string 130 + Query IncomingQuery 131 + Response chan FinalResponse 132 + Ctx context.Context 133 + Done sync.Once 134 + } 135 + 136 + // ResolvedLabels represents labels for a subject/labeler pair (empty slice means success, but no labels) 137 + type ResolvedLabels = comatprototypes.LabelSubscribeLabels_Labels //[]Label // keep seq for future use but in the mean time use 67 138 + 139 + // Labeler Domain Cache Entry 140 + type LabelerDomainEntry struct { 141 + Domain string 142 + ExpiresAt time.Time 143 + Err error // Store resolution errors temporarily 144 + } 145 + 146 + // globalPromiseTrackers maps Promise ID to the state object tracking its completion. 147 + var globalPromiseTrackers = struct { 148 + mu sync.Mutex 149 + m map[string]*PromiseTracker 150 + }{m: make(map[string]*PromiseTracker)} 151 + 152 + type Results map[string]map[string][]Label 153 + 154 + // PromiseTracker manages the state of a single incoming promise, aggregating partial results. 155 + type PromiseTracker struct { 156 + P *QueryPromise 157 + 158 + Mu sync.Mutex 159 + 160 + TotalPairs int // Total (subject, labeler) pairs requested 161 + CompletedPairs int 162 + Done chan struct{} 163 + 164 + // Aggregated Results and Errors 165 + Results Results 166 + Errors *QueryError 167 + } 168 + 169 + func (r Results) FilterErrorLabels() Results { 170 + // please document this! but empty labels objects is inferred to be an error 171 + // since earlier we make it so if error it result in an empty labels object 172 + out := make(Results, len(r)) 173 + 174 + for k, v := range r { 175 + if len(v) == 0 { 176 + continue 177 + } 178 + out[k] = v 179 + } 180 + 181 + return out 182 + } 183 + 184 + func NewPromiseTracker(p *QueryPromise, totalPairs int) *PromiseTracker { 185 + // Initialize Results map 186 + results := make(Results) 187 + for _, l := range p.Query.Labelers { 188 + results[l] = make(map[string][]Label) 189 + } 190 + 191 + return &PromiseTracker{ 192 + P: p, 193 + TotalPairs: totalPairs, 194 + CompletedPairs: 0, 195 + Done: make(chan struct{}), 196 + Results: results, 197 + Errors: &QueryError{}, 198 + } 199 + } 200 + 201 + // warning take great care when using this 202 + func SliceToPtrSlice[T any](in []T) []*T { 203 + out := make([]*T, len(in)) 204 + for i := range in { 205 + out[i] = &in[i] 206 + } 207 + return out 208 + } 209 + 210 + // warning take great care when using this 211 + func PtrSliceToSlice[T any](in []*T) []T { 212 + out := make([]T, len(in)) 213 + for i, p := range in { 214 + if p != nil { 215 + out[i] = *p 216 + } 217 + } 218 + return out 219 + } 220 + 221 + // StoreResult atomically stores a result (or error) and checks if the promise is complete. 222 + func (t *PromiseTracker) StoreResult(labelerDID, subjectURI string, labels []*Label, isExternalFetch bool) { 223 + t.Mu.Lock() 224 + defer t.Mu.Unlock() 225 + 226 + if t.CompletedPairs >= t.TotalPairs { 227 + return 228 + } 229 + 230 + // Store Labels 231 + // only if labelerDID doesnt error out 232 + if !t.Errors.HasLabelerError(labelerDID) { 233 + if t.Results[labelerDID] == nil { 234 + t.Results[labelerDID] = make(map[string][]Label) 235 + } 236 + t.Results[labelerDID][subjectURI] = PtrSliceToSlice(labels) 237 + } 238 + 239 + t.CompletedPairs++ 240 + 241 + // Check Completion 242 + if t.CompletedPairs == t.TotalPairs { 243 + close(t.Done) 244 + } 245 + } 246 + 247 + // FinalizeAndRespond constructs the final response, sends it, and cleans up the tracker. 248 + func (t *PromiseTracker) FinalizeAndRespond(p *QueryPromise, timeoutError string) { 249 + log.Println("response done (finalizing)!") 250 + p.Done.Do(func() { 251 + resp := FinalResponse{ 252 + Labels: t.Results.FilterErrorLabels(), 253 + } 254 + 255 + if timeoutError != "" { 256 + resp.Labels = nil 257 + resp.Error = &QueryError{ 258 + LabelerResolutionFailure: []string{timeoutError}, 259 + } 260 + } else if len(t.Errors.LabelerResolutionFailure) > 0 || 261 + len(t.Errors.QueryLabelsTooManyPages) > 0 || 262 + len(t.Errors.LabelerQueryFailure) > 0 { // Check the new error field 263 + resp.Error = t.Errors 264 + } 265 + 266 + // Cleanup: Remove tracker from global map 267 + globalPromiseTrackers.mu.Lock() 268 + delete(globalPromiseTrackers.m, p.ID) 269 + globalPromiseTrackers.mu.Unlock() 270 + 271 + p.Response <- resp 272 + }) 273 + } 274 + 275 + // --- Component Implementations --- 276 + 277 + // InFlightManager handles deduplication and tracks waiting promises. 278 + type InFlightManager struct { 279 + mu sync.Mutex 280 + // Key: subject|labelerDID. Value: list of promises waiting for this result. 281 + requests map[string][]*QueryPromise 282 + } 283 + 284 + func NewInFlightManager() *InFlightManager { 285 + return &InFlightManager{requests: make(map[string][]*QueryPromise)} 286 + } 287 + 288 + func (m *InFlightManager) key(subject, labeler string) string { 289 + return subject + "|" + labeler 290 + } 291 + 292 + // CheckAndRegister registers the promise to wait. Returns true if already in flight. 293 + func (m *InFlightManager) CheckAndRegister(subject, labeler string, p *QueryPromise) bool { 294 + k := m.key(subject, labeler) 295 + m.mu.Lock() 296 + defer m.mu.Unlock() 297 + 298 + if _, exists := m.requests[k]; exists { 299 + m.requests[k] = append(m.requests[k], p) 300 + return true 301 + } 302 + 303 + m.requests[k] = []*QueryPromise{p} 304 + return false 305 + } 306 + 307 + // Resolve retrieves waiting promises and clears the entry. 308 + func (m *InFlightManager) Resolve(subject, labeler string) []*QueryPromise { 309 + k := m.key(subject, labeler) 310 + m.mu.Lock() 311 + defer m.mu.Unlock() 312 + 313 + waiters, exists := m.requests[k] 314 + if !exists { 315 + return nil 316 + } 317 + 318 + delete(m.requests, k) 319 + return waiters 320 + } 321 + 322 + // RemovePromise removes a specific promise from all in-flight requests (used for cleanup) 323 + func (m *InFlightManager) RemovePromise(p *QueryPromise) { 324 + m.mu.Lock() 325 + defer m.mu.Unlock() 326 + 327 + for k, waiters := range m.requests { 328 + filtered := make([]*QueryPromise, 0, len(waiters)) 329 + for _, waiter := range waiters { 330 + if waiter.ID != p.ID { 331 + filtered = append(filtered, waiter) 332 + } 333 + } 334 + if len(filtered) == 0 { 335 + delete(m.requests, k) 336 + } else { 337 + m.requests[k] = filtered 338 + } 339 + } 340 + } 341 + 342 + // --- Main Service Coordinator --- 343 + 344 + type LabelResolutionService struct { 345 + incomingPromises chan *QueryPromise 346 + busQueueIn chan Req 347 + 348 + lru *lru.Cache[string, ResolvedLabels] 349 + labelerResolver identity.Directory 350 + inFlightManager *InFlightManager 351 + 352 + wg sync.WaitGroup 353 + } 354 + 355 + func NewLabelResolutionService() *LabelResolutionService { 356 + // Create serialization functions for our cache values 357 + serialize := func(v ResolvedLabels) ([]byte, error) { 358 + var buf bytes.Buffer 359 + err := v.MarshalCBOR(&buf) 360 + if err != nil { 361 + return nil, err 362 + } 363 + //return json.Marshal(v) 364 + return buf.Bytes(), nil 365 + } 366 + 367 + deserialize := func(data []byte) (ResolvedLabels, error) { 368 + //var result ResolvedLabels 369 + 370 + // Unmarshal from CBOR 371 + var buf bytes.Buffer = *bytes.NewBuffer(data) 372 + var labels ResolvedLabels 373 + err := labels.UnmarshalCBOR(&buf) 374 + if err != nil { 375 + return labels, err 376 + } 377 + return labels, nil 378 + //return result, json.Unmarshal(data, &result) 379 + } 380 + 381 + // Initialize the LRU cache with a capacity of 1,000,000 entries 382 + cache, err := lru.New[string]("./badger_cache", serialize, deserialize) 383 + if err != nil { 384 + log.Fatalf("Failed to initialize LRU cache: %v", err) 385 + } 386 + cache.SetCapacity(1000000) 387 + 388 + dir := identity.DefaultDirectory() // Cached with 24hr TTL 389 + 390 + lrs := &LabelResolutionService{ 391 + incomingPromises: make(chan *QueryPromise), 392 + busQueueIn: make(chan Req, 1000), 393 + lru: cache, 394 + labelerResolver: dir, 395 + inFlightManager: NewInFlightManager(), 396 + } 397 + 398 + lrs.wg.Add(3) 399 + go lrs.runLabelResolverPipeline() 400 + go lrs.runBusQueue() 401 + // Note: runQueryLabelsHandler is implicitly part of runBusQueue's flush logic via handleBatchedRequests 402 + 403 + return lrs 404 + } 405 + 406 + func (s *LabelResolutionService) Stop() { 407 + log.Println("Stopping Label Resolution Service components...") 408 + close(s.incomingPromises) 409 + // Do not close busQueueIn immediately, let the BusQueue consumer drain it. 410 + s.wg.Wait() 411 + if err := s.lru.Close(); err != nil { 412 + log.Printf("Error closing LRU cache: %v", err) 413 + } 414 + } 415 + 416 + func (s *LabelResolutionService) HandlePromise(p *QueryPromise) { 417 + s.incomingPromises <- p 418 + } 419 + 420 + // --- 1. Label Resolver Pipeline (Handles Promises and LRU Checks) --- 421 + 422 + func (s *LabelResolutionService) runLabelResolverPipeline() { 423 + defer s.wg.Done() 424 + for p := range s.incomingPromises { 425 + s.resolvePromise(p) 426 + } 427 + } 428 + 429 + // resolvePromise handles the full lifecycle of a single incoming request. 430 + func (s *LabelResolutionService) resolvePromise(p *QueryPromise) { 431 + q := p.Query 432 + 433 + totalPairs := len(q.Subjects) * len(q.Labelers) 434 + if totalPairs == 0 { 435 + // Empty query case 436 + p.Response <- FinalResponse{Labels: make(Results)} 437 + return 438 + } 439 + 440 + tracker := NewPromiseTracker(p, totalPairs) 441 + 442 + // Phase 1: Check cache and register in-flight status 443 + s.initialCacheCheckAndQueue(p, tracker) 444 + 445 + // Phase 2: Wait for completion or timeout 446 + select { 447 + case <-p.Ctx.Done(): 448 + // Cleanup on timeout 449 + s.inFlightManager.RemovePromise(p) 450 + globalPromiseTrackers.mu.Lock() 451 + delete(globalPromiseTrackers.m, p.ID) 452 + globalPromiseTrackers.mu.Unlock() 453 + 454 + if p.Ctx.Err() == context.DeadlineExceeded { 455 + tracker.FinalizeAndRespond(p, "one of the query you asked didnt return in time, try again later") 456 + } else { 457 + tracker.FinalizeAndRespond(p, "query cancelled") 458 + } 459 + case <-tracker.Done: 460 + tracker.FinalizeAndRespond(p, "") 461 + } 462 + } 463 + 464 + func (s *LabelResolutionService) initialCacheCheckAndQueue(p *QueryPromise, tracker *PromiseTracker) { 465 + q := p.Query 466 + 467 + globalPromiseTrackers.mu.Lock() 468 + globalPromiseTrackers.m[p.ID] = tracker 469 + globalPromiseTrackers.mu.Unlock() 470 + 471 + requestsToSend := make(map[string][]string) // labelerDID -> subjects 472 + mapMu := sync.Mutex{} 473 + 474 + var wg sync.WaitGroup 475 + 476 + for _, subject := range q.Subjects { 477 + wg.Add(1) 478 + go func(subject string) { 479 + defer wg.Done() 480 + 481 + for _, labeler := range q.Labelers { 482 + cacheKey := subject + "|" + labeler 483 + 484 + value, found := s.lru.Get(cacheKey) 485 + log.Println("GET LRUKV " + cacheKey) 486 + labels := value.Labels 487 + 488 + if found { 489 + // Cache hit 490 + tracker.StoreResult(labeler, subject, labels, false) 491 + continue 492 + } 493 + 494 + // Cache miss 495 + alreadyInFlight := s.inFlightManager.CheckAndRegister(subject, labeler, p) 496 + 497 + if !alreadyInFlight { 498 + mapMu.Lock() 499 + requestsToSend[labeler] = append(requestsToSend[labeler], subject) 500 + mapMu.Unlock() 501 + } else { 502 + // Must decrement because no StoreResult will happen here 503 + tracker.TotalPairs-- 504 + } 505 + } 506 + }(subject) 507 + } 508 + wg.Wait() 509 + 510 + // 3. Send new requests to Bus Queue only for non-duplicate requests 511 + if len(requestsToSend) > 0 { 512 + subjectsToFetch := make(map[string][]string) // subject -> []labelers 513 + 514 + mapMu.Lock() 515 + for labeler, subjects := range requestsToSend { 516 + for _, subject := range subjects { 517 + subjectsToFetch[subject] = append(subjectsToFetch[subject], labeler) 518 + } 519 + } 520 + mapMu.Unlock() 521 + 522 + for subject, labelers := range subjectsToFetch { 523 + s.busQueueIn <- Req{Subject: subject, Labelers: labelers} 524 + } 525 + } 526 + } 527 + 528 + // --- 2. Bus Queue (Batching) --- 529 + 530 + func (s *LabelResolutionService) runBusQueue() { 531 + log.Println("bus queue wowww") 532 + defer s.wg.Done() 533 + 534 + // Batching pool: subject -> map[labelerDID]struct{} 535 + pool := make(map[string]map[string]struct{}) 536 + poolMu := sync.Mutex{} 537 + 538 + ticker := time.NewTicker(BusQueueBatchInterval) 539 + defer ticker.Stop() 540 + 541 + flush := func() { 542 + poolMu.Lock() 543 + defer poolMu.Unlock() 544 + 545 + if len(pool) == 0 { 546 + return 547 + } 548 + 549 + // Invert structure for Outsplitter: map[labelerDID][]subjectURI 550 + invertedRequests := make(map[string][]string) 551 + 552 + for subject, labelersMap := range pool { 553 + for labeler := range labelersMap { 554 + invertedRequests[labeler] = append(invertedRequests[labeler], subject) 555 + } 556 + } 557 + 558 + s.handleBatchedRequests(invertedRequests) 559 + pool = make(map[string]map[string]struct{}) 560 + } 561 + 562 + for { 563 + select { 564 + case req, ok := <-s.busQueueIn: 565 + if !ok { 566 + flush() 567 + return 568 + } 569 + 570 + poolMu.Lock() 571 + if _, exists := pool[req.Subject]; !exists { 572 + pool[req.Subject] = make(map[string]struct{}) 573 + } 574 + for _, labeler := range req.Labelers { 575 + pool[req.Subject][labeler] = struct{}{} 576 + } 577 + poolMu.Unlock() 578 + 579 + case <-ticker.C: 580 + flush() 581 + } 582 + } 583 + } 584 + 585 + // --- 3. Outsplitter and QueryLabels Handler --- 586 + 587 + type ExternalRequest struct { 588 + LabelerDID string 589 + Subjects []string 590 + } 591 + 592 + func (s *LabelResolutionService) handleBatchedRequests(invertedRequests map[string][]string) { 593 + var requests []ExternalRequest 594 + 595 + // Outsplitter logic: split large subject lists 596 + for labelerDID, subjects := range invertedRequests { 597 + for i := 0; i < len(subjects); i += MaxSubjectsPerExternalRequest { 598 + end := i + MaxSubjectsPerExternalRequest 599 + if end > len(subjects) { 600 + end = len(subjects) 601 + } 602 + requests = append(requests, ExternalRequest{LabelerDID: labelerDID, Subjects: subjects[i:end]}) 603 + } 604 + } 605 + 606 + var wg sync.WaitGroup 607 + for _, req := range requests { 608 + wg.Add(1) 609 + go func(r ExternalRequest) { 610 + defer wg.Done() 611 + s.executeExternalQuery(r) 612 + }(req) 613 + } 614 + wg.Wait() 615 + } 616 + 617 + func parseIntDefault(s string) (int64, error) { 618 + if s == "" { 619 + return 0, nil 620 + } 621 + return strconv.ParseInt(s, 10, 64) 622 + } 623 + 624 + func (s *LabelResolutionService) executeExternalQuery(r ExternalRequest) { 625 + log.Println("external query for " + r.LabelerDID + " " + "init") 626 + labelerDID := r.LabelerDID 627 + 628 + // 1. DID Resolution Check 629 + //domainEntry, err := s.labelerResolver.Resolve(labelerDID) 630 + did, err := syntax.ParseDID(labelerDID) 631 + if err != nil { 632 + s.handleExternalFailure(labelerDID, r.Subjects, "ResolutionFailure") // wrong did format 633 + return 634 + } 635 + ident, err := s.labelerResolver.LookupDID(context.TODO(), did) 636 + if err != nil { 637 + s.handleExternalFailure(labelerDID, r.Subjects, "ResolutionFailure") // unreachable 638 + return 639 + } 640 + log.Println("external query for " + r.LabelerDID + " " + "resolved the labeler did") 641 + 642 + labelerURL := ident.GetServiceEndpoint("atproto_labeler") 643 + if labelerURL == "" { 644 + s.handleExternalFailure(labelerDID, r.Subjects, "ResolutionFailure") // no service endpoint 645 + return 646 + } 647 + 648 + // domainEntry.Domain is the serviceEndpoint (e.g., https://mod.bsky.app) 649 + queryURL := fmt.Sprintf("%s/xrpc/com.atproto.label.queryLabels", strings.TrimSuffix(labelerURL, "/")) 650 + 651 + var allLabels []Label 652 + cursor := "" 653 + lastCursor := "-999" 654 + 655 + // 2. Query Labels Execution 656 + for page := 0; page < MaxQueryPages; page++ { 657 + 658 + if page > 0 && cursor == "" || cursor == "0" || lastCursor == cursor { 659 + break 660 + } 661 + 662 + params := url.Values{} 663 + if cursor != "" { 664 + params.Set("cursor", cursor) 665 + } 666 + for _, uri := range r.Subjects { 667 + params.Add("uriPatterns", uri) 668 + } 669 + 670 + respLabels, nextCursor, httpErr := s.queryLabels(queryURL, params) 671 + log.Println("external query for " + r.LabelerDID + " " + "querylabels at page " + strconv.Itoa(page) + " with cursor " + cursor + " and nextCursor " + nextCursor) 672 + 673 + if httpErr != nil { 674 + // Resolution was successful, but the subsequent query failed (network, 5xx, etc.) 675 + s.handleExternalFailure(labelerDID, r.Subjects, "QueryFailure") 676 + return 677 + } 678 + 679 + // Prevent null page cursor fetching (if we requested a page and got nothing) 680 + if page > 0 && len(respLabels) == 0 /*&& nextCursor == ""*/ { 681 + break 682 + } 683 + 684 + // prevent loops from faulty labelers 685 + cursorInt, errci := parseIntDefault(cursor) 686 + lastCursorInt, errlci := parseIntDefault(lastCursor) 687 + 688 + if errci != nil || errlci != nil || cursorInt <= lastCursorInt { 689 + break 690 + } 691 + 692 + allLabels = append(allLabels, respLabels...) 693 + lastCursor = cursor 694 + cursor = nextCursor 695 + 696 + if page == MaxQueryPages-1 && cursor != "" { 697 + s.handleExternalFailure(labelerDID, r.Subjects, "QueryLabelsTooManyPages") 698 + return 699 + } 700 + } 701 + 702 + log.Println("external query for " + r.LabelerDID + " " + "finished querylabels") 703 + s.handleExternalSuccess(labelerDID, r.Subjects, allLabels) 704 + } 705 + 706 + func (s *LabelResolutionService) handleExternalFailure(labelerDID string, subjects []string, failureType string) { 707 + for _, subject := range subjects { 708 + waiters := s.inFlightManager.Resolve(subject, labelerDID) 709 + 710 + for _, p := range waiters { 711 + tracker, found := globalPromiseTrackers.m[p.ID] 712 + if !found { 713 + continue 714 + } 715 + 716 + tracker.Mu.Lock() 717 + switch failureType { 718 + case "QueryLabelsTooManyPages": 719 + if !contains(tracker.Errors.QueryLabelsTooManyPages, labelerDID) { 720 + tracker.Errors.QueryLabelsTooManyPages = append(tracker.Errors.QueryLabelsTooManyPages, labelerDID) 721 + } 722 + case "ResolutionFailure": 723 + if !contains(tracker.Errors.LabelerResolutionFailure, labelerDID) { 724 + tracker.Errors.LabelerResolutionFailure = append(tracker.Errors.LabelerResolutionFailure, labelerDID) 725 + } 726 + case "QueryFailure": 727 + if !contains(tracker.Errors.LabelerQueryFailure, labelerDID) { 728 + tracker.Errors.LabelerQueryFailure = append(tracker.Errors.LabelerQueryFailure, labelerDID) 729 + } 730 + } 731 + tracker.Mu.Unlock() 732 + 733 + // Mark pair complete with empty result (failure reported in error map) 734 + tracker.StoreResult(labelerDID, subject, nil, true) 735 + } 736 + } 737 + } 738 + 739 + func (s *LabelResolutionService) handleExternalSuccess( 740 + labelerDID string, 741 + subjects []string, 742 + fetchedLabels []Label, 743 + ) { 744 + // Group labels by subject 745 + subjectLabels := make(map[string][]*Label) 746 + 747 + for i := range fetchedLabels { 748 + lbl := &fetchedLabels[i] 749 + subjectLabels[lbl.Uri] = append(subjectLabels[lbl.Uri], lbl) 750 + } 751 + 752 + for _, subject := range subjects { 753 + labels := subjectLabels[subject] 754 + 755 + resolved := ResolvedLabels{ 756 + Labels: labels, 757 + Seq: 67, // TODO: real seq tracking 758 + } 759 + 760 + cacheKey := subject + "|" + labelerDID 761 + s.lru.Put(cacheKey, resolved) 762 + log.Println("PUT LRUKV " + cacheKey) 763 + 764 + waiters := s.inFlightManager.Resolve(subject, labelerDID) 765 + 766 + for _, p := range waiters { 767 + tracker, found := globalPromiseTrackers.m[p.ID] 768 + if !found { 769 + continue 770 + } 771 + tracker.StoreResult(labelerDID, subject, labels, true) 772 + } 773 + } 774 + } 775 + 776 + func (s *LabelResolutionService) queryLabels(baseURL string, params url.Values) ([]Label, string, error) { 777 + // Build the full URL with query parameters 778 + fullURL := baseURL + "?" + params.Encode() 779 + 780 + // Create HTTP request with timeout 781 + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) 782 + defer cancel() 783 + 784 + req, err := http.NewRequestWithContext(ctx, "GET", fullURL, nil) 785 + if err != nil { 786 + return nil, "", err 787 + } 788 + 789 + resp, err := http.DefaultClient.Do(req) 790 + if err != nil { 791 + return nil, "", err 792 + } 793 + defer resp.Body.Close() 794 + 795 + if resp.StatusCode != http.StatusOK { 796 + return nil, "", fmt.Errorf("HTTP error: %d", resp.StatusCode) 797 + } 798 + 799 + // Parse the response 800 + var result struct { 801 + Labels []Label `json:"labels"` 802 + Cursor string `json:"cursor,omitempty"` 803 + } 804 + 805 + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { 806 + return nil, "", err 807 + } 808 + 809 + return result.Labels, result.Cursor, nil 810 + } 811 + 812 + func contains(s []string, e string) bool { 813 + for _, a := range s { 814 + if a == e { 815 + return true 816 + } 817 + } 818 + return false 819 + } 820 + 821 + // --- HTTP Server and Routing --- 822 + type Server struct { 823 + Service *LabelResolutionService 824 + } 825 + 826 + func (s *Server) root(c echo.Context) error { 827 + c.String(http.StatusOK, ` ____ __________ ____ _ _____ ____ ______ 828 + / __ \/ ____/ __ \ / __ \ | / / | / __ \/ ____/ 829 + / /_/ / __/ / / / / / / / / | /| / / /| | / /_/ / /_ 830 + / _, _/ /___/ /_/ / / /_/ /| |/ |/ / ___ |/ _, _/ __/ 831 + /_/ |_/_____/_____/ /_____/ |__/|__/_/ |_/_/ |_/_/ 832 + __ __ __ 833 + / /___ _/ /_ ___ / /___ ___ ___ _________ ____ 834 + / / __ '/ __ \/ _ \/ / __ '__ \/ _ \/ ___/ __ '/ _ \ 835 + / / /_/ / /_/ / __/ / / / / / / __/ / / /_/ / __/ 836 + /_/\__,_/_.___/\___/_/_/ /_/ /_/\___/_/ \__, /\___/ 837 + /____/ 838 + 839 + This is an AT Protocol cache for querying labels 840 + 841 + The only API route is /xrpc/app.reddwarf.labelmerge.queryLabels 842 + 843 + Code: https://tangled.org/whey.party/red-dwarf-server 844 + Protocol: https://atproto.com 845 + Try it on: https://reddwarf.app 846 + `) 847 + return nil 848 + } 849 + 850 + func (s *Server) queryLabels(c *gin.Context) { 851 + log.Println("NEW REQUEST FOR uhhh a bunch of stuff") 852 + 853 + var ( 854 + query IncomingQuery 855 + err error 856 + ) 857 + 858 + switch c.Request.Method { 859 + case http.MethodGet: 860 + query, err = s.parseGETQuery(c) 861 + case http.MethodPost: 862 + query, err = s.parsePOSTQuery(c) 863 + default: 864 + c.AbortWithStatus(http.StatusMethodNotAllowed) 865 + return 866 + } 867 + 868 + if err != nil { 869 + c.String(http.StatusBadRequest, "Invalid query parameters: %v", err) 870 + return 871 + } 872 + 873 + if len(query.Labelers) > MaxLabelersPerQuery || len(query.Subjects) > MaxSubjectsPerQuery { 874 + c.String( 875 + http.StatusRequestEntityTooLarge, 876 + "Query exceeds limits (Subjects: %d, Labelers: %d)", 877 + MaxSubjectsPerQuery, 878 + MaxLabelersPerQuery, 879 + ) 880 + return 881 + } 882 + 883 + query.Strict = c.Query("strict") == "true" || query.Strict 884 + 885 + // Setup context with timeout 886 + ctx, cancel := context.WithTimeout(c.Request.Context(), IncomingQueryTimeout) 887 + defer cancel() 888 + 889 + promise := &QueryPromise{ 890 + ID: strconv.FormatInt(time.Now().UnixNano(), 10), 891 + Query: query, 892 + Response: make(chan FinalResponse, 1), 893 + Ctx: ctx, 894 + } 895 + 896 + s.Service.HandlePromise(promise) 897 + 898 + select { 899 + case result := <-promise.Response: 900 + // strict mode error handling 901 + isErrorResponse := 902 + result.Error != nil && 903 + (len(result.Error.LabelerResolutionFailure) > 0 || 904 + len(result.Error.QueryLabelsTooManyPages) > 0 || 905 + len(result.Error.LabelerQueryFailure) > 0) 906 + 907 + if query.Strict && isErrorResponse { 908 + c.String(http.StatusInternalServerError, "Strict failure encountered: %v", result.Error) 909 + return 910 + } 911 + 912 + c.JSON(http.StatusOK, result) 913 + 914 + case <-ctx.Done(): 915 + timeoutMsg := "one of the query you asked didnt return in time, try again later" 916 + 917 + if query.Strict { 918 + c.String(http.StatusRequestTimeout, timeoutMsg) 919 + return 920 + } 921 + 922 + c.JSON(http.StatusOK, FinalResponse{ 923 + Error: &QueryError{ 924 + LabelerResolutionFailure: []string{timeoutMsg}, 925 + }, 926 + }) 927 + } 928 + } 929 + 930 + func (s *Server) parseGETQuery(c *gin.Context) (IncomingQuery, error) { 931 + subjects := c.QueryArray("s") 932 + labelers := c.QueryArray("l") 933 + 934 + if len(subjects) == 0 || len(labelers) == 0 { 935 + return IncomingQuery{}, fmt.Errorf("must specify subjects (s) and labelers (l)") 936 + } 937 + 938 + return IncomingQuery{ 939 + Subjects: subjects, 940 + Labelers: labelers, 941 + }, nil 942 + } 943 + 944 + func (s *Server) parsePOSTQuery(c *gin.Context) (IncomingQuery, error) { 945 + var query IncomingQuery 946 + if err := c.ShouldBindJSON(&query); err != nil { 947 + return IncomingQuery{}, err 948 + } 949 + 950 + if len(query.Subjects) == 0 || len(query.Labelers) == 0 { 951 + return IncomingQuery{}, fmt.Errorf("must specify subjects (s) and labelers (l)") 952 + } 953 + 954 + return query, nil 955 + } 956 + 957 + func main() { 958 + log.SetFlags(log.LstdFlags | log.Lshortfile) 959 + 960 + service := NewLabelResolutionService() 961 + defer service.Stop() 962 + 963 + server := &Server{Service: service} 964 + 965 + e := echo.New() 966 + e.Use(middleware.Recover()) 967 + 968 + // your own routes 969 + e.GET("/", server.root) 970 + 971 + // register lexgen-generated handlers 972 + if err := server.RegisterHandlersAppReddwarfLabelmerge(e); err != nil { 973 + log.Fatalf("failed to register handlers: %v", err) 974 + } 975 + 976 + log.Println("Label Resolution Service running on", ServerPort) 977 + if err := e.Start(ServerPort); err != nil { 978 + log.Fatalf("Server failed: %v", err) 979 + } 980 + }

+44

cmd/labelmerge/stubs.go

··· 1 + package main 2 + 3 + import ( 4 + "strconv" 5 + 6 + "github.com/labstack/echo/v4" 7 + "go.opentelemetry.io/otel" 8 + appreddwarflabelmerge "tangled.org/whey.party/red-dwarf-server/labelmerge/lex" 9 + ) 10 + 11 + func (s *Server) RegisterHandlersAppReddwarfLabelmerge(e *echo.Echo) error { 12 + e.GET("/xrpc/app.reddwarf.labelmerge.queryLabels", s.HandleAppReddwarfLabelmergeQueryLabels) 13 + return nil 14 + } 15 + 16 + func (s *Server) HandleAppReddwarfLabelmergeQueryLabels(c echo.Context) error { 17 + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandleAppReddwarfLabelmergeQueryLabels") 18 + defer span.End() 19 + 20 + inputl := c.QueryParams()["l"] 21 + 22 + inputs := c.QueryParams()["s"] 23 + 24 + var strict *bool 25 + if p := c.QueryParam("strict"); p != "" { 26 + strict_val, err := strconv.ParseBool(p) 27 + if err != nil { 28 + return err 29 + } 30 + strict = &strict_val 31 + } 32 + var out *appreddwarflabelmerge.QueryLabels_Output 33 + var handleErr error 34 + // func (s *Server) handleAppReddwarfLabelmergeQueryLabels(ctx context.Context,l []string,s []string,strict *bool) (*appreddwarflabelmerge.QueryLabels_Output, error) 35 + out, handleErr = s.handleAppReddwarfLabelmergeQueryLabels(ctx, inputl, inputs, strict) 36 + if handleErr != nil { 37 + return handleErr 38 + } 39 + return c.JSON(200, out) 40 + } 41 + 42 + func (s *Server) RegisterHandlersComAtproto(e *echo.Echo) error { 43 + return nil 44 + }

+64

cmd/test/main.go

··· 1 + package main 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "log/slog" 7 + "os" 8 + "os/signal" 9 + "syscall" 10 + "time" 11 + 12 + labelstream "tangled.org/whey.party/red-dwarf-server/labelmerge/stream" 13 + ) 14 + 15 + func main() { 16 + log := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{ 17 + Level: slog.LevelInfo, 18 + })) 19 + 20 + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) 21 + defer stop() 22 + 23 + manager := labelstream.NewLabelerSubscriptionManager(log) 24 + 25 + modLabelerDID := "did:plc:ar7c4by46qjdydhdevvrndac" 26 + 27 + if err := manager.AddLabeler(modLabelerDID, "1412186"); err != nil { 28 + log.Error("Failed to add labeler", "err", err) 29 + os.Exit(1) 30 + } 31 + 32 + manager.Start() 33 + 34 + log.Info("Starting event consumption loop. Press Ctrl+C to stop.") 35 + 36 + for { 37 + select { 38 + case event, ok := <-manager.Events(): 39 + if !ok { 40 + log.Info("Event channel closed, exiting consumer.") 41 + return 42 + } 43 + 44 + action := "APPLICATION" 45 + if event.Value.Neg != nil && *event.Value.Neg { 46 + action = "NEGATION" 47 + } 48 + 49 + fmt.Printf("\n[EVENT FROM %s] Cursor: %d, Action: %s, Label: %s, URI: %s\n", 50 + event.SourceDid, 51 + event.Cursor, 52 + action, 53 + event.Value.Val, 54 + event.Value.Uri, 55 + ) 56 + 57 + case <-ctx.Done(): 58 + log.Info("Shutdown signal received.") 59 + manager.Stop() 60 + time.Sleep(1 * time.Second) 61 + return 62 + } 63 + } 64 + }

+12 -1

go.mod

··· 3 3 go 1.25.4 4 4 5 5 require ( 6 - github.com/bluesky-social/indigo v0.0.0-20251206005924-d49b45419635 6 + github.com/bluesky-social/indigo v0.0.0-20260128065308-121c189aef50 7 7 github.com/ericvolp12/jwt-go-secp256k1 v0.0.2 8 8 github.com/gin-contrib/cors v1.7.6 9 9 github.com/gin-gonic/gin v1.11.0 ··· 12 12 github.com/gorilla/websocket v1.5.3 13 13 github.com/hashicorp/golang-lru/arc/v2 v2.0.7 14 14 github.com/klauspost/compress v1.18.2 15 + github.com/labstack/echo/v4 v4.11.3 15 16 github.com/prometheus/client_golang v1.23.2 16 17 gitlab.com/yawning/secp256k1-voi v0.0.0-20230925100816-f2616030848b 17 18 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 ··· 20 21 ) 21 22 22 23 require ( 24 + github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b // indirect 23 25 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect 24 26 github.com/dgraph-io/ristretto/v2 v2.2.0 // indirect 25 27 github.com/dustin/go-humanize v1.0.1 // indirect 26 28 github.com/gogo/protobuf v1.3.2 // indirect 27 29 github.com/google/flatbuffers v25.2.10+incompatible // indirect 30 + github.com/hashicorp/go-cleanhttp v0.5.2 // indirect 31 + github.com/hashicorp/go-retryablehttp v0.7.5 // indirect 28 32 github.com/hashicorp/golang-lru v1.0.2 // indirect 29 33 github.com/ipfs/bbloom v0.0.4 // indirect 30 34 github.com/ipfs/go-block-format v0.2.0 // indirect ··· 46 50 github.com/ipld/go-codec-dagpb v1.6.0 // indirect 47 51 github.com/ipld/go-ipld-prime v0.21.0 // indirect 48 52 github.com/jbenet/goprocess v0.1.4 // indirect 53 + github.com/jinzhu/inflection v1.0.0 // indirect 54 + github.com/jinzhu/now v1.1.5 // indirect 55 + github.com/labstack/gommon v0.4.1 // indirect 49 56 github.com/lestrrat-go/blackmagic v1.0.1 // indirect 50 57 github.com/lestrrat-go/httpcc v1.0.1 // indirect 51 58 github.com/lestrrat-go/httprc v1.0.4 // indirect 52 59 github.com/lestrrat-go/iter v1.0.2 // indirect 53 60 github.com/lestrrat-go/jwx/v2 v2.0.12 // indirect 54 61 github.com/lestrrat-go/option v1.0.1 // indirect 62 + github.com/mattn/go-colorable v0.1.13 // indirect 55 63 github.com/opentracing/opentracing-go v1.2.0 // indirect 56 64 github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f // indirect 57 65 github.com/segmentio/asm v1.2.0 // indirect 66 + github.com/valyala/bytebufferpool v1.0.0 // indirect 67 + github.com/valyala/fasttemplate v1.2.2 // indirect 58 68 go.uber.org/atomic v1.11.0 // indirect 59 69 go.uber.org/multierr v1.11.0 // indirect 60 70 go.uber.org/zap v1.26.0 // indirect 71 + gorm.io/gorm v1.25.9 // indirect 61 72 ) 62 73 63 74 require (

+33 -6

go.sum

··· 1 1 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= 2 + github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b h1:5/++qT1/z812ZqBvqQt6ToRswSuPZ/B33m6xVHRzADU= 3 + github.com/RussellLuo/slidingwindow v0.0.0-20200528002341-535bb99d338b/go.mod h1:4+EPqMRApwwE/6yo6CxiHoSnBzjRr3jsqer7frxP8y4= 2 4 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= 3 5 github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A= 4 6 github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= 5 7 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= 6 8 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= 7 - github.com/bluesky-social/indigo v0.0.0-20251202051123-81f317e322bc h1:2t+uAvfzJiCsTMwn5fW85t/IGa0+2I7BXS2ORastK4o= 8 - github.com/bluesky-social/indigo v0.0.0-20251202051123-81f317e322bc/go.mod h1:Pm2I1+iDXn/hLbF7XCg/DsZi6uDCiOo7hZGWprSM7k0= 9 - github.com/bluesky-social/indigo v0.0.0-20251206005924-d49b45419635 h1:kNeRrgGJH2g5OvjLqtaQ744YXqduliZYpFkJ/ld47c0= 10 - github.com/bluesky-social/indigo v0.0.0-20251206005924-d49b45419635/go.mod h1:Pm2I1+iDXn/hLbF7XCg/DsZi6uDCiOo7hZGWprSM7k0= 9 + github.com/bluesky-social/indigo v0.0.0-20260128065308-121c189aef50 h1:rpGmYvqqniy0gn0WJc4tZtlvPRrY5fVEwpV3WgFtdRk= 10 + github.com/bluesky-social/indigo v0.0.0-20260128065308-121c189aef50/go.mod h1:KIy0FgNQacp4uv2Z7xhNkV3qZiUSGuRky97s7Pa4v+o= 11 11 github.com/bluesky-social/jetstream v0.0.0-20251009222037-7d7efa58d7f1 h1:ovcRKN1iXZnY5WApVg+0Hw2RkwMH0ziA7lSAA8vellU= 12 12 github.com/bluesky-social/jetstream v0.0.0-20251009222037-7d7efa58d7f1/go.mod h1:5PtGi4r/PjEVBBl+0xWuQn4mBEjr9h6xsfDBADS6cHs= 13 13 github.com/bytedance/sonic v1.14.0 h1:/OfKt8HFw0kh2rj8N0F6C/qPGRESq0BbaNZgcNXXzQQ= ··· 32 32 github.com/dgraph-io/badger/v4 v4.8.0/go.mod h1:U6on6e8k/RTbUWxqKR0MvugJuVmkxSNc79ap4917h4w= 33 33 github.com/dgraph-io/ristretto/v2 v2.2.0 h1:bkY3XzJcXoMuELV8F+vS8kzNgicwQFAaGINAEJdWGOM= 34 34 github.com/dgraph-io/ristretto/v2 v2.2.0/go.mod h1:RZrm63UmcBAaYWC1DotLYBmTvgkrs0+XhBd7Npn7/zI= 35 + github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da h1:aIftn67I1fkbMa512G+w+Pxci9hJPB8oMnkcP3iZF38= 36 + github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw= 35 37 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= 36 38 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= 37 39 github.com/earthboundkid/versioninfo/v2 v2.24.1 h1:SJTMHaoUx3GzjjnUO1QzP3ZXK6Ee/nbWyCm58eY3oUg= ··· 63 65 github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY= 64 66 github.com/go-playground/validator/v10 v10.27.0 h1:w8+XrWVMhGkxOaaowyKH35gFydVHOvC0/uWoy2Fzwn4= 65 67 github.com/go-playground/validator/v10 v10.27.0/go.mod h1:I5QpIEbmr8On7W0TktmJAumgzX4CA1XNl4ZmDuVHKKo= 68 + github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg= 69 + github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA= 66 70 github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0= 67 71 github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I= 68 72 github.com/goccy/go-json v0.10.5 h1:Fq85nIqj+gXn/S5ahsiTlK3TmC85qgirsdTP/+DeaC4= ··· 87 91 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= 88 92 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= 89 93 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= 94 + github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ= 95 + github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48= 96 + github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI= 97 + github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ= 98 + github.com/hashicorp/go-retryablehttp v0.7.5 h1:bJj+Pj19UZMIweq/iie+1u5YCdGrnxCT9yvm0e+Nd5M= 99 + github.com/hashicorp/go-retryablehttp v0.7.5/go.mod h1:Jy/gPYAdjqffZ/yFGCFV2doI5wjtH1ewM9u8iYVjtX8= 90 100 github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= 91 101 github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= 92 102 github.com/hashicorp/golang-lru/arc/v2 v2.0.7 h1:QxkVTxwColcduO+LP7eJO56r2hFiG8zEbfAAzRv52KQ= ··· 146 156 github.com/ipfs/go-peertaskqueue v0.8.0/go.mod h1:cz8hEnnARq4Du5TGqiWKgMr/BOSQ5XOgMOh1K5YYKKM= 147 157 github.com/ipfs/go-verifcid v0.0.3 h1:gmRKccqhWDocCRkC+a59g5QW7uJw5bpX9HWBevXa0zs= 148 158 github.com/ipfs/go-verifcid v0.0.3/go.mod h1:gcCtGniVzelKrbk9ooUSX/pM3xlH73fZZJDzQJRvOUw= 149 - github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4 h1:oFo19cBmcP0Cmg3XXbrr0V/c+xU9U1huEZp8+OgBzdI= 150 - github.com/ipld/go-car v0.6.1-0.20230509095817-92d28eb23ba4/go.mod h1:6nkFF8OmR5wLKBzRKi7/YFJpyYR7+oEn1DX+mMWnlLA= 151 159 github.com/ipld/go-car v0.6.2 h1:Hlnl3Awgnq8icK+ze3iRghk805lu8YNq3wlREDTF2qc= 152 160 github.com/ipld/go-car v0.6.2/go.mod h1:oEGXdwp6bmxJCZ+rARSkDliTeYnVzv3++eXajZ+Bmr8= 153 161 github.com/ipld/go-codec-dagpb v1.6.0 h1:9nYazfyu9B1p3NAgfVdpRco3Fs2nFC72DqVsMj6rOcc= ··· 159 167 github.com/jbenet/go-cienv v0.1.0/go.mod h1:TqNnHUmJgXau0nCzC7kXWeotg3J9W34CUv5Djy1+FlA= 160 168 github.com/jbenet/goprocess v0.1.4 h1:DRGOFReOMqqDNXwW70QkacFW0YN9QnwLV0Vqk+3oU0o= 161 169 github.com/jbenet/goprocess v0.1.4/go.mod h1:5yspPrukOVuOLORacaBi858NqyClJPQxYZlqdZVfqY4= 170 + github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= 171 + github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= 172 + github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= 173 + github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= 162 174 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= 163 175 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= 164 176 github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= ··· 178 190 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= 179 191 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= 180 192 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= 193 + github.com/labstack/echo/v4 v4.11.3 h1:Upyu3olaqSHkCjs1EJJwQ3WId8b8b1hxbogyommKktM= 194 + github.com/labstack/echo/v4 v4.11.3/go.mod h1:UcGuQ8V6ZNRmSweBIJkPvGfwCMIlFmiqrPqiEBfPYws= 195 + github.com/labstack/gommon v0.4.1 h1:gqEff0p/hTENGMABzezPoPSRtIh1Cvw0ueMOe0/dfOk= 196 + github.com/labstack/gommon v0.4.1/go.mod h1:TyTrpPqxR5KMk8LKVtLmfMjeQ5FEkBYdxLYPw/WfrOM= 181 197 github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= 182 198 github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= 183 199 github.com/lestrrat-go/blackmagic v1.0.1 h1:lS5Zts+5HIC/8og6cGHb0uCcNCa3OUt1ygh3Qz2Fe80= ··· 213 229 github.com/libp2p/go-netroute v0.2.0/go.mod h1:Vio7LTzZ+6hoT4CMZi5/6CpY3Snzh2vgZhWgxMNwlQI= 214 230 github.com/libp2p/go-openssl v0.1.0 h1:LBkKEcUv6vtZIQLVTegAil8jbNpJErQ9AnT+bWV+Ooo= 215 231 github.com/libp2p/go-openssl v0.1.0/go.mod h1:OiOxwPpL3n4xlenjx2h7AwSGaFSC/KZvf6gNdOBQMtc= 232 + github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= 233 + github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= 216 234 github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94= 235 + github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= 217 236 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= 218 237 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= 219 238 github.com/mattn/go-pointer v0.0.1 h1:n+XhsuGeVO6MEAp7xyEukFINEa+Quek5psIR/ylA6o0= ··· 290 309 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= 291 310 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= 292 311 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= 312 + github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= 293 313 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= 294 314 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= 295 315 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= ··· 305 325 github.com/ugorji/go/codec v1.3.0 h1:Qd2W2sQawAfG8XSvzwhBeoGq71zXOC/Q1E9y/wUcsUA= 306 326 github.com/ugorji/go/codec v1.3.0/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4= 307 327 github.com/urfave/cli v1.22.10/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0= 328 + github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= 329 + github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= 330 + github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo= 331 + github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ= 308 332 github.com/warpfork/go-testmark v0.12.1 h1:rMgCpJfwy1sJ50x0M0NgyphxYYPMOODIJHhsXyEHU0s= 309 333 github.com/warpfork/go-testmark v0.12.1/go.mod h1:kHwy7wfvGSPh1rQJYKayD4AbtNaeyZdcGi9tNJTaa5Y= 310 334 github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0 h1:GDDkbFiaK8jsSDJfjId/PEGEShv6ugrt4kYsC5UIDaQ= ··· 405 429 golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 406 430 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 407 431 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 432 + golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 408 433 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 409 434 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= 410 435 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= ··· 461 486 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= 462 487 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= 463 488 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= 489 + gorm.io/gorm v1.25.9 h1:wct0gxZIELDk8+ZqF/MVnHLkA1rvYlBWUMv2EdsK1g8= 490 + gorm.io/gorm v1.25.9/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8= 464 491 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg= 465 492 lukechampine.com/blake3 v1.4.1 h1:I3Smz7gso8w4/TunLKec6K2fn+kyKtDxr/xcQEN84Wg= 466 493 lukechampine.com/blake3 v1.4.1/go.mod h1:QFosUxmjB8mnrWFSNwKmvxHpfY72bmD2tQ0kBMM3kwo=

+78

labelmerge/lex/generation/defs/app.reddwarf.labelmerge.queryLabels.json

··· 1 + { 2 + "lexicon": 1, 3 + "id": "app.reddwarf.labelmerge.queryLabels", 4 + "defs": { 5 + "main": { 6 + "type": "query", 7 + "description": "Find labels relevant to the provided AT-URI patterns. Public endpoint for moderation services, though may return different or additional results with auth.", 8 + "parameters": { 9 + "type": "params", 10 + "properties": { 11 + "s": { 12 + "type": "array", 13 + "items": { 14 + "type": "string" 15 + }, 16 + "description": "List of label subjects (strings)." 17 + }, 18 + "l": { 19 + "type": "array", 20 + "items": { 21 + "type": "string", 22 + "format": "did" 23 + }, 24 + "description": "List of label sources (labeler DIDs) to filter on." 25 + }, 26 + "strict": { 27 + "type": "boolean", 28 + "description": "If true then any errors will throw the entire query" 29 + } 30 + }, 31 + "required": [ 32 + "s", 33 + "l" 34 + ] 35 + }, 36 + "output": { 37 + "encoding": "application/json", 38 + "schema": { 39 + "type": "object", 40 + "properties": { 41 + "labels": { 42 + "type": "array", 43 + "items": { 44 + "type": "ref", 45 + "ref": "com.atproto.label.defs#label" 46 + } 47 + }, 48 + "error": { 49 + "type": "array", 50 + "items": { 51 + "type": "ref", 52 + "ref": "#error" 53 + } 54 + } 55 + }, 56 + "required": [ 57 + "labels" 58 + ] 59 + } 60 + } 61 + }, 62 + "error": { 63 + "type": "object", 64 + "properties": { 65 + "s": { 66 + "type": "string", 67 + "format": "did" 68 + }, 69 + "e": { 70 + "type": "string" 71 + } 72 + }, 73 + "required": [ 74 + "s" 75 + ] 76 + } 77 + } 78 + }

+193

labelmerge/lex/generation/external/com.atproto.label.defs.json

··· 1 + { 2 + "id": "com.atproto.label.defs", 3 + "defs": { 4 + "label": { 5 + "type": "object", 6 + "required": [ 7 + "src", 8 + "uri", 9 + "val", 10 + "cts" 11 + ], 12 + "properties": { 13 + "cid": { 14 + "type": "string", 15 + "format": "cid", 16 + "description": "Optionally, CID specifying the specific version of 'uri' resource this label applies to." 17 + }, 18 + "cts": { 19 + "type": "string", 20 + "format": "datetime", 21 + "description": "Timestamp when this label was created." 22 + }, 23 + "exp": { 24 + "type": "string", 25 + "format": "datetime", 26 + "description": "Timestamp at which this label expires (no longer applies)." 27 + }, 28 + "neg": { 29 + "type": "boolean", 30 + "description": "If true, this is a negation label, overwriting a previous label." 31 + }, 32 + "sig": { 33 + "type": "bytes", 34 + "description": "Signature of dag-cbor encoded label." 35 + }, 36 + "src": { 37 + "type": "string", 38 + "format": "did", 39 + "description": "DID of the actor who created this label." 40 + }, 41 + "uri": { 42 + "type": "string", 43 + "format": "uri", 44 + "description": "AT URI of the record, repository (account), or other resource that this label applies to." 45 + }, 46 + "val": { 47 + "type": "string", 48 + "maxLength": 128, 49 + "description": "The short string name of the value or type of this label." 50 + }, 51 + "ver": { 52 + "type": "integer", 53 + "description": "The AT Protocol version of the label object." 54 + } 55 + }, 56 + "description": "Metadata tag on an atproto resource (eg, repo or record)." 57 + }, 58 + "selfLabel": { 59 + "type": "object", 60 + "required": [ 61 + "val" 62 + ], 63 + "properties": { 64 + "val": { 65 + "type": "string", 66 + "maxLength": 128, 67 + "description": "The short string name of the value or type of this label." 68 + } 69 + }, 70 + "description": "Metadata tag on an atproto record, published by the author within the record. Note that schemas should use #selfLabels, not #selfLabel." 71 + }, 72 + "labelValue": { 73 + "type": "string", 74 + "knownValues": [ 75 + "!hide", 76 + "!no-promote", 77 + "!warn", 78 + "!no-unauthenticated", 79 + "dmca-violation", 80 + "doxxing", 81 + "porn", 82 + "sexual", 83 + "nudity", 84 + "nsfl", 85 + "gore" 86 + ] 87 + }, 88 + "selfLabels": { 89 + "type": "object", 90 + "required": [ 91 + "values" 92 + ], 93 + "properties": { 94 + "values": { 95 + "type": "array", 96 + "items": { 97 + "ref": "#selfLabel", 98 + "type": "ref" 99 + }, 100 + "maxLength": 10 101 + } 102 + }, 103 + "description": "Metadata tags on an atproto record, published by the author within the record." 104 + }, 105 + "labelValueDefinition": { 106 + "type": "object", 107 + "required": [ 108 + "identifier", 109 + "severity", 110 + "blurs", 111 + "locales" 112 + ], 113 + "properties": { 114 + "blurs": { 115 + "type": "string", 116 + "description": "What should this label hide in the UI, if applied? 'content' hides all of the target; 'media' hides the images/video/audio; 'none' hides nothing.", 117 + "knownValues": [ 118 + "content", 119 + "media", 120 + "none" 121 + ] 122 + }, 123 + "locales": { 124 + "type": "array", 125 + "items": { 126 + "ref": "#labelValueDefinitionStrings", 127 + "type": "ref" 128 + } 129 + }, 130 + "severity": { 131 + "type": "string", 132 + "description": "How should a client visually convey this label? 'inform' means neutral and informational; 'alert' means negative and warning; 'none' means show nothing.", 133 + "knownValues": [ 134 + "inform", 135 + "alert", 136 + "none" 137 + ] 138 + }, 139 + "adultOnly": { 140 + "type": "boolean", 141 + "description": "Does the user need to have adult content enabled in order to configure this label?" 142 + }, 143 + "identifier": { 144 + "type": "string", 145 + "maxLength": 100, 146 + "description": "The value of the label being defined. Must only include lowercase ascii and the '-' character ([a-z-]+).", 147 + "maxGraphemes": 100 148 + }, 149 + "defaultSetting": { 150 + "type": "string", 151 + "default": "warn", 152 + "description": "The default setting for this label.", 153 + "knownValues": [ 154 + "ignore", 155 + "warn", 156 + "hide" 157 + ] 158 + } 159 + }, 160 + "description": "Declares a label value and its expected interpretations and behaviors." 161 + }, 162 + "labelValueDefinitionStrings": { 163 + "type": "object", 164 + "required": [ 165 + "lang", 166 + "name", 167 + "description" 168 + ], 169 + "properties": { 170 + "lang": { 171 + "type": "string", 172 + "format": "language", 173 + "description": "The code of the language these strings are written in." 174 + }, 175 + "name": { 176 + "type": "string", 177 + "maxLength": 640, 178 + "description": "A short human-readable name for the label.", 179 + "maxGraphemes": 64 180 + }, 181 + "description": { 182 + "type": "string", 183 + "maxLength": 100000, 184 + "description": "A longer description of what the label means and why it might be applied.", 185 + "maxGraphemes": 10000 186 + } 187 + }, 188 + "description": "Strings which describe the label in the UI, localized into a specific language." 189 + } 190 + }, 191 + "$type": "com.atproto.lexicon.schema", 192 + "lexicon": 1 193 + }

+32

labelmerge/lex/generation/lexgen.json

··· 1 + [ 2 + { 3 + "package": "bsky", 4 + "prefix": "app.bsky", 5 + "outdir": "api/bsky", 6 + "import": "github.com/bluesky-social/indigo/api/bsky" 7 + }, 8 + { 9 + "package": "atproto", 10 + "prefix": "com.atproto", 11 + "outdir": "api/atproto", 12 + "import": "github.com/bluesky-social/indigo/api/atproto" 13 + }, 14 + { 15 + "package": "chat", 16 + "prefix": "chat.bsky", 17 + "outdir": "api/chat", 18 + "import": "github.com/bluesky-social/indigo/api/chat" 19 + }, 20 + { 21 + "package": "ozone", 22 + "prefix": "tools.ozone", 23 + "outdir": "api/ozone", 24 + "import": "github.com/bluesky-social/indigo/api/ozone" 25 + }, 26 + { 27 + "package": "labelmerge", 28 + "prefix": "app.reddwarf.labelmerge", 29 + "outdir": "labelmerge/lex", 30 + "import": "tangled.org/whey.party/red-dwarf-server/labelmerge/lex" 31 + } 32 + ]

+39

labelmerge/lex/generation/readme.md

··· 1 + # USING CODEGEN 2 + 3 + i dont really understand how indigo lexgen works but this two works i guess 4 + 5 + run this in the project root 6 + 7 + ### struct gen 8 + you really only need to do this when lexicon changes 9 + ```bash 10 + go run github.com/bluesky-social/indigo/cmd/lexgen/ \ 11 + --package labelmerge_lex --outdir ./labelmerge/lex \ 12 + --external-lexicons ./labelmerge/lex/generation/external/com.atproto.label.defs.json \ 13 + --build-file ./labelmerge/lex/generation/lexgen.json \ 14 + ./labelmerge/lex/generation/defs 15 + ``` 16 + 17 + ### server gen 18 + this is really only needed to be done once per project and never again. 19 + 20 + written here for future reference. 21 + ```bash 22 + go run github.com/bluesky-social/indigo/cmd/lexgen/ \ 23 + --gen-server \ 24 + --gen-handlers \ 25 + --package main \ 26 + --outdir ./cmd/labelmerge/ \ 27 + --external-lexicons ./labelmerge/lex/generation/external/ \ 28 + --types-import app.reddwarf.labelmerge:tangled.org/whey.party/red-dwarf-server/labelmerge/lex \ 29 + --types-import com.atproto:github.com/bluesky-social/indigo/api/atproto \ 30 + --build-file ./labelmerge/lex/generation/lexgen.json \ 31 + ./labelmerge/lex/defs/ 32 + ``` 33 + 34 + ### typescript client gen 35 + 36 + uses a separate tool but its kinda related so im gonna put it here too 37 + ```bash 38 + npx @atproto/lex-cli gen-api ./your/typescript/project/path ./labelmerge/lex/generation/defs/app.reddwarf.labelmerge.queryLabels.json ./labelmerge/lex/generation/external/com.atproto.label.defs.json 39 + ```

+45

labelmerge/lex/labelmergequeryLabels.go

··· 1 + // Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. 2 + 3 + // Lexicon schema: app.reddwarf.labelmerge.queryLabels 4 + 5 + package labelmerge_lex 6 + 7 + import ( 8 + "context" 9 + 10 + comatproto "github.com/bluesky-social/indigo/api/atproto" 11 + lexutil "github.com/bluesky-social/indigo/lex/util" 12 + ) 13 + 14 + // QueryLabels_Error is a "error" in the app.reddwarf.labelmerge.queryLabels schema. 15 + type QueryLabels_Error struct { 16 + E *string `json:"e,omitempty" cborgen:"e,omitempty"` 17 + S string `json:"s" cborgen:"s"` 18 + } 19 + 20 + // QueryLabels_Output is the output of a app.reddwarf.labelmerge.queryLabels call. 21 + type QueryLabels_Output struct { 22 + Error []*QueryLabels_Error `json:"error,omitempty" cborgen:"error,omitempty"` 23 + Labels []*comatproto.LabelDefs_Label `json:"labels" cborgen:"labels"` 24 + } 25 + 26 + // QueryLabels calls the XRPC method "app.reddwarf.labelmerge.queryLabels". 27 + // 28 + // l: List of label sources (labeler DIDs) to filter on. 29 + // s: List of label subjects (strings). 30 + // strict: If true then any errors will throw the entire query 31 + func QueryLabels(ctx context.Context, c lexutil.LexClient, l []string, s []string, strict bool) (*QueryLabels_Output, error) { 32 + var out QueryLabels_Output 33 + 34 + params := map[string]interface{}{} 35 + params["l"] = l 36 + params["s"] = s 37 + if strict { 38 + params["strict"] = strict 39 + } 40 + if err := c.LexDo(ctx, lexutil.Query, "", "app.reddwarf.labelmerge.queryLabels", params, nil, &out); err != nil { 41 + return nil, err 42 + } 43 + 44 + return &out, nil 45 + }

+221

labelmerge/lru/lru.go

··· 1 + package lru 2 + 3 + import ( 4 + "container/list" 5 + "fmt" 6 + "sync" 7 + 8 + badger "github.com/dgraph-io/badger/v4" 9 + ) 10 + 11 + // Cache is a persistent LRU cache implementation 12 + type Cache[K comparable, V any] struct { 13 + db *badger.DB 14 + mu sync.RWMutex 15 + 16 + // In-memory LRU tracking (Key -> List Element) 17 + cache map[K]*list.Element 18 + list *list.List 19 + 20 + // Serialization functions 21 + serialize func(V) ([]byte, error) 22 + deserialize func([]byte) (V, error) 23 + 24 + // Hooks 25 + OnAdd func(K) 26 + OnRemove func(K) 27 + } 28 + 29 + // New creates a new persistent LRU cache 30 + func New[K comparable, V any](dbPath string, serialize func(V) ([]byte, error), deserialize func([]byte) (V, error)) (*Cache[K, V], error) { 31 + opts := badger.DefaultOptions(dbPath).WithLogger(nil) // Suppress Badger logs 32 + db, err := badger.Open(opts) 33 + if err != nil { 34 + return nil, fmt.Errorf("failed to open Badger DB: %w", err) 35 + } 36 + 37 + c := &Cache[K, V]{ 38 + db: db, 39 + cache: make(map[K]*list.Element), 40 + list: list.New(), 41 + serialize: serialize, 42 + deserialize: deserialize, 43 + // Default NOP hooks 44 + OnAdd: func(K) {}, 45 + OnRemove: func(K) {}, 46 + } 47 + 48 + return c, nil 49 + } 50 + 51 + // Close closes the underlying BadgerDB 52 + func (c *Cache[K, V]) Close() error { 53 + return c.db.Close() 54 + } 55 + 56 + // LRUItem is used within the in-memory doubly linked list to track order 57 + type LRUItem[K comparable, V any] struct { 58 + Key K 59 + Value V 60 + } 61 + 62 + // Get retrieves a value from the cache 63 + func (c *Cache[K, V]) Get(key K) (V, bool) { 64 + c.mu.Lock() 65 + defer c.mu.Unlock() 66 + 67 + // Check in-memory cache first 68 + if ele, ok := c.cache[key]; ok { 69 + // Move to front (mark as recently used) 70 + c.list.MoveToFront(ele) 71 + return ele.Value.(*LRUItem[K, V]).Value, true 72 + } 73 + 74 + // Not in memory, check BadgerDB 75 + var zero V 76 + err := c.db.View(func(txn *badger.Txn) error { 77 + item, err := txn.Get([]byte(fmt.Sprintf("%v", key))) 78 + if err == badger.ErrKeyNotFound { 79 + return nil 80 + } 81 + if err != nil { 82 + return err 83 + } 84 + 85 + return item.Value(func(val []byte) error { 86 + value, err := c.deserialize(val) 87 + if err != nil { 88 + return err 89 + } 90 + 91 + // Add to LRU list 92 + ele := c.list.PushFront(&LRUItem[K, V]{Key: key, Value: value}) 93 + c.cache[key] = ele 94 + c.OnAdd(key) 95 + 96 + return nil 97 + }) 98 + }) 99 + 100 + if err != nil { 101 + return zero, false 102 + } 103 + 104 + return zero, false 105 + } 106 + 107 + // Put stores a value in the cache 108 + func (c *Cache[K, V]) Put(key K, value V) { 109 + c.mu.Lock() 110 + defer c.mu.Unlock() 111 + 112 + // Serialize the value 113 + serializedValue, err := c.serialize(value) 114 + if err != nil { 115 + return 116 + } 117 + 118 + // Write to BadgerDB 119 + err = c.db.Update(func(txn *badger.Txn) error { 120 + return txn.Set([]byte(fmt.Sprintf("%v", key)), serializedValue) 121 + }) 122 + if err != nil { 123 + return 124 + } 125 + 126 + // Update LRU list 127 + if ele, ok := c.cache[key]; ok { 128 + // Update existing value 129 + ele.Value.(*LRUItem[K, V]).Value = value 130 + c.list.MoveToFront(ele) 131 + } else { 132 + // Add new item 133 + ele := c.list.PushFront(&LRUItem[K, V]{Key: key, Value: value}) 134 + c.cache[key] = ele 135 + c.OnAdd(key) 136 + } 137 + } 138 + 139 + // Remove removes a value from the cache 140 + func (c *Cache[K, V]) Remove(key K) { 141 + c.mu.Lock() 142 + defer c.mu.Unlock() 143 + 144 + // Remove from BadgerDB 145 + err := c.db.Update(func(txn *badger.Txn) error { 146 + return txn.Delete([]byte(fmt.Sprintf("%v", key))) 147 + }) 148 + if err != nil && err != badger.ErrKeyNotFound { 149 + return 150 + } 151 + 152 + // Remove from LRU list 153 + if ele, ok := c.cache[key]; ok { 154 + c.list.Remove(ele) 155 + delete(c.cache, key) 156 + c.OnRemove(key) 157 + } 158 + } 159 + 160 + // Len returns the number of items in the cache 161 + func (c *Cache[K, V]) Len() int { 162 + c.mu.RLock() 163 + defer c.mu.RUnlock() 164 + return len(c.cache) 165 + } 166 + 167 + // Keys returns all keys in the cache 168 + func (c *Cache[K, V]) Keys() []K { 169 + c.mu.RLock() 170 + defer c.mu.RUnlock() 171 + 172 + keys := make([]K, 0, len(c.cache)) 173 + for k := range c.cache { 174 + keys = append(keys, k) 175 + } 176 + return keys 177 + } 178 + 179 + // Clear removes all items from the cache 180 + func (c *Cache[K, V]) Clear() error { 181 + c.mu.Lock() 182 + defer c.mu.Unlock() 183 + 184 + // Clear in-memory cache 185 + c.cache = make(map[K]*list.Element) 186 + c.list.Init() 187 + 188 + // Clear BadgerDB 189 + return c.db.DropAll() 190 + } 191 + 192 + // SetCapacity sets the maximum number of items the cache can hold 193 + // When the capacity is exceeded, the least recently used items are evicted 194 + func (c *Cache[K, V]) SetCapacity(capacity int) { 195 + c.mu.Lock() 196 + defer c.mu.Unlock() 197 + 198 + for len(c.cache) > capacity { 199 + // Get the least recently used item (back of the list) 200 + ele := c.list.Back() 201 + if ele == nil { 202 + break 203 + } 204 + 205 + item := ele.Value.(*LRUItem[K, V]) 206 + key := item.Key 207 + 208 + // Remove from BadgerDB 209 + err := c.db.Update(func(txn *badger.Txn) error { 210 + return txn.Delete([]byte(fmt.Sprintf("%v", key))) 211 + }) 212 + if err != nil && err != badger.ErrKeyNotFound { 213 + continue 214 + } 215 + 216 + // Remove from LRU list 217 + c.list.Remove(ele) 218 + delete(c.cache, key) 219 + c.OnRemove(key) 220 + } 221 + }

+386

labelmerge/stream/stream.go

··· 1 + package stream 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "log/slog" 7 + "net/http" 8 + "net/url" 9 + "strconv" 10 + "strings" 11 + "sync" 12 + "time" 13 + 14 + comatproto "github.com/bluesky-social/indigo/api/atproto" 15 + "github.com/bluesky-social/indigo/atproto/identity" 16 + "github.com/bluesky-social/indigo/atproto/syntax" 17 + "github.com/bluesky-social/indigo/events" 18 + "github.com/bluesky-social/indigo/events/schedulers/parallel" 19 + "github.com/gorilla/websocket" 20 + ) 21 + 22 + const LabelSubscribePath = "xrpc/com.atproto.label.subscribeLabels" 23 + const MaxWorkers = 4 24 + const EventChannelCapacity = 1000 25 + 26 + // LabelEvent is the simplified, unified event object sent to the consumer. 27 + type LabelEvent struct { 28 + SourceDid string // The DID of the labeler service that sent the event 29 + Cursor int64 // The sequence number (seq) of the event 30 + Value *comatproto.LabelDefs_Label 31 + } 32 + 33 + // SubscriptionContext holds the state for a single labeler connection. 34 + type SubscriptionContext struct { 35 + DID string 36 + ServiceURL string // Resolved WSS URL 37 + CurrentCursor string // Sequence string used for connection restarts 38 + WorkerCancel context.CancelFunc 39 + WorkerCtx context.Context 40 + IsRunning bool 41 + mu sync.Mutex 42 + } 43 + 44 + // LabelerSubscriptionManager manages connections to multiple ATProto labeler services. 45 + type LabelerSubscriptionManager struct { 46 + log *slog.Logger 47 + managerCtx context.Context 48 + managerCancel context.CancelFunc 49 + wg sync.WaitGroup 50 + 51 + resolver identity.Directory // DID Resolver dependency 52 + 53 + subscriptions map[string]*SubscriptionContext 54 + subMu sync.RWMutex 55 + 56 + eventCh chan LabelEvent 57 + } 58 + 59 + // NewLabelerSubscriptionManager creates a new instance of the manager. 60 + func NewLabelerSubscriptionManager(log *slog.Logger) *LabelerSubscriptionManager { 61 + ctx, cancel := context.WithCancel(context.Background()) 62 + dir := identity.DefaultDirectory() // Cached with 24hr TTL 63 + return &LabelerSubscriptionManager{ 64 + log: log, 65 + managerCtx: ctx, 66 + managerCancel: cancel, 67 + subscriptions: make(map[string]*SubscriptionContext), 68 + eventCh: make(chan LabelEvent, EventChannelCapacity), 69 + resolver: dir, 70 + } 71 + } 72 + 73 + // Events returns the read-only channel where all aggregated LabelEvents are sent. 74 + func (m *LabelerSubscriptionManager) Events() <-chan LabelEvent { 75 + return m.eventCh 76 + } 77 + 78 + // Start initiates the subscription manager. It attempts to resolve the URL and start workers 79 + // for all currently added labelers. 80 + func (m *LabelerSubscriptionManager) Start() { 81 + m.subMu.RLock() 82 + defer m.subMu.RUnlock() 83 + 84 + m.log.Info("Starting Labeler Subscription Manager", "count", len(m.subscriptions)) 85 + 86 + for _, sub := range m.subscriptions { 87 + // Initial resolution check (the worker loop handles retries) 88 + did, err1 := syntax.ParseDID(sub.DID) 89 + if err1 != nil { 90 + m.log.Warn(sub.DID + "ResolutionFailure invalid did") 91 + } else { 92 + ident, err2 := m.resolver.LookupDID(context.TODO(), did) 93 + if err2 != nil { 94 + m.log.Warn(sub.DID + "ResolutionFailure not reachable") 95 + } else { 96 + labelerURL := ident.GetServiceEndpoint("atproto_labeler") 97 + if labelerURL == "" { 98 + m.log.Warn(sub.DID + "ResolutionFailure no service endpoint") 99 + } else { 100 + sub.ServiceURL = labelerURL 101 + m.log.Info("Initial DID resolved successfully", "did", sub.DID, "service_url", sub.ServiceURL) 102 + } 103 + } 104 + } 105 + m.startWorker(sub) 106 + } 107 + } 108 + 109 + // Stop gracefully shuts down all active connections and the manager. 110 + func (m *LabelerSubscriptionManager) Stop() { 111 + m.managerCancel() 112 + m.log.Info("Waiting for all labeler workers to shut down...") 113 + m.wg.Wait() 114 + m.log.Info("Manager stopped gracefully.") 115 + close(m.eventCh) 116 + } 117 + 118 + // AddLabeler registers a new labeler DID. URL resolution is handled lazily (in Start or in the worker loop). 119 + func (m *LabelerSubscriptionManager) AddLabeler(did string, cursor string) error { 120 + if did == "" { 121 + return fmt.Errorf("did cannot be empty") 122 + } 123 + 124 + m.subMu.Lock() 125 + defer m.subMu.Unlock() 126 + 127 + if _, exists := m.subscriptions[did]; exists { 128 + m.log.Warn("Labeler already exists", "did", did) 129 + return nil 130 + } 131 + 132 + workerCtx, workerCancel := context.WithCancel(m.managerCtx) 133 + 134 + sub := &SubscriptionContext{ 135 + DID: did, 136 + ServiceURL: "", // Must be resolved before use 137 + CurrentCursor: cursor, 138 + WorkerCtx: workerCtx, 139 + WorkerCancel: workerCancel, 140 + IsRunning: false, 141 + } 142 + 143 + m.subscriptions[did] = sub 144 + 145 + // If the overall manager is running, try to resolve and start the worker immediately. 146 + select { 147 + case <-m.managerCtx.Done(): 148 + // Manager is shutting down, just register the sub but don't start 149 + default: 150 + // Attempt immediate resolution and start 151 + // entry, err := m.resolver.Resolve(sub.DID) 152 + // if err == nil { 153 + // sub.ServiceURL = entry.Domain 154 + // m.log.Info("Immediate DID resolution successful for new labeler", "did", sub.DID, "service_url", sub.ServiceURL) 155 + // } else { 156 + // m.log.Warn("Immediate DID resolution failed for new labeler. Worker will retry.", "did", sub.DID, "err", err) 157 + // } 158 + // m.startWorker(sub) 159 + did, err1 := syntax.ParseDID(sub.DID) 160 + if err1 != nil { 161 + m.log.Warn(sub.DID + "ResolutionFailure invalid did") 162 + } else { 163 + ident, err2 := m.resolver.LookupDID(context.TODO(), did) 164 + if err2 != nil { 165 + m.log.Warn(sub.DID + "ResolutionFailure not reachable") 166 + } else { 167 + labelerURL := ident.GetServiceEndpoint("atproto_labeler") 168 + if labelerURL == "" { 169 + m.log.Warn(sub.DID + "ResolutionFailure no service endpoint") 170 + } else { 171 + sub.ServiceURL = labelerURL 172 + m.log.Info("Initial DID resolved successfully", "did", sub.DID, "service_url", sub.ServiceURL) 173 + } 174 + } 175 + } 176 + m.startWorker(sub) 177 + } 178 + 179 + m.log.Info("Labeler added", "did", did, "start_cursor", cursor) 180 + return nil 181 + } 182 + 183 + // RemoveLabeler stops the stream worker for the given DID and removes it from the manager. 184 + func (m *LabelerSubscriptionManager) RemoveLabeler(did string) { 185 + m.subMu.Lock() 186 + defer m.subMu.Unlock() 187 + 188 + sub, exists := m.subscriptions[did] 189 + if !exists { 190 + return 191 + } 192 + 193 + m.log.Info("Removing labeler subscription", "did", did) 194 + sub.WorkerCancel() 195 + delete(m.subscriptions, did) 196 + } 197 + 198 + // startWorker spins up the connection management goroutine for a single labeler. 199 + func (m *LabelerSubscriptionManager) startWorker(sub *SubscriptionContext) { 200 + sub.mu.Lock() 201 + if sub.IsRunning { 202 + sub.mu.Unlock() 203 + return 204 + } 205 + sub.IsRunning = true 206 + sub.mu.Unlock() 207 + 208 + m.wg.Add(1) 209 + go func() { 210 + defer m.wg.Done() 211 + defer func() { 212 + sub.mu.Lock() 213 + sub.IsRunning = false 214 + sub.mu.Unlock() 215 + }() 216 + m.manageSubscription(sub) 217 + }() 218 + } 219 + 220 + // manageSubscription handles connection retries, cursors, and running the stream processor. 221 + func (m *LabelerSubscriptionManager) manageSubscription(sub *SubscriptionContext) { 222 + didLog := m.log.With("did", sub.DID) 223 + didLog.Info("Worker started") 224 + 225 + for { 226 + select { 227 + case <-sub.WorkerCtx.Done(): 228 + didLog.Info("Worker received stop signal, shutting down.") 229 + return 230 + default: 231 + // Proceed 232 + } 233 + var err error 234 + 235 + // 1. Ensure ServiceURL is resolved 236 + if sub.ServiceURL == "" { 237 + didLog.Info("Service URL not resolved, attempting DID resolution.") 238 + // entry, err := m.resolver.Resolve(sub.DID) 239 + // if err != nil { 240 + // didLog.Error("DID resolution failed, retrying...", "err", err) 241 + // goto WaitAndRetry 242 + // } 243 + // sub.ServiceURL = entry.Domain 244 + // didLog.Info("DID resolution successful", "service_url", sub.ServiceURL) 245 + did, err1 := syntax.ParseDID(sub.DID) 246 + if err1 != nil { 247 + m.log.Warn(sub.DID + "ResolutionFailure invalid did") 248 + } else { 249 + ident, err2 := m.resolver.LookupDID(context.TODO(), did) 250 + if err2 != nil { 251 + m.log.Warn(sub.DID + "ResolutionFailure not reachable") 252 + goto WaitAndRetry 253 + } else { 254 + labelerURL := ident.GetServiceEndpoint("atproto_labeler") 255 + if labelerURL == "" { 256 + m.log.Warn(sub.DID + "ResolutionFailure no service endpoint") 257 + } else { 258 + sub.ServiceURL = labelerURL 259 + m.log.Info("Initial DID resolved successfully", "did", sub.DID, "service_url", sub.ServiceURL) 260 + } 261 + } 262 + } 263 + } 264 + 265 + // 2. Attempt to stream 266 + err = m.dialAndStream(sub) 267 + 268 + if sub.WorkerCtx.Err() != nil { 269 + return 270 + } 271 + 272 + if err != nil { 273 + didLog.Error("Stream failed, attempting restart", "err", err, "cursor", sub.CurrentCursor) 274 + } else { 275 + didLog.Info("Stream closed cleanly, attempting restart.") 276 + } 277 + 278 + WaitAndRetry: 279 + // Wait before retrying 280 + didLog.Info("Waiting 5s before reconnecting/retrying resolution...") 281 + select { 282 + case <-time.After(5 * time.Second): 283 + // Proceed with retry 284 + case <-sub.WorkerCtx.Done(): 285 + return // Exit if canceled during wait 286 + } 287 + } 288 + } 289 + 290 + func httpToWS(s string) string { 291 + if strings.HasPrefix(s, "https://") { 292 + return "wss://" + s[len("https://"):] 293 + } 294 + if strings.HasPrefix(s, "http://") { 295 + return "ws://" + s[len("http://"):] 296 + } 297 + return s 298 + } 299 + 300 + // dialAndStream establishes the WebSocket connection and processes the stream. 301 + func (m *LabelerSubscriptionManager) dialAndStream(sub *SubscriptionContext) error { 302 + didLog := m.log.With("did", sub.DID) 303 + 304 + fullURL := httpToWS(sub.ServiceURL) + "/" + LabelSubscribePath 305 + if sub.CurrentCursor != "" { 306 + fullURL = fmt.Sprintf("%s?cursor=%s", fullURL, sub.CurrentCursor) 307 + } 308 + 309 + u, err := url.Parse(fullURL) 310 + if err != nil { 311 + return fmt.Errorf("failed to parse URL: %w", err) 312 + } 313 + 314 + // 1. Establish WebSocket Connection 315 + dialer := websocket.DefaultDialer 316 + con, resp, err := dialer.Dial(u.String(), http.Header{ 317 + "User-Agent": []string{"LabelerSubscriptionManager/1.0"}, 318 + }) 319 + 320 + if err != nil { 321 + if resp != nil { 322 + didLog.Error("WebSocket connection failed", "status", resp.StatusCode) 323 + } 324 + // If dial fails due to network/DNS/TLS error, clear the service URL to force re-resolution 325 + // on the next loop iteration. 326 + sub.mu.Lock() 327 + sub.ServiceURL = "" 328 + sub.mu.Unlock() 329 + 330 + return fmt.Errorf("failed to dial websocket to %s: %w", u.String(), err) 331 + } 332 + defer con.Close() 333 + didLog.Info("Successfully connected to Labeler firehose", "url", u.String()) 334 + 335 + // 2. Define Event Callbacks 336 + rsc := &events.RepoStreamCallbacks{ 337 + LabelLabels: func(evt *comatproto.LabelSubscribeLabels_Labels) error { 338 + if evt.Seq == 0 || evt.Labels == nil { 339 + return nil 340 + } 341 + 342 + // Update the cursor immediately 343 + sub.mu.Lock() 344 + sub.CurrentCursor = strconv.FormatInt(evt.Seq, 10) 345 + sub.mu.Unlock() 346 + 347 + // Process and simplify each label event 348 + for _, label := range evt.Labels { 349 + select { 350 + case m.eventCh <- LabelEvent{ 351 + SourceDid: sub.DID, 352 + Cursor: evt.Seq, 353 + Value: label, 354 + }: 355 + // Sent successfully 356 + default: 357 + didLog.Warn("Event channel full, dropping label event", "seq", evt.Seq, "uri", label.Uri) 358 + } 359 + } 360 + return nil 361 + }, 362 + 363 + LabelInfo: func(evt *comatproto.LabelSubscribeLabels_Info) error { 364 + if evt.Message != nil { 365 + didLog.Info("Stream Info Message", "name", evt.Name, "message", *evt.Message) 366 + } 367 + return nil 368 + }, 369 + 370 + Error: func(evt *events.ErrorFrame) error { 371 + didLog.Error("Stream processing error frame", "error_type", evt.Error, "message", evt.Message) 372 + return fmt.Errorf("atproto stream error: %s", evt.Message) 373 + }, 374 + } 375 + 376 + // 3. Create Scheduler and Start Processing 377 + scheduler := parallel.NewScheduler( 378 + MaxWorkers, 379 + EventChannelCapacity, 380 + fullURL, 381 + rsc.EventHandler, 382 + ) 383 + 384 + // HandleRepoStream blocks until the connection closes or the context cancels. 385 + return events.HandleRepoStream(sub.WorkerCtx, con, scheduler, didLog) 386 + }

Configure Feed

Configure Feed