···

       
75
       
75
       
 
       
            if collection.startswith(prefix):

     

       
76
       
76
       
 
       
                redis_pipe.incr('dev.edavis.atproto.collection.' + prefix.replace('.', '_'))

     

       
77
       
77
       
 
       


     

       
78
       
78
       
-
       
        repo_did = event['did']

     

       
79
       
79
       
-
       
        repo_update_time = datetime.now(timezone.utc)

     

       
80
       
80
       
-
       


     

       
81
       
78
       
 
       
        if collection == 'app.bsky.feed.post':

     

       
82
       
79
       
 
       
            embed = payload['record'].get('embed')

     

       
83
       
80
       
 
       
            if embed is not None and embed.get('$type', ''):

     
···

       
90
       
87
       
 
       


     

       
91
       
88
       
 
       
        op_count += 1

     

       
92
       
89
       
 
       
        if op_count % 500 == 0:

     

       
93
       
93
       
-
       
            current_time_ms = datetime.now(timezone.utc).timestamp()

     

       
94
       
90
       
 
       
            event_time_ms = event['time_us'] / 1_000_000

     

       
95
       
95
       
-
       
            current_lag = current_time_ms - event_time_ms

     

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

     

       
91
       
91
       
+
       
            sys.stdout.write(f'timestamp: {event_time_ms}\n')

     

       
97
       
92
       
 
       
            redis_pipe.execute()

     

       
98
       
93
       
 
       
            sys.stdout.flush()

     

       
99
       
94