Enable mypy for the appservice handler and fix types

mypy.ini

@@ -15,6 +15,7 @@ files =
   synapse/events/builder.py,
   synapse/events/spamcheck.py,
   synapse/federation,
+  synapse/handlers/appservice.py,
   synapse/handlers/account_data.py,
   synapse/handlers/auth.py,
   synapse/handlers/cas_handler.py,

synapse/handlers/appservice.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 import logging
-from typing import List, Union
+from typing import Dict, List, Optional
 
 from prometheus_client import Counter
 
@@ -23,14 +23,15 @@
 import synapse
 from synapse.api.constants import EventTypes
 from synapse.appservice import ApplicationService
+from synapse.events import EventBase
 from synapse.handlers.presence import format_user_presence_state
 from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.metrics import (
     event_processing_loop_counter,
     event_processing_loop_room_count,
 )
 from synapse.metrics.background_process_metrics import run_as_background_process
-from synapse.types import Collection, RoomStreamToken, UserID
+from synapse.types import Collection, JsonDict, RoomStreamToken, UserID
 from synapse.util.metrics import Measure
 
 logger = logging.getLogger(__name__)
@@ -86,7 +87,7 @@ async def notify_interested_services(self, max_token: RoomStreamToken):
                     if not events:
                         break
 
-                    events_by_room = {}
+                    events_by_room: Dict[str, List[EventBase]] = {}
                     for event in events:
                         events_by_room.setdefault(event.room_id, []).append(event)
 
@@ -166,10 +167,7 @@ async def handle_room_events(events):
                 self.is_processing = False
 
     async def notify_interested_services_ephemeral(
-        self,
-        stream_key: str,
-        new_token: Union[int, RoomStreamToken],
-        users: Collection[UserID] = [],
+        self, stream_key: str, new_token: Optional[int], users: Collection[UserID] = [],
     ):
         """This is called by the notifier in the background
         when a ephemeral event handled by the homeserver.
@@ -195,7 +193,8 @@ async def notify_interested_services_ephemeral(
         logger.info("Checking interested services for %s" % (stream_key))
         with Measure(self.clock, "notify_interested_services_ephemeral"):
             for service in services:
-                if stream_key == "typing_key":
+                # Only handle typing if we have the latest token
+                if stream_key == "typing_key" and new_token is not None:
                     events = await self._handle_typing(service, new_token)
                     if events:
                         self.scheduler.submit_ephemeral_events_for_as(service, events)
@@ -223,7 +222,7 @@ async def _handle_typing(self, service: ApplicationService, new_token: int):
             # For performance reasons, we don't persist the previous
             # token in the DB and instead fetch the latest typing information
             # for appservices.
-            from_key=new_token - 1,
+            from_key=int(new_token) - 1,
         )
         return typing
 
@@ -237,8 +236,10 @@ async def _handle_receipts(self, service: ApplicationService):
         )
         return receipts
 
-    async def _handle_presence(self, service: ApplicationService, users: List[str]):
-        events = []
+    async def _handle_presence(
+        self, service: ApplicationService, users: Collection[UserID]
+    ):
+        events: List[JsonDict] = []
         presence_source = self.event_sources.sources["presence"]
         from_key = await self.store.get_type_stream_id_for_appservice(
             service, "presence"
@@ -325,7 +326,7 @@ async def query_3pe(self, kind, protocol, fields):
 
     async def get_3pe_protocols(self, only_protocol=None):
         services = self.store.get_app_services()
-        protocols = {}
+        protocols: Dict[str, List[JsonDict]] = {}
 
         # Collect up all the individual protocol responses out of the ASes
         for s in services:

synapse/notifier.py

@@ -336,8 +336,11 @@ async def _notify_app_services_ephemeral(
         users: Collection[UserID] = [],
     ):
         try:
+            stream_token = None
+            if isinstance(new_token, int):
+                stream_token = new_token
             await self.appservice_handler.notify_interested_services_ephemeral(
-                stream_key, new_token, users
+                stream_key, stream_token, users
             )
         except Exception:
             logger.exception("Error notifying application services of event")