alice.pds.demo.boltless.dev
This repository has no description
1.3 kB
72 lines
1package main
2
3import (
4 "sync"
5
6 "github.com/bluesky-social/indigo/atproto/syntax"
7)
8
9// deduplicating index work queue
10type Queue struct {
11 mu sync.Mutex
12 order []syntax.DID
13 pending map[syntax.DID]indexRequest
14 size int
15}
16
17func NewQueue(size int) *Queue {
18 return &Queue{
19 pending: make(map[syntax.DID]indexRequest),
20 size: size,
21 }
22}
23
24func (q *Queue) Enqueue(req indexRequest) bool {
25 q.mu.Lock()
26 defer q.mu.Unlock()
27
28 if _, exists := q.pending[req.Repo]; exists {
29 q.pending[req.Repo] = req // replace payload, keep position
30 return true
31 }
32
33 if len(q.order) >= q.size {
34 return false // queue full
35 }
36
37 q.order = append(q.order, req.Repo)
38 q.pending[req.Repo] = req
39 return true
40}
41
42func (q *Queue) Pop() (indexRequest, bool) {
43 q.mu.Lock()
44 defer q.mu.Unlock()
45
46 if len(q.order) == 0 {
47 return indexRequest{}, false
48 }
49
50 did := q.order[0]
51 q.order = q.order[1:]
52 if len(q.order) == 0 {
53 q.order = nil // release the backing array
54 }
55
56 req, ok := q.pending[did]
57 delete(q.pending, did)
58 return req, ok
59}
60
61func (q *Queue) Snapshot() []indexRequest {
62 q.mu.Lock()
63 defer q.mu.Unlock()
64
65 out := make([]indexRequest, 0, len(q.order))
66 for _, did := range q.order {
67 if req, ok := q.pending[did]; ok {
68 out = append(out, req)
69 }
70 }
71 return out
72}