models: Extend EventHistory model for durable pub/sub delivery

kernelci/api/models.py

@@ -16,6 +16,7 @@
 from datetime import datetime, timedelta
 from typing import Any, Optional, Dict, List, ClassVar, Literal
 import enum
+import os
 from operator import attrgetter
 import json
 from typing_extensions import Annotated
@@ -439,7 +440,7 @@
         default='checkout',
         description='Type of the object',
     )
     data: CheckoutData = Field(
         description="Checkout details",
         default=None
     )
@@ -514,7 +515,7 @@
         default='kbuild',
         description='Type of the object',
     )
     data: KbuildData = Field(
         description="Kbuild details",
         default=None
     )
@@ -609,7 +610,7 @@
         default='test',
         description='Type of the object',
     )
     data: TestData = Field(
         description="Test details",
         default=None
     )
@@ -626,7 +627,7 @@
         default='job',
         description='Type of the object',
     )
     data: TestData = Field(
         description="Test suite details",
         default=None
     )
@@ -721,7 +722,7 @@
                      "FAIL if the regression is still 'active', ie. the test "
                      "is still failing"),
     )
     data: RegressionData = Field(
         description="Regression details",
         default=None
     )
@@ -855,12 +856,29 @@
 
 
 # eventhistory model
+# Environment variable to configure TTL (default 7 days = 604800 seconds)
+# Set EVENT_HISTORY_TTL_SECONDS to override
+EVENT_HISTORY_TTL_SECONDS = int(os.getenv('EVENT_HISTORY_TTL_SECONDS', '604800'))
+
+
 class EventHistory(DatabaseModel):
     """Event history object model"""
     timestamp: datetime = Field(
         description='Timestamp of event creation',
         default_factory=datetime.now
     )
+    sequence_id: Optional[int] = Field(
+        default=None,
+        description='Sequential ID for pub/sub ordering (auto-generated)'
+    )
+    channel: Optional[str] = Field(
+        default='node',
+        description='Pub/Sub channel name'
+    )
+    owner: Optional[str] = Field(
+        default=None,
+        description='Username of event publisher'
+    )
     data: Dict[str, Any] = Field(
         description='Event data',
         default={}
@@ -871,9 +889,11 @@
         """
         Create the index with the expiresAfterSeconds option for the
         timestamp field to automatically remove documents after a certain
-        time.
-        Default is 86400 seconds (24 hours).
+        time. Default is 604800 seconds (7 days), configurable via
+        EVENT_HISTORY_TTL_SECONDS environment variable.
+        Also creates compound index for efficient pub/sub catch-up queries.
         """
         return [
-            cls.Index('timestamp', {'expireAfterSeconds': 86400}),
+            cls.Index('timestamp', {'expireAfterSeconds': EVENT_HISTORY_TTL_SECONDS}),
+            cls.Index([('channel', 1), ('sequence_id', 1)], {}),
         ]

kernelci/api/models_base.py

@@ -12,7 +12,7 @@
 
 """Common KernelCI API model definitions"""
 
-from typing import Optional, Any, Dict
+from typing import Optional, Any, Dict, List, Tuple, Union
 from bson import ObjectId
 from pydantic import (
     BaseModel,
@@ -74,8 +74,14 @@
     """Database model"""
     @dataclass
     class Index():
-        """Index class"""
-        field: str
+        """Index class
+
+        field: Either a single field name (str) for simple indexes,
+               or a list of (field, direction) tuples for compound indexes.
+               Direction is typically 1 (ascending) or -1 (descending).
+        attributes: MongoDB index options (e.g., {'expireAfterSeconds': 3600})
+        """
+        field: Union[str, List[Tuple[str, int]]]
         attributes: dict[str, Any]
 
     def update(self):
@@ -99,7 +105,7 @@
         # https://github.com/pydantic/pydantic/issues/6575
         if info.exclude:
             for field in info.exclude:
                 values.pop(field, None)
 
         for field_name, value in values.items():
             if isinstance(value, ObjectId):