···

       
7
       
 
       
	"os"

     

       
8
       
 
       
	"os/signal"

     

       
9
       
 
       
	"strings"

     

       
0
       
 
       

     

       
10
       
 
       
	"syscall"

     

       
0
       
 
       

     

       
11
       
 
       


     

       
12
       
 
       
	appbsky "github.com/bluesky-social/indigo/api/bsky"

     

       
13
       
 
       
	jetstream "github.com/bluesky-social/jetstream/pkg/models"

     
···

       
15
       
 
       
	"github.com/redis/go-redis/v9"

     

       
16
       
 
       
)

     

       
17
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
18
       
 
       
const JetstreamUrl = `wss://jetstream1.us-west.bsky.network/subscribe`

     

       
19
       
 
       


     

       
20
       
 
       
var AppBskyAllowlist = map[string]bool{

     
···

       
59
       
 
       
	return false

     

       
60
       
 
       
}

     

       
61
       
 
       


     

       
62
       
-
       
func handler(ctx context.Context, events <-chan jetstream.Event) {

     

       
63
       
 
       
	rdb := redis.NewClient(&redis.Options{

     

       
64
       
 
       
		Addr:     "localhost:6379",

     

       
65
       
 
       
		Password: "",

     
···

       
69
       
 
       
	var eventCount int

     

       
70
       
 
       


     

       
71
       
 
       
eventLoop:

     

       
72
       
-
       
	for event := range events {

     

       
73
       
 
       
		select {

     

       
74
       
 
       
		case <-ctx.Done():

     

       
75
       
 
       
			break eventLoop

     

       
76
       
 
       
		default:

     

       
77
       
 
       
		}

     

       
78
       
 
       


     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
79
       
 
       
		if event.Kind != jetstream.EventKindCommit {

     

       
80
       
 
       
			continue

     

       
81
       
 
       
		}

     
···

       
145
       
 
       
			if _, err := pipe.Exec(ctx); err != nil {

     

       
146
       
 
       
				log.Printf("failed to exec pipe\n")

     

       
147
       
 
       
			}

     

       
0
       
 
       

     

       
148
       
 
       
		}

     

       
149
       
 
       
	}

     

       
150
       
 
       
}

     
···

       
164
       
 
       
		log.Printf("websocket closed\n")

     

       
165
       
 
       
	}()

     

       
166
       
 
       


     

       
167
       
-
       
	jetstreamEvents := make(chan jetstream.Event)

     

       
168
       
-
       
	go handler(ctx, jetstreamEvents)

     

       
169
       
 
       


     

       
170
       
 
       
	log.Printf("starting up\n")

     

       
171
       
-
       
	var event jetstream.Event

     

       
172
       
 
       
	go func() {

     

       
173
       
 
       
		for {

     

       
174
       
-
       
			event = jetstream.Event{}

     

       
175
       
 
       
			err := conn.ReadJSON(&event)

     

       
176
       
 
       
			if err != nil {

     

       
177
       
 
       
				log.Printf("ReadJSON error: %v\n", err)

     

       
178
       
 
       
				stop()

     

       
179
       
 
       
				break

     

       
180
       
-
       
			} else {

     

       
181
       
-
       
				jetstreamEvents <- event

     

       
182
       
 
       
			}

     

       
0
       
 
       

     

       
183
       
 
       
		}

     

       
184
       
 
       
	}()

     

       
185
       
 
       


     

···

       
7
       
 
       
	"os"

     

       
8
       
 
       
	"os/signal"

     

       
9
       
 
       
	"strings"

     

       
10
       
+
       
	"sync"

     

       
11
       
 
       
	"syscall"

     

       
12
       
+
       
	"time"

     

       
13
       
 
       


     

       
14
       
 
       
	appbsky "github.com/bluesky-social/indigo/api/bsky"

     

       
15
       
 
       
	jetstream "github.com/bluesky-social/jetstream/pkg/models"

     
···

       
17
       
 
       
	"github.com/redis/go-redis/v9"

     

       
18
       
 
       
)

     

       
19
       
 
       


     

       
20
       
+
       
type Queue struct {

     

       
21
       
+
       
	lk     sync.Mutex

     

       
22
       
+
       
	events []jetstream.Event

     

       
23
       
+
       
}

     

       
24
       
+
       


     

       
25
       
+
       
func NewQueue(capacity int) *Queue {

     

       
26
       
+
       
	return &Queue{

     

       
27
       
+
       
		events: make([]jetstream.Event, 0, capacity),

     

       
28
       
+
       
	}

     

       
29
       
+
       
}

     

       
30
       
+
       


     

       
31
       
+
       
func (q *Queue) Enqueue(event jetstream.Event) {

     

       
32
       
+
       
	q.lk.Lock()

     

       
33
       
+
       
	defer q.lk.Unlock()

     

       
34
       
+
       


     

       
35
       
+
       
	q.events = append(q.events, event)

     

       
36
       
+
       
}

     

       
37
       
+
       


     

       
38
       
+
       
func (q *Queue) Dequeue() (jetstream.Event, bool) {

     

       
39
       
+
       
	q.lk.Lock()

     

       
40
       
+
       
	defer q.lk.Unlock()

     

       
41
       
+
       


     

       
42
       
+
       
	var event jetstream.Event

     

       
43
       
+
       


     

       
44
       
+
       
	if len(q.events) == 0 {

     

       
45
       
+
       
		return event, false

     

       
46
       
+
       
	}

     

       
47
       
+
       


     

       
48
       
+
       
	event = q.events[0]

     

       
49
       
+
       
	q.events = q.events[1:]

     

       
50
       
+
       
	return event, true

     

       
51
       
+
       
}

     

       
52
       
+
       


     

       
53
       
+
       
func (q *Queue) Size() int {

     

       
54
       
+
       
	q.lk.Lock()

     

       
55
       
+
       
	defer q.lk.Unlock()

     

       
56
       
+
       


     

       
57
       
+
       
	return len(q.events)

     

       
58
       
+
       
}

     

       
59
       
+
       


     

       
60
       
 
       
const JetstreamUrl = `wss://jetstream1.us-west.bsky.network/subscribe`

     

       
61
       
 
       


     

       
62
       
 
       
var AppBskyAllowlist = map[string]bool{

     
···

       
101
       
 
       
	return false

     

       
102
       
 
       
}

     

       
103
       
 
       


     

       
104
       
+
       
func handler(ctx context.Context, queue *Queue) {

     

       
105
       
 
       
	rdb := redis.NewClient(&redis.Options{

     

       
106
       
 
       
		Addr:     "localhost:6379",

     

       
107
       
 
       
		Password: "",

     
···

       
111
       
 
       
	var eventCount int

     

       
112
       
 
       


     

       
113
       
 
       
eventLoop:

     

       
114
       
+
       
	for {

     

       
115
       
 
       
		select {

     

       
116
       
 
       
		case <-ctx.Done():

     

       
117
       
 
       
			break eventLoop

     

       
118
       
 
       
		default:

     

       
119
       
 
       
		}

     

       
120
       
 
       


     

       
121
       
+
       
		event, ok := queue.Dequeue()

     

       
122
       
+
       
		if !ok {

     

       
123
       
+
       
			time.Sleep(100 * time.Millisecond)

     

       
124
       
+
       
			continue

     

       
125
       
+
       
		}

     

       
126
       
+
       


     

       
127
       
 
       
		if event.Kind != jetstream.EventKindCommit {

     

       
128
       
 
       
			continue

     

       
129
       
 
       
		}

     
···

       
193
       
 
       
			if _, err := pipe.Exec(ctx); err != nil {

     

       
194
       
 
       
				log.Printf("failed to exec pipe\n")

     

       
195
       
 
       
			}

     

       
196
       
+
       
			log.Printf("queue size: %d\n", queue.Size())

     

       
197
       
 
       
		}

     

       
198
       
 
       
	}

     

       
199
       
 
       
}

     
···

       
213
       
 
       
		log.Printf("websocket closed\n")

     

       
214
       
 
       
	}()

     

       
215
       
 
       


     

       
216
       
+
       
	queue := NewQueue(100_000)

     

       
217
       
+
       
	go handler(ctx, queue)

     

       
218
       
 
       


     

       
219
       
 
       
	log.Printf("starting up\n")

     

       
0
       
 
       

     

       
220
       
 
       
	go func() {

     

       
221
       
 
       
		for {

     

       
222
       
+
       
			var event jetstream.Event

     

       
223
       
 
       
			err := conn.ReadJSON(&event)

     

       
224
       
 
       
			if err != nil {

     

       
225
       
 
       
				log.Printf("ReadJSON error: %v\n", err)

     

       
226
       
 
       
				stop()

     

       
227
       
 
       
				break

     

       
0
       
 
       

     

       
0
       
 
       

     

       
228
       
 
       
			}

     

       
229
       
+
       
			queue.Enqueue(event)

     

       
230
       
 
       
		}

     

       
231
       
 
       
	}()

     

       
232