feeds: cutover to jetstream

feedgen.py

···
 
 import asyncio
 from io import BytesIO
+import json
 import logging
 
 from atproto import CAR
···
 logging.getLogger('firehose').setLevel(logging.DEBUG)
 
 async def firehose_events(firehose_manager):
-    relay_url = 'wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos'
-    seq = firehose_manager.get_sequence_number()
-    if seq is not None:
-        relay_url += f'?cursor={seq}'
+    relay_url = 'ws://localhost:6008/subscribe'
 
     logger = logging.getLogger('feeds.events')
     logger.info(f'opening websocket connection to {relay_url}')
 
     async with websockets.connect(relay_url, ping_timeout=60) as firehose:
         while True:
-            frame = BytesIO(await firehose.recv())
-            header = dag_cbor.decode(frame, allow_concat=True)
-            if header['op'] != 1 or header['t'] != '#commit':
-                continue
-
-            payload = dag_cbor.decode(frame)
-            if payload['tooBig']:
-                continue
-
-            blocks = payload.pop('blocks')
-            car_parsed = CAR.from_bytes(blocks)
-            message = payload.copy()
-            del message['ops']
-            message['commit'] = message['commit'].encode('base32')
-
-            for op in payload['ops']:
-                repo_op = op.copy()
-                if op['cid'] is not None:
-                    repo_op['cid'] = repo_op['cid'].encode('base32')
-                    repo_op['record'] = car_parsed.blocks.get(repo_op['cid'])
-
-                message['op'] = repo_op
-                yield message
+            payload = BytesIO(await firehose.recv())
+            yield json.load(payload)
 
 async def main():
     firehose_manager = FirehoseManager()

feeds/battle.py

···
         self.logger = logging.getLogger('feeds.battle')
 
     def process_commit(self, commit):
-        op = commit['op']
-        if op['action'] != 'create':
+        if commit['opType'] != 'c':
             return
 
-        collection, _ = op['path'].split('/')
-        if collection != 'app.bsky.feed.post':
+        if commit['collection'] != 'app.bsky.feed.post':
             return
 
-        record = op.get('record')
+        record = commit.get('record')
         if record is None:
             return
 
-        repo = commit['repo']
-        path = op['path']
-        post_uri = f'at://{repo}/{path}'
+        repo = commit['did']
+        rkey = commit['rkey']
+        post_uri = f'at://{repo}/app.bsky.feed.post/{rkey}'
         length = grapheme.length(record.get('text', ''))
         ts = self.safe_timestamp(record.get('createdAt')).timestamp()
 

feeds/homeruns.py

···
         self.logger = logging.getLogger('feeds.homeruns')
 
     def process_commit(self, commit):
-        if commit['repo'] != MLBHRS_DID:
+        if commit['did'] != MLBHRS_DID:
             return
 
-        op = commit['op']
-        if op['action'] != 'create':
+        if commit['opType'] != 'c':
             return
 
-        collection, _ = op['path'].split('/')
-        if collection != 'app.bsky.feed.post':
+        if commit['collection'] != 'app.bsky.feed.post':
             return
 
-        record = op.get('record')
+        record = commit.get('record')
         if record is None:
             return
 
-        uri = 'at://{repo}/{path}'.format(
-            repo = commit['repo'],
-            path = op['path']
+        uri = 'at://{repo}/app.bsky.feed.post/{rkey}'.format(
+            repo = commit['did'],
+            rkey = commit['rkey']
         )
         tags = record.get('tags', [])
 

feeds/popular.py

···
         self.logger = logging.getLogger('feeds.popular')
 
     def process_commit(self, commit):
-        op = commit['op']
-        if op['action'] != 'create':
+        if commit['opType'] != 'c':
             return
 
-        collection, _ = op['path'].split('/')
-        if collection != 'app.bsky.feed.like':
+        if commit['collection'] != 'app.bsky.feed.like':
             return
 
-        record = op.get('record')
+        record = commit.get('record')
         if record is None:
             return
 
         ts = self.safe_timestamp(record.get('createdAt')).timestamp()
-        like_subject_uri = op['record']['subject']['uri']
+        like_subject_uri = record['subject']['uri']
 
         self.transaction_begin(self.db_cnx)
 

feeds/rapidfire.py

···
         self.logger = logging.getLogger('feeds.rapidfire')
 
     def process_commit(self, commit):
-        op = commit['op']
-        if op['action'] != 'create':
+        if commit['opType'] != 'c':
             return
 
-        collection, _ = op['path'].split('/')
-        if collection != 'app.bsky.feed.post':
+        if commit['collection'] != 'app.bsky.feed.post':
             return
 
-        record = op.get('record')
+        record = commit.get('record')
         if record is None:
             return
 
···
             record.get('embed') is None,
             record.get('facets') is None
         ]):
-            repo = commit['repo']
-            path = op['path']
-            post_uri = f'at://{repo}/{path}'
+            repo = commit['did']
+            rkey = commit['rkey']
+            post_uri = f'at://{repo}/app.bsky.feed.post/{rkey}'
             ts = self.safe_timestamp(record.get('createdAt')).timestamp()
 
             self.transaction_begin(self.db_cnx)