Fix `have_seen_event` cache not being invalidated

Fix https://github.com/matrix-org/synapse/issues/13856

`_invalidate_caches_for_event` doesn't run in monolith mode which
means we never even tried to clear the `have_seen_event` and other
caches. And even in worker mode, it only runs on the workers, not
the master (AFAICT).

Additionally there is bug with the key being wrong so
`_invalidate_caches_for_event` never invalidates the
`have_seen_event` cache even when it does run.

Wrong:
```py
self.have_seen_event.invalidate((room_id, event_id))
```

Correct:
```py
self.have_seen_event.invalidate(((room_id, event_id),))
```

2 files changed, 34 insertions, 2 deletions

diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py
index 501dbbc990..094628ec18 100644
--- a/synapse/storage/controllers/persist_events.py
+++ b/synapse/storage/controllers/persist_events.py
@@ -43,7 +43,7 @@ from prometheus_client import Counter, Histogram
 from twisted.internet import defer
 
 from synapse.api.constants import EventTypes, Membership
-from synapse.events import EventBase
+from synapse.events import EventBase, relation_from_event
 from synapse.events.snapshot import EventContext
 from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable
 from synapse.logging.opentracing import (
@@ -431,6 +431,22 @@ class EventsPersistenceStorageController:
             else:
                 events.append(event)
 
+                # We expect events to be persisted by this point and this makes
+                # mypy happy about `stream_ordering` not being optional below
+                assert event.internal_metadata.stream_ordering
+                # Invalidate related caches after we persist a new event
+                relation = relation_from_event(event)
+                self.main_store._invalidate_caches_for_event(
+                    stream_ordering=event.internal_metadata.stream_ordering,
+                    event_id=event.event_id,
+                    room_id=event.room_id,
+                    etype=event.type,
+                    state_key=event.state_key if hasattr(event, "state_key") else None,
+                    redacts=event.redacts,
+                    relates_to=relation.parent_id if relation else None,
+                    backfilled=backfilled,
+                )
+
         return (
             events,
             self.main_store.get_room_max_token(),
@@ -463,6 +479,22 @@ class EventsPersistenceStorageController:
         replaced_event = replaced_events.get(event.event_id)
         if replaced_event:
             event = await self.main_store.get_event(replaced_event)
+        else:
+            # We expect events to be persisted by this point and this makes
+            # mypy happy about `stream_ordering` not being optional below
+            assert event.internal_metadata.stream_ordering
+            # Invalidate related caches after we persist a new event
+            relation = relation_from_event(event)
+            self.main_store._invalidate_caches_for_event(
+                stream_ordering=event.internal_metadata.stream_ordering,
+                event_id=event.event_id,
+                room_id=event.room_id,
+                etype=event.type,
+                state_key=event.state_key if hasattr(event, "state_key") else None,
+                redacts=event.redacts,
+                relates_to=relation.parent_id if relation else None,
+                backfilled=backfilled,
+            )
 
         event_stream_id = event.internal_metadata.stream_ordering
         # stream ordering should have been assigned by now
diff --git a/synapse/storage/databases/main/cache.py b/synapse/storage/databases/main/cache.py
index 2c421151c1..2dbf93a4e5 100644
--- a/synapse/storage/databases/main/cache.py
+++ b/synapse/storage/databases/main/cache.py
@@ -223,7 +223,7 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
         # process triggering the invalidation is responsible for clearing any external
         # cached objects.
         self._invalidate_local_get_event_cache(event_id)
-        self.have_seen_event.invalidate((room_id, event_id))
+        self.have_seen_event.invalidate(((room_id, event_id),))
 
         self.get_latest_event_ids_in_room.invalidate((room_id,))