this repo has no description
firehose_utils.py: abstract further to subscribe_commits()
+33
-9
+33
-9
firehose_utils.py
+33
-9
firehose_utils.py
···
1
+
import dag_cbor
2
+
import redis
1
3
from atproto import CAR
4
+
from io import BytesIO
2
5
3
-
def commit_ops(payload):
4
-
# TODO(ejd): figure out how to validate blocks
5
-
car_parsed = CAR.from_bytes(payload['blocks'])
6
-
for op in payload['ops']:
7
-
repo_op = op.copy()
8
-
if op['cid'] is not None:
9
-
repo_op['cid'] = op['cid'].encode('base32')
10
-
repo_op['record'] = car_parsed.blocks[repo_op['cid']]
11
-
yield repo_op
6
+
def subscribe_commits():
7
+
redis_cnx = redis.Redis()
8
+
redis_sub = redis_cnx.pubsub(ignore_subscribe_messages=True)
9
+
redis_sub.subscribe('bsky-tools:firehose:stream')
10
+
11
+
for event in redis_sub.listen():
12
+
frame = BytesIO(event['data'])
13
+
header = dag_cbor.decode(frame, allow_concat=True)
14
+
if header['op'] != 1 or header['t'] != '#commit':
15
+
continue
16
+
17
+
payload = dag_cbor.decode(frame)
18
+
if payload['tooBig']:
19
+
# TODO(ejd): figure out how to get blocks out-of-band
20
+
continue
21
+
22
+
# TODO(ejd): figure out how to validate blocks
23
+
blocks = payload.pop('blocks')
24
+
car_parsed = CAR.from_bytes(blocks)
25
+
26
+
message = payload.copy()
27
+
del message['ops']
28
+
message['commit'] = message['commit'].encode('base32')
29
+
30
+
for commit_op in payload['ops']:
31
+
op = commit_op.copy()
32
+
if op['cid'] is not None:
33
+
op['cid'] = op['cid'].encode('base32')
34
+
op['record'] = car_parsed.blocks[op['cid']]
35
+
yield message, op