chore: remove old event stream from codebase

https://issues.redhat.com/browse/AAP-27632
Remove the old EventStream from fanout

src/aap_eda/api/constants.py

@@ -14,36 +14,4 @@
 
 # EDA_SERVER_VAULT_LABEL is reserved for system vault password identifiers
 EDA_SERVER_VAULT_LABEL = "EDA_SERVER"
-
-PG_NOTIFY_TEMPLATE_RULEBOOK_NAME = "_PG_NOTIFY_TEMPLATE_RULEBOOK_"
-PG_NOTIFY_TEMPLATE_RULEBOOK_DATA = """
----
-- name: PG Notify Template Event Stream
-  hosts: all
-  sources:
-    - name: my_range
-      ansible.eda.range:
-        limit: 5
-      complementary_source:
-        type: ansible.eda.pg_listener
-        name: Postgres Listener
-        args:
-          dsn: "{{ EDA_PG_NOTIFY_DSN }}"
-          channels:
-            - "{{ EDA_PG_NOTIFY_CHANNEL }}"
-      extra_vars:
-        EDA_PG_NOTIFY_DSN: "{{ settings.PG_NOTIFY_DSN }}"
-        EDA_PG_NOTIFY_CHANNEL: "{{ event_stream.channel_name }}"
-      encrypt_vars:
-        - EDA_PG_NOTIFY_DSN
-  rules:
-    - name: Post event
-      condition: true
-      action:
-        pg_notify:
-          dsn: "{{ EDA_PG_NOTIFY_DSN }}"
-          channel: "{{ EDA_PG_NOTIFY_CHANNEL }}"
-          event: "{{ event }}"
-"""
-
 SOURCE_MAPPING_ERROR_KEY = "rulebook"

src/aap_eda/api/exceptions.py

@@ -97,36 +97,6 @@ class InvalidWebsocketHost(APIException):
     )
 
 
-class MissingEventStreamRulebook(APIException):
-    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
-    default_detail = (
-        "Configuration Error: Event stream template rulebook not found"
-    )
-
-
-class MissingEventStreamRulebookKeys(APIException):
-    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
-    default_detail = (
-        "Configuration Error: Event stream template rulebook is missing "
-        "required keys in complementary_source: type, name and args"
-    )
-
-
-class MissingEventStreamRulebookSource(APIException):
-    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
-    default_detail = (
-        "Configuration Error: Event stream template rulebook is missing "
-        "required complementary_source"
-    )
-
-
-class InvalidEventStreamRulebook(APIException):
-    status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
-    default_detail = (
-        "Configuration Error: Event stream template rulebook is invalid"
-    )
-
-
 class InvalidWebhookSource(APIException):
     status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
     default_detail = (

src/aap_eda/api/filters/__init__.py

@@ -20,7 +20,6 @@
 from .credential_type import CredentialTypeFilter
 from .decision_environment import DecisionEnvironmentFilter
 from .eda_credential import EdaCredentialFilter
-from .event_stream import EventStreamFilter
 from .organization import OrganizationFilter
 from .project import ProjectFilter
 from .rulebook import (
@@ -52,8 +51,6 @@
     "ActivationInstanceLogFilter",
     # user
     "UserFilter",
-    # event_stream
-    "EventStreamFilter",
     # organization
     "OrganizationFilter",
     # team

src/aap_eda/api/serializers/__init__.py

@@ -38,11 +38,6 @@
     EdaCredentialCreateSerializer,
     EdaCredentialSerializer,
 )
-from .event_stream import (
-    EventStreamCreateSerializer,
-    EventStreamOutSerializer,
-    EventStreamSerializer,
-)
 from .organization import (
     OrganizationCreateSerializer,
     OrganizationRefSerializer,
@@ -123,10 +118,6 @@
     "EdaCredentialCreateSerializer",
     # decision environment
     "DecisionEnvironmentSerializer",
-    # event streams
-    "EventStreamSerializer",
-    "EventStreamCreateSerializer",
-    "EventStreamOutSerializer",
     # organizations
     "OrganizationSerializer",
     "OrganizationCreateSerializer",

src/aap_eda/api/serializers/activation.py

@@ -24,18 +24,13 @@
 
 from aap_eda.api.constants import (
     EDA_SERVER_VAULT_LABEL,
-    PG_NOTIFY_TEMPLATE_RULEBOOK_DATA,
     SOURCE_MAPPING_ERROR_KEY,
 )
-from aap_eda.api.exceptions import (
-    InvalidEventStreamRulebook,
-    InvalidWebhookSource,
-)
+from aap_eda.api.exceptions import InvalidWebhookSource
 from aap_eda.api.serializers.decision_environment import (
     DecisionEnvironmentRefSerializer,
 )
 from aap_eda.api.serializers.eda_credential import EdaCredentialSerializer
-from aap_eda.api.serializers.event_stream import EventStreamOutSerializer
 from aap_eda.api.serializers.organization import OrganizationRefSerializer
 from aap_eda.api.serializers.project import (
     ANSIBLE_VAULT_STRING,
@@ -58,12 +53,7 @@
     get_rulebook_hash,
     swap_webhook_sources,
 )
-from aap_eda.core.utils.strings import (
-    substitute_extra_vars,
-    substitute_source_args,
-    substitute_variables,
-    swap_sources,
-)
+from aap_eda.core.utils.strings import substitute_variables
 
 logger = logging.getLogger(__name__)
 REQUIRED_KEYS = ["webhook_id", "webhook_name", "source_name", "rulebook_hash"]
@@ -106,43 +96,6 @@ def _update_webhook_source(validated_data: dict, vault_data: VaultData) -> str:
         raise InvalidWebhookSource(e) from e
 
 
-def _updated_ruleset(validated_data: dict, vault_data: VaultData):
-    try:
-        sources_info = []
-
-        for event_stream_id in validated_data["event_streams"]:
-            event_stream = models.EventStream.objects.get(id=event_stream_id)
-
-            if event_stream.rulebook:
-                rulesets = yaml.safe_load(event_stream.rulebook.rulesets)
-            else:
-                rulesets = yaml.safe_load(PG_NOTIFY_TEMPLATE_RULEBOOK_DATA)
-
-            extra_vars = rulesets[0]["sources"][0].get("extra_vars", {})
-            encrypt_vars = rulesets[0]["sources"][0].get("encrypt_vars", [])
-
-            if bool(encrypt_vars):
-                vault_data.password_used = True
-
-            extra_vars = substitute_extra_vars(
-                event_stream.__dict__,
-                extra_vars,
-                encrypt_vars,
-                vault_data.password,
-            )
-
-            source = rulesets[0]["sources"][0]["complementary_source"]
-            source = substitute_source_args(
-                event_stream.__dict__, source, extra_vars
-            )
-            sources_info.append(source)
-
-        return swap_sources(validated_data["rulebook_rulesets"], sources_info)
-    except Exception as e:
-        logger.error(f"Failed to update rulesets: {e}")
-        raise InvalidEventStreamRulebook(e)
-
-
 def _update_k8s_service_name(validated_data: dict) -> str:
     service_name = validated_data.get("k8s_service_name")
     return service_name or create_k8s_service_name(validated_data["name"])
@@ -246,12 +199,6 @@ def replace_vault_data(extra_var):
 class ActivationSerializer(serializers.ModelSerializer):
     """Serializer for the Activation model."""
 
-    event_streams = serializers.ListField(
-        required=False,
-        allow_null=True,
-        child=EventStreamOutSerializer(),
-    )
-
     eda_credentials = serializers.ListField(
         required=False,
         allow_null=True,
@@ -287,7 +234,6 @@ class Meta:
             "modified_at",
             "status_message",
             "awx_token_id",
-            "event_streams",
             "eda_credentials",
             "log_level",
             "webhooks",
@@ -307,11 +253,6 @@ class ActivationListSerializer(serializers.ModelSerializer):
     rules_count = serializers.IntegerField()
     rules_fired_count = serializers.IntegerField()
 
-    event_streams = serializers.ListField(
-        required=False,
-        allow_null=True,
-        child=EventStreamOutSerializer(),
-    )
     eda_credentials = serializers.ListField(
         required=False,
         allow_null=True,
@@ -352,7 +293,6 @@ class Meta:
             "modified_at",
             "status_message",
             "awx_token_id",
-            "event_streams",
             "log_level",
             "eda_credentials",
             "k8s_service_name",
@@ -366,10 +306,6 @@ def to_representation(self, activation):
         rules_count, rules_fired_count = get_rules_count(
             activation.ruleset_stats
         )
-        event_streams = [
-            EventStreamOutSerializer(event_stream).data
-            for event_stream in activation.event_streams.all()
-        ]
         eda_credentials = [
             EdaCredentialSerializer(credential).data
             for credential in activation.eda_credentials.all()
@@ -405,7 +341,6 @@ def to_representation(self, activation):
             "modified_at": activation.modified_at,
             "status_message": activation.status_message,
             "awx_token_id": activation.awx_token_id,
-            "event_streams": event_streams,
             "log_level": activation.log_level,
             "eda_credentials": eda_credentials,
             "k8s_service_name": activation.k8s_service_name,
@@ -431,7 +366,6 @@ class Meta:
             "user",
             "restart_policy",
             "awx_token_id",
-            "event_streams",
             "log_level",
             "eda_credentials",
             "k8s_service_name",
@@ -463,12 +397,6 @@ class Meta:
         validators=[validators.check_if_awx_token_exists],
         required=False,
     )
-    event_streams = serializers.ListField(
-        required=False,
-        allow_null=True,
-        child=serializers.IntegerField(),
-        validators=[validators.check_if_event_streams_exists],
-    )
     eda_credentials = serializers.ListField(
         required=False,
         allow_null=True,
@@ -503,11 +431,6 @@ def create(self, validated_data):
 
         vault_data = VaultData()
 
-        if validated_data.get("event_streams"):
-            validated_data["rulebook_rulesets"] = _updated_ruleset(
-                validated_data, vault_data
-            )
-
         if validated_data.get("source_mappings", []):
             validated_data["rulebook_rulesets"] = _update_webhook_source(
                 validated_data, vault_data
@@ -555,7 +478,6 @@ class Meta:
             "git_hash",
             "status_message",
             "activation_id",
-            "event_stream_id",
             "organization_id",
             "started_at",
             "ended_at",
@@ -586,11 +508,6 @@ class ActivationReadSerializer(serializers.ModelSerializer):
     rules_count = serializers.IntegerField()
     rules_fired_count = serializers.IntegerField()
     restarted_at = serializers.DateTimeField(required=False, allow_null=True)
-    event_streams = serializers.ListField(
-        required=False,
-        allow_null=True,
-        child=EventStreamOutSerializer(),
-    )
     eda_credentials = serializers.ListField(
         required=False,
         allow_null=True,
@@ -635,7 +552,6 @@ class Meta:
             "status_message",
             "awx_token_id",
             "eda_credentials",
-            "event_streams",
             "log_level",
             "k8s_service_name",
             "webhooks",
@@ -682,10 +598,6 @@ def to_representation(self, activation):
             if activation.organization
             else None
         )
-        event_streams = [
-            EventStreamOutSerializer(event_stream).data
-            for event_stream in activation.event_streams.all()
-        ]
         eda_credentials = [
             EdaCredentialSerializer(credential).data
             for credential in activation.eda_credentials.all()
@@ -726,7 +638,6 @@ def to_representation(self, activation):
             "restarted_at": restarted_at,
             "status_message": activation.status_message,
             "awx_token_id": activation.awx_token_id,
-            "event_streams": event_streams,
             "log_level": activation.log_level,
             "eda_credentials": eda_credentials,
             "k8s_service_name": activation.k8s_service_name,