chore: add helper to sanitize pg identifiers (#1295)

For dispatcherd we need to sanitize the channel name as we already do
for event streams. There is no need to fill the code with `.replace()`
calls.
Instead create a helper that can sanitize the name in a more reusable
and reliable way.

---------

Signed-off-by: Alex <[email protected]>

Author: Alex-Izquierdo

src/aap_eda/core/models/event_stream.py

@@ -16,6 +16,8 @@
 
 from django.db import models
 
+from aap_eda.utils import sanitize_postgres_identifier
+
 from .base import BaseOrgModel, PrimordialModel, UniqueNamedModel
 
 __all__ = "EventStream"
@@ -92,9 +94,7 @@ class EventStream(BaseOrgModel, UniqueNamedModel, PrimordialModel):
 
     def _get_channel_name(self) -> str:
         """Generate the channel name based on the UUID and prefix."""
-        return (
-            f"{EDA_EVENT_STREAM_CHANNEL_PREFIX}"
-            f"{str(self.uuid).replace('-','_')}"
-        )
+        channel_name = f"{EDA_EVENT_STREAM_CHANNEL_PREFIX}{str(self.uuid)}"
+        return sanitize_postgres_identifier(channel_name)
 
     channel_name = property(_get_channel_name)

src/aap_eda/utils/__init__.py

@@ -13,6 +13,8 @@
 #  limitations under the License.
 import importlib.metadata
 import logging
+import re
+from functools import cache
 
 logger = logging.getLogger(__name__)
 
@@ -32,3 +34,39 @@ def get_package_version(package_name: str) -> str:
             package_name,
         )
         return "unknown"
+
+
+@cache
+def sanitize_postgres_identifier(identifier: str) -> str:
+    """
+    Sanitize an input string to conform to PostgreSQL identifier rules.
+
+    Initially intended to be used for pg_notify channel names.
+    """
+    max_identifier_length = 63
+    if not identifier:
+        raise ValueError("Identifier cannot be empty.")
+
+    if len(identifier) > max_identifier_length:
+        raise ValueError(
+            f"Identifier exceeds {max_identifier_length} characters."
+        )
+
+    sanitized = identifier
+
+    # Ensure it starts with a valid character
+    if not re.match(r"[A-Za-z_]", identifier[0]):
+        if len(identifier) == max_identifier_length:
+            raise ValueError(
+                f"Identifier has invalid first character and "
+                f"{max_identifier_length} characters. "
+                "It can not be sanitized to a valid "
+                "PostgreSQL identifier below "
+                f"{max_identifier_length} characters."
+            )
+        sanitized = f"_{identifier}"
+
+    # Replace invalid characters with underscores
+    sanitized = re.sub(r"\W", "_", sanitized)
+
+    return sanitized

tests/unit/test_utils.py

@@ -25,6 +25,7 @@
 from aap_eda.utils import (
     get_package_version,
     logger as utils_logger,
+    sanitize_postgres_identifier,
     str_to_bool,
 )
 from aap_eda.utils.openapi import generate_query_params
@@ -174,3 +175,41 @@ def test_generate_query_params(serializer, expected_params):
 
     for param, expected in zip(query_params, expected_params):
         assert param.__dict__ == expected.__dict__
+
+
+@pytest.mark.parametrize(
+    "input_str,expected_output",
+    [
+        ("valid_name", "valid_name"),
+        ("Valid123", "Valid123"),
+        ("_underscore", "_underscore"),
+        ("hello-world", "hello_world"),
+        ("bad@name!", "bad_name_"),
+        ("some space", "some_space"),
+        ("123name", "_123name"),
+        ("9abc", "_9abc"),
+        ("@@@", "____"),
+        ("123", "_123"),
+        ("a" * 63, "a" * 63),
+        (("1" + "a" * 61), ("_1" + "a" * 61)),
+    ],
+)
+def test_sanitize_postgres_identifier_valid_cases(input_str, expected_output):
+    assert sanitize_postgres_identifier(input_str) == expected_output
+
+
+def test_empty_identifier_raises():
+    with pytest.raises(ValueError, match="Identifier cannot be empty."):
+        sanitize_postgres_identifier("")
+
+
+def test_identifier_exceeding_length_limit_raises():
+    too_long = "a" * 64
+    with pytest.raises(ValueError, match="exceeds 63 characters"):
+        sanitize_postgres_identifier(too_long)
+
+
+def test_identifier_with_invalid_first_character_raises():
+    too_long = "1" + "a" * 62
+    with pytest.raises(ValueError, match="invalid first character"):
+        sanitize_postgres_identifier(too_long)