this repo has no description
edavis.dev
1#!/usr/bin/env python3
2
3import asyncio
4from datetime import datetime, timezone
5import json
6import os
7import sqlite3
8import sys
9
10import redis
11import websockets
12
13app_bsky_allowlist = (
14 'app.bsky.actor.profile',
15 'app.bsky.feed.generator',
16 'app.bsky.feed.like',
17 'app.bsky.feed.post',
18 'app.bsky.feed.postgate',
19 'app.bsky.feed.repost',
20 'app.bsky.feed.threadgate',
21 'app.bsky.graph.block',
22 'app.bsky.graph.follow',
23 'app.bsky.graph.list',
24 'app.bsky.graph.listblock',
25 'app.bsky.graph.listitem',
26 'app.bsky.graph.starterpack',
27 'app.bsky.labeler.service',
28 'chat.bsky.actor.declaration',
29)
30
31other_allowlist = (
32 'social.psky',
33 'blue.zio.atfile',
34 'com.shinolabs.pinksea',
35 'com.whtwnd',
36 'events.smokesignal',
37 'fyi.unravel',
38 'xyz.statusphere',
39)
40
41async def bsky_activity():
42 relay_url = 'wss://jetstream1.us-west.bsky.network/subscribe'
43
44 sys.stdout.write(f'opening websocket connection to {relay_url}\n')
45 sys.stdout.flush()
46
47 async with websockets.connect(relay_url, ping_timeout=60) as firehose:
48 while True:
49 yield json.loads(await firehose.recv())
50
51async def main():
52 redis_cnx = redis.Redis()
53 redis_pipe = redis_cnx.pipeline()
54
55 sys.stdout.write('starting up\n')
56 sys.stdout.flush()
57
58 op_count = 0
59 async for event in bsky_activity():
60 if event['kind'] != 'commit':
61 continue
62
63 payload = event.get('commit')
64 if payload is None:
65 continue
66
67 if payload['operation'] != 'create':
68 continue
69
70 collection = payload['collection']
71 if not collection.startswith(app_bsky_allowlist + other_allowlist):
72 continue
73
74 for prefix in other_allowlist:
75 if collection.startswith(prefix):
76 redis_pipe.incr('dev.edavis.atproto.collection.' + prefix.replace('.', '_'))
77
78 repo_did = event['did']
79 repo_update_time = datetime.now(timezone.utc)
80
81 if collection == 'app.bsky.feed.post':
82 embed = payload['record'].get('embed')
83 if embed is not None and embed.get('$type', ''):
84 embed_type = embed['$type']
85 redis_pipe.incr(f'app.bsky.feed.post:embed:{embed_type}')
86
87 redis_pipe \
88 .incr(collection) \
89 .incr('dev.edavis.muninsky.ops')
90
91 op_count += 1
92 if op_count % 500 == 0:
93 current_time_ms = datetime.now(timezone.utc).timestamp()
94 event_time_ms = event['time_us'] / 1_000_000
95 current_lag = current_time_ms - event_time_ms
96 sys.stdout.write(f'lag: {current_lag:.2f}\n')
97 redis_pipe.execute()
98 sys.stdout.flush()
99
100if __name__ == '__main__':
101 asyncio.run(main())