SIENTIAPDE-2011: Compute query-time levels union / ignore_models_id intersection

Push level and model-exclusion filtering down to the Mongo query instead of
fetching the full notification_queue unfiltered on every run.

- Fix await ordering in load_notification_package so sending_configs (active
  receiver_groups) resolves before load_latest_data is called, letting the
  query filter be built from group data.
- Add _compute_query_time_filter: unions active groups' `levels` (falls back
  to the workflow's default filter if any group has no restriction, since
  that group wants every level) and intersects their `ignore_models_id`
  (a model is only safe to drop from the fetch if every active group
  ignores it).
- Add unit tests covering mixed groups, a model ignored by only one group
  (must not be excluded from the fetch), and the empty active-groups case.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-09-14 11:46:19 -03:00
parent ff7b441140
commit 1711f3b0d1
2 changed files with 113 additions and 5 deletions

View File

@@ -68,18 +68,23 @@ class LoadNotificationPackage:
# Load notification package from collection "notification_queue", using a
# static filter
sending_configs = await sending_configs_handler
base_data_filter = self._compute_query_time_filter(
sending_configs, input_data['base_data_filter']
)
notification_package = await workflow.start_local_activity_method(
Activities.load_latest_data,
{
**metadata,
'collection_name': 'notification_queue',
'last_data_timestamp': last_timestamp,
'base_data_filter': input_data['base_data_filter'],
'base_data_filter': base_data_filter,
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy,
)
sending_configs = await sending_configs_handler
if not sending_configs or not notification_package:
return {
@@ -106,3 +111,42 @@ class LoadNotificationPackage:
'notification_package': notification_package,
'sending_configs': sending_configs,
}
def _compute_query_time_filter(
self, sending_configs: list[dict[str, Any]], default_filter: dict[str, Any]
) -> dict[str, Any]:
"""
Compute the Mongo query-time filter from all active receiver groups.
- levels: union of every group's `levels`. If ANY active group has no
restriction (missing/empty `levels`), that group wants every level, so
the query can't restrict levels at all -> falls back to whatever level
restriction `default_filter` itself carries (e.g. ERROR-only for Alerts).
- ignore_models_id: intersection of every group's `ignore_models_id`. A model
is only safe to drop from the fetch if EVERY active group ignores it.
A group with no `ignore_models_id` is treated as "ignores nothing" (empty
set), which naturally zeroes out the intersection via set math.
"""
if not sending_configs:
return default_filter
sending_configs_levels = [group.get('levels') for group in sending_configs]
if all(sending_configs_levels):
levels_union: set[str] = set()
for levels in sending_configs_levels:
if levels:
levels_union.update(levels)
query_filter = {**default_filter, 'level': {'$in': sorted(levels_union)}}
else:
query_filter = dict(default_filter)
sending_configs_ignore_models_id = [
set(group.get('ignore_models_id', [])) for group in sending_configs
]
models_intersection = set.intersection(*sending_configs_ignore_models_id)
if models_intersection:
query_filter['model_id'] = {'$nin': sorted(models_intersection)}
return query_filter