···

       
1
       
 
       
#!/usr/bin/env python3

     

       
2
       
 
       


     

       
3
       
-
       
import dag_cbor

     

       
4
       
 
       
import os

     

       
5
       
 
       
import redis

     

       
6
       
 
       
import sqlite3

     

       
7
       
 
       
import sys

     

       
8
       
 
       
from datetime import datetime, timezone

     

       
9
       
-
       
from firehose_utils import commit_ops

     

       
10
       
-
       
from io import BytesIO

     

       
11
       
 
       


     

       
12
       
 
       
app_bsky_allowlist = set([

     

       
13
       
 
       
    'app.bsky.actor.profile',

     
···

       
27
       
 
       
def main():

     

       
28
       
 
       
    redis_cnx = redis.Redis()

     

       
29
       
 
       
    redis_pipe = redis_cnx.pipeline()

     

       
30
       
-
       
    redis_sub = redis_cnx.pubsub(ignore_subscribe_messages=True)

     

       
31
       
 
       


     

       
32
       
 
       
    if os.path.exists('/opt/muninsky/users.db'):

     

       
33
       
 
       
        db_fname = '/opt/muninsky/users.db'

     
···

       
43
       
 
       
        CREATE UNIQUE INDEX IF NOT EXISTS did_idx on users(did);

     

       
44
       
 
       
        CREATE INDEX IF NOT EXISTS ts_idx on users(ts);

     

       
45
       
 
       
        """)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
46
       
 
       


     

       
47
       
 
       
    op_count = 0

     

       
48
       
-
       
    redis_sub.subscribe('bsky-tools:firehose:stream')

     

       
49
       
-
       
    for event in redis_sub.listen():

     

       
50
       
-
       
        frame = BytesIO(event['data'])

     

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

     

       
52
       
-
       
        if header['op'] != 1 or header['t'] != '#commit':

     

       
53
       
 
       
            continue

     

       
54
       
 
       


     

       
55
       
-
       
        payload = dag_cbor.decode(frame)

     

       
56
       
-
       
        if payload['tooBig']:

     

       
57
       
-
       
            # TODO(ejd): how handle these?

     

       
58
       
 
       
            continue

     

       
59
       
 
       


     

       
60
       
-
       
        for op in commit_ops(payload):

     

       
61
       
-
       
            if op['action'] != 'create':

     

       
62
       
-
       
                continue

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
63
       
 
       


     

       
64
       
-
       
            collection, _ = op['path'].split('/')

     

       
65
       
-
       
            if collection not in app_bsky_allowlist:

     

       
66
       
-
       
                continue

     

       
67
       
 
       


     

       
68
       
-
       
            repo_did = payload['repo']

     

       
69
       
-
       
            ts = datetime.now(timezone.utc).timestamp()

     

       
70
       
-
       
            db_cnx.execute(

     

       
71
       
-
       
                'insert into users values (:did, :ts) on conflict (did) do update set ts = :ts',

     

       
72
       
-
       
                {'did': repo_did, 'ts': ts}

     

       
73
       
-
       
            )

     

       
74
       
 
       


     

       
75
       
-
       
            redis_pipe \

     

       
76
       
-
       
                .incr(collection) \

     

       
77
       
-
       
                .incr('dev.edavis.muninsky.ops')

     

       
78
       
-
       


     

       
79
       
-
       
            op_count += 1

     

       
80
       
-
       
            if op_count % 500 == 0:

     

       
81
       
-
       
                payload_seq = payload['seq']

     

       
82
       
-
       
                sys.stdout.write(f'checkpoint: seq: {payload_seq}\n')

     

       
83
       
-
       
                redis_pipe.set('dev.edavis.muninsky.seq', payload_seq)

     

       
84
       
-
       
                redis_pipe.execute()

     

       
85
       
-
       
                db_cnx.commit()

     

       
86
       
-
       
                sys.stdout.flush()

     

       
87
       
 
       


     

       
88
       
 
       
if __name__ == '__main__':

     

       
89
       
 
       
    main()

     

···

       
1
       
 
       
#!/usr/bin/env python3

     

       
2
       
 
       


     

       
0
       
 
       

     

       
3
       
 
       
import os

     

       
4
       
 
       
import redis

     

       
5
       
 
       
import sqlite3

     

       
6
       
 
       
import sys

     

       
7
       
 
       
from datetime import datetime, timezone

     

       
8
       
+
       
from firehose_utils import subscribe_commits

     

       
0
       
 
       

     

       
9
       
 
       


     

       
10
       
 
       
app_bsky_allowlist = set([

     

       
11
       
 
       
    'app.bsky.actor.profile',

     
···

       
25
       
 
       
def main():

     

       
26
       
 
       
    redis_cnx = redis.Redis()

     

       
27
       
 
       
    redis_pipe = redis_cnx.pipeline()

     

       
0
       
 
       

     

       
28
       
 
       


     

       
29
       
 
       
    if os.path.exists('/opt/muninsky/users.db'):

     

       
30
       
 
       
        db_fname = '/opt/muninsky/users.db'

     
···

       
40
       
 
       
        CREATE UNIQUE INDEX IF NOT EXISTS did_idx on users(did);

     

       
41
       
 
       
        CREATE INDEX IF NOT EXISTS ts_idx on users(ts);

     

       
42
       
 
       
        """)

     

       
43
       
+
       


     

       
44
       
+
       
    sys.stdout.write('starting up\n')

     

       
45
       
+
       
    sys.stdout.flush()

     

       
46
       
 
       


     

       
47
       
 
       
    op_count = 0

     

       
48
       
+
       
    for commit, op in subscribe_commits():

     

       
49
       
+
       
        if op['action'] != 'create':

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
50
       
 
       
            continue

     

       
51
       
 
       


     

       
52
       
+
       
        collection, _ = op['path'].split('/')

     

       
53
       
+
       
        if collection not in app_bsky_allowlist:

     

       
0
       
 
       

     

       
54
       
 
       
            continue

     

       
55
       
 
       


     

       
56
       
+
       
        repo_did = commit['repo']

     

       
57
       
+
       
        now = datetime.now(timezone.utc)

     

       
58
       
+
       
        db_cnx.execute(

     

       
59
       
+
       
            'insert into users values (:did, :ts) on conflict (did) do update set ts = :ts',

     

       
60
       
+
       
            {'did': repo_did, 'ts': now.timestamp()}

     

       
61
       
+
       
        )

     

       
62
       
 
       


     

       
63
       
+
       
        redis_pipe \

     

       
64
       
+
       
            .incr(collection) \

     

       
65
       
+
       
            .incr('dev.edavis.muninsky.ops')

     

       
66
       
 
       


     

       
67
       
+
       
        op_count += 1

     

       
68
       
+
       
        if op_count % 1000 == 0:

     

       
69
       
+
       
            payload_seq = commit['seq']

     

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

     

       
71
       
+
       
            payload_lag = now - payload_time

     

       
0
       
 
       

     

       
72
       
 
       


     

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

     

       
74
       
+
       
            redis_pipe.set('dev.edavis.muninsky.seq', payload_seq)

     

       
75
       
+
       
            redis_pipe.execute()

     

       
76
       
+
       
            db_cnx.commit()

     

       
77
       
+
       
            sys.stdout.flush()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
78
       
 
       


     

       
79
       
 
       
if __name__ == '__main__':

     

       
80
       
 
       
    main()