···

       
48
       
48
       
 
       
    redis_cnx = redis.Redis()

     

       
49
       
49
       
 
       
    redis_pipe = redis_cnx.pipeline()

     

       
50
       
50
       
 
       


     

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

     

       
52
       
52
       
-
       
        db_fname = '/opt/muninsky/users.db'

     

       
53
       
53
       
-
       
    else:

     

       
54
       
54
       
-
       
        db_fname = 'users.db'

     

       
55
       
55
       
-
       


     

       
56
       
56
       
-
       
    db_cnx = sqlite3.connect(db_fname)

     

       
57
       
57
       
-
       
    with db_cnx:

     

       
58
       
58
       
-
       
        db_cnx.executescript("""

     

       
59
       
59
       
-
       
        PRAGMA journal_mode = WAL;

     

       
60
       
60
       
-
       
        PRAGMA synchronous = off;

     

       
61
       
61
       
-
       
        CREATE TABLE IF NOT EXISTS users (did TEXT, ts TIMESTAMP);

     

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

     

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

     

       
64
       
64
       
-
       
        """)

     

       
65
       
65
       
-
       


     

       
66
       
51
       
 
       
    sys.stdout.write('starting up\n')

     

       
67
       
52
       
 
       
    sys.stdout.flush()

     

       
68
       
53
       
 
       


     
···

       
84
       
69
       
 
       


     

       
85
       
70
       
 
       
        repo_did = event['did']

     

       
86
       
71
       
 
       
        repo_update_time = datetime.now(timezone.utc)

     

       
87
       
87
       
-
       
        db_cnx.execute(

     

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

     

       
89
       
89
       
-
       
            {'did': repo_did, 'ts': repo_update_time.timestamp()}

     

       
90
       
90
       
-
       
        )

     

       
91
       
72
       
 
       


     

       
92
       
73
       
 
       
        if collection == 'app.bsky.feed.post':

     

       
93
       
74
       
 
       
            embed = payload['record'].get('embed')

     
···

       
106
       
87
       
 
       
            current_lag = current_time_ms - event_time_ms

     

       
107
       
88
       
 
       
            sys.stdout.write(f'lag: {current_lag:.2f}\n')

     

       
108
       
89
       
 
       
            redis_pipe.execute()

     

       
109
       
109
       
-
       
            db_cnx.commit()

     

       
110
       
90
       
 
       
            sys.stdout.flush()

     

       
111
       
91
       
 
       


     

       
112
       
92
       
 
       
if __name__ == '__main__':