···

       
1
       
1
       
-
       
#!/usr/bin/env python3

     

       
2
       
2
       
-
       


     

       
3
       
3
       
-
       
import asyncio

     

       
4
       
4
       
-
       
import dag_cbor

     

       
5
       
5
       
-
       
import redis

     

       
6
       
6
       
-
       
import sys

     

       
7
       
7
       
-
       
import websockets

     

       
8
       
8
       
-
       
from datetime import datetime, timezone

     

       
9
       
9
       
-
       
from io import BytesIO

     

       
10
       
10
       
-
       


     

       
11
       
11
       
-
       
async def main():

     

       
12
       
12
       
-
       
    redis_cnx = redis.Redis()

     

       
13
       
13
       
-
       
    relay_url = 'ws://127.0.0.1:9060'

     

       
14
       
14
       
-
       
    sys.stdout.write(f'opening websocket connection to {relay_url}\n')

     

       
15
       
15
       
-
       
    sys.stdout.flush()

     

       
16
       
16
       
-
       


     

       
17
       
17
       
-
       
    async with websockets.connect(relay_url, ping_timeout=None) as firehose:

     

       
18
       
18
       
-
       
        current_minute = None

     

       
19
       
19
       
-
       
        while True:

     

       
20
       
20
       
-
       
            frame = BytesIO(await firehose.recv())

     

       
21
       
21
       
-
       
            header = dag_cbor.decode(frame, allow_concat=True)

     

       
22
       
22
       
-
       
            if header['op'] != 1:

     

       
23
       
23
       
-
       
                continue

     

       
24
       
24
       
-
       


     

       
25
       
25
       
-
       
            # checkpoint the seq

     

       
26
       
26
       
-
       
            now = datetime.now(timezone.utc)

     

       
27
       
27
       
-
       
            if now.time().minute != current_minute:

     

       
28
       
28
       
-
       
                current_minute = now.time().minute

     

       
29
       
29
       
-
       


     

       
30
       
30
       
-
       
                payload = dag_cbor.decode(frame)

     

       
31
       
31
       
-
       
                payload_seq = payload['seq']

     

       
32
       
32
       
-
       
                payload_time = datetime.strptime(payload['time'], '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=timezone.utc)

     

       
33
       
33
       
-
       
                payload_lag = now - payload_time

     

       
34
       
34
       
-
       


     

       
35
       
35
       
-
       
                redis_cnx.set('bsky-tools:firehose:subscribe-repos:seq', payload_seq)

     

       
36
       
36
       
-
       
                sys.stdout.write(f'seq: {payload_seq}, lag: {payload_lag.total_seconds()}\n')

     

       
37
       
37
       
-
       


     

       
38
       
38
       
-
       
            sys.stdout.flush()

     

       
39
       
39
       
-
       


     

       
40
       
40
       
-
       
if __name__ == '__main__':

     

       
41
       
41
       
-
       
    asyncio.run(main())