this repo has no description
edavis.dev
1#!/usr/bin/env python3
2
3import asyncio
4from datetime import datetime, timezone
5from io import BytesIO
6import json
7import os
8import sqlite3
9import sys
10
11from atproto import CAR
12import redis
13import dag_cbor
14import websockets
15
16app_bsky_allowlist = set([
17 'app.bsky.actor.profile',
18 'app.bsky.feed.generator',
19 'app.bsky.feed.like',
20 'app.bsky.feed.post',
21 'app.bsky.feed.repost',
22 'app.bsky.feed.threadgate',
23 'app.bsky.graph.block',
24 'app.bsky.graph.follow',
25 'app.bsky.graph.list',
26 'app.bsky.graph.listblock',
27 'app.bsky.graph.listitem',
28 'app.bsky.labeler.service',
29 'chat.bsky.actor.declaration',
30])
31
32async def bsky_activity():
33 relay_url = 'ws://localhost:6008/subscribe'
34
35 sys.stdout.write(f'opening websocket connection to {relay_url}\n')
36 sys.stdout.flush()
37
38 async with websockets.connect(relay_url, ping_timeout=60) as firehose:
39 while True:
40 payload = BytesIO(await firehose.recv())
41
42 yield json.load(payload)
43
44async def main():
45 redis_cnx = redis.Redis()
46 redis_pipe = redis_cnx.pipeline()
47
48 if os.path.exists('/opt/muninsky/users.db'):
49 db_fname = '/opt/muninsky/users.db'
50 else:
51 db_fname = 'users.db'
52
53 db_cnx = sqlite3.connect(db_fname)
54 with db_cnx:
55 db_cnx.executescript("""
56 PRAGMA journal_mode = WAL;
57 PRAGMA synchronous = off;
58 CREATE TABLE IF NOT EXISTS users (did TEXT, ts TIMESTAMP);
59 CREATE UNIQUE INDEX IF NOT EXISTS did_idx on users(did);
60 CREATE INDEX IF NOT EXISTS ts_idx on users(ts);
61 """)
62
63 sys.stdout.write('starting up\n')
64 sys.stdout.flush()
65
66 op_count = 0
67 async for payload in bsky_activity():
68 if payload['opType'] != 'c':
69 continue
70
71 collection = payload['collection']
72 if collection not in app_bsky_allowlist:
73 continue
74
75 repo_did = payload['did']
76 repo_update_time = datetime.now(timezone.utc)
77 db_cnx.execute(
78 'insert into users values (:did, :ts) on conflict (did) do update set ts = :ts',
79 {'did': repo_did, 'ts': repo_update_time.timestamp()}
80 )
81
82 redis_pipe \
83 .incr(collection) \
84 .incr('dev.edavis.muninsky.ops')
85
86 op_count += 1
87 if op_count % 500 == 0:
88 now = datetime.now(timezone.utc)
89 payload_seq = payload['seq']
90 payload_lag = now - repo_update_time
91
92 sys.stdout.write(f'seq: {payload_seq}, lag: {payload_lag.total_seconds()}\n')
93 redis_pipe.set('dev.edavis.muninsky.seq', payload_seq)
94 redis_pipe.execute()
95 db_cnx.commit()
96 sys.stdout.flush()
97
98if __name__ == '__main__':
99 asyncio.run(main())