atproto_firehose load_dids: don't load disabled ATProto users

for #1669

atproto_firehose.py

@@ -84,7 +84,8 @@ def _load_dids():
         if not DEBUG:
             Timer(STORE_CURSOR_FREQ.total_seconds(), _load_dids).start()
 
-        atproto_query = ATProto.query(ATProto.enabled_protocols != None,
+        atproto_query = ATProto.query(ATProto.status == None,
+                                      ATProto.enabled_protocols != None,
                                       ATProto.updated > atproto_loaded_at)
         loaded_at = ATProto.query().order(-ATProto.updated).get().updated
         new_atproto = [key.id() for key in atproto_query.iter(keys_only=True)]

index.yaml

@@ -85,6 +85,12 @@ indexes:
   - name: updated
   - name: enabled_protocols
 
+- kind: ATProto
+  properties:
+  - name: status
+  - name: enabled_protocols
+  - name: updated
+
 - kind: AtpRepo
   properties:
     - name: status

tests/test_atproto_firehose.py

@@ -430,6 +430,17 @@ class ATProtoFirehoseSubscribeTest(ATProtoTestCase):
         self.assert_enqueues({'$type': 'app.bsky.feed.post'}, repo='did:plc:eve')
         self.assertIn('did:plc:eve', atproto_firehose.atproto_dids)
 
+    def test_load_dids_disabled_atproto_user(self):
+        self.cursor.cursor = 1
+        self.cursor.put()
+
+        self.store_object(id='did:plc:eve', raw=DID_DOC)
+        eve = self.make_user('did:plc:eve', cls=ATProto, enabled_protocols=['efake'],
+                             manual_opt_out=True)
+
+        self.subscribe()
+        self.assertNotIn('did:plc:eve', atproto_firehose.atproto_dids)
+
     def test_load_dids_atprepo(self):
         FakeWebsocketClient.to_receive = [({'op': 1, 't': '#info'}, {})]
         self.subscribe()