Improve event filters to reject earlier (#89337)

* Improve event filters to reject earlier

- Avoid running the callbacks for state added/removed from
  a domain if there are no listeners that care about the domain

- Remove some impossible checks in the listeners that will
  never match since they were already rejected by the filter

* leave one guard since there is a race when we return control via await

homeassistant/helpers/event.py

@@ -412,18 +412,21 @@ def async_track_entity_registry_updated_event(
     return remove_listener
 
 
+@callback
+def _async_domain_has_listeners(
+    domain: str, callbacks: dict[str, list[HassJob[[Event], Any]]]
+) -> bool:
+    """Check if the domain has any listeners."""
+    return domain in callbacks or MATCH_ALL in callbacks
+
+
 @callback
 def _async_dispatch_domain_event(
     hass: HomeAssistant, event: Event, callbacks: dict[str, list[HassJob[[Event], Any]]]
 ) -> None:
+    """Dispatch domain event listeners."""
     domain = split_entity_id(event.data["entity_id"])[0]
-
+    for job in callbacks.get(domain, []) + callbacks.get(MATCH_ALL, []):
-    if domain not in callbacks and MATCH_ALL not in callbacks:
-        return
-
-    listeners = callbacks.get(domain, []) + callbacks.get(MATCH_ALL, [])
-
-    for job in listeners:
         try:
             hass.async_run_hass_job(job, event)
         except Exception:  # pylint: disable=broad-except
@@ -460,14 +463,13 @@ def _async_track_state_added_domain(
         @callback
         def _async_state_change_filter(event: Event) -> bool:
             """Filter state changes by entity_id."""
-            return event.data.get("old_state") is None
+            return event.data.get("old_state") is None and _async_domain_has_listeners(
+                split_entity_id(event.data["entity_id"])[0], domain_callbacks
+            )
 
         @callback
         def _async_state_change_dispatcher(event: Event) -> None:
             """Dispatch state changes by entity_id."""
-            if event.data.get("old_state") is not None:
-                return
-
             _async_dispatch_domain_event(hass, event, domain_callbacks)
 
         hass.data[TRACK_STATE_ADDED_DOMAIN_LISTENER] = hass.bus.async_listen(
@@ -514,14 +516,13 @@ def async_track_state_removed_domain(
         @callback
         def _async_state_change_filter(event: Event) -> bool:
             """Filter state changes by entity_id."""
-            return event.data.get("new_state") is None
+            return event.data.get("new_state") is None and _async_domain_has_listeners(
+                split_entity_id(event.data["entity_id"])[0], domain_callbacks
+            )
 
         @callback
         def _async_state_change_dispatcher(event: Event) -> None:
             """Dispatch state changes by entity_id."""
-            if event.data.get("new_state") is not None:
-                return
-
             _async_dispatch_domain_event(hass, event, domain_callbacks)
 
         hass.data[TRACK_STATE_REMOVED_DOMAIN_LISTENER] = hass.bus.async_listen(