Feature: Add cleanup strategy for old database triggers #433

state-manager/app/tasks/trigger_cron.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 from uuid import uuid4
 from app.models.db.trigger import DatabaseTriggers
 from app.models.trigger_models import TriggerStatusEnum, TriggerTypeEnum
@@ -14,6 +14,9 @@
 logger = LogsManager().get_logger()
 
 async def get_due_triggers(cron_time: datetime) -> DatabaseTriggers | None:
+    """
+    Fetch a trigger that is due and mark it as TRIGGERING.
+    """
     data = await DatabaseTriggers.get_pymongo_collection().find_one_and_update(
         {
             "trigger_time": {"$lte": cron_time},
@@ -26,23 +29,38 @@ async def get_due_triggers(cron_time: datetime) -> DatabaseTriggers | None:
     )
     return DatabaseTriggers(**data) if data else None
 
+
 async def call_trigger_graph(trigger: DatabaseTriggers):
+    """
+    Call the associated graph for a trigger.
+    """
     await trigger_graph(
         namespace_name=trigger.namespace,
         graph_name=trigger.graph_name,
         body=TriggerGraphRequestModel(),
         x_exosphere_request_id=str(uuid4())
     )
 
+
 async def mark_as_failed(trigger: DatabaseTriggers):
+    """
+    Mark a trigger as FAILED.
+    """
     await DatabaseTriggers.get_pymongo_collection().update_one(
         {"_id": trigger.id},
         {"$set": {"trigger_status": TriggerStatusEnum.FAILED}}
     )
 
+
 async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime):
-    assert trigger.expression is not None
+    """
+    Create the next scheduled triggers based on the cron expression.
+    """
+    if not trigger.expression:
+        return
+
     iter = croniter.croniter(trigger.expression, trigger.trigger_time)
+    ttl_days = getattr(get_settings(), "trigger_ttl_days", 30)  # default 30 days
 
     while True:
         next_trigger_time = iter.get_next(datetime)
@@ -54,8 +72,10 @@ async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime):
                 graph_name=trigger.graph_name,
                 namespace=trigger.namespace,
                 trigger_time=next_trigger_time,
-                trigger_status=TriggerStatusEnum.PENDING
+                trigger_status=TriggerStatusEnum.PENDING,
+                expires_at=datetime.utcnow() + timedelta(days=ttl_days)
             ).insert()
+
         except DuplicateKeyError:
             logger.error(f"Duplicate trigger found for expression {trigger.expression}")
         except Exception as e:
@@ -65,14 +85,22 @@ async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime):
         if next_trigger_time > cron_time:
             break
 
+
 async def mark_as_triggered(trigger: DatabaseTriggers):
+    """
+    Mark a trigger as TRIGGERED.
+    """
     await DatabaseTriggers.get_pymongo_collection().update_one(
         {"_id": trigger.id},
         {"$set": {"trigger_status": TriggerStatusEnum.TRIGGERED}}
     )
 
+
 async def handle_trigger(cron_time: datetime):
-    while(trigger:= await get_due_triggers(cron_time)):
+    """
+    Handle due triggers one by one.
+    """
+    while (trigger := await get_due_triggers(cron_time)):
         try:
             await call_trigger_graph(trigger)
             await mark_as_triggered(trigger)
@@ -82,7 +110,12 @@ async def handle_trigger(cron_time: datetime):
         finally:
             await create_next_triggers(trigger, cron_time)
 
+
 async def trigger_cron():
-    cron_time = datetime.now()
-    logger.info(f"starting trigger_cron: {cron_time}")
-    await asyncio.gather(*[handle_trigger(cron_time) for _ in range(get_settings().trigger_workers)])
+    """
+    Main loop for cron trigger handling.
+    """
+    cron_time = datetime.utcnow()
+    logger.info(f"Starting trigger_cron: {cron_time}")
+    workers = getattr(get_settings(), "trigger_workers", 1)
+    await asyncio.gather(*[handle_trigger(cron_time) for _ in range(workers)])

state_manager/.dockerignore

@@ -0,0 +1,32 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*.pyo
+*.pyd
+*.so
+
+# C extensions
+*.egg
+*.egg-info
+.eggs
+
+# Virtual environments
+.venv/
+
+# Git, CI and editor config
+.git/
+.github/
+.vscode/
+
+# Logs and temp files
+*.log
+
+# Other
+.env
+Dockerfile 
+tests/
+pytest.ini
+.pytest_cache/
+.coverage
+.coverage.*
+coverage.xml

state_manager/.gitignore

@@ -0,0 +1,66 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Virtual Environment
+venv/
+env/
+ENV/
+.env
+.venv/
+
+# IDE
+.vscode/
+*.swp
+*.swo
+.idea/
+*.iws
+*.iml
+*.ipr
+
+
+# Local development
+.env.local
+.env.development.local
+.env.test.local
+.env.production.local
+
+# Database
+*.db
+*.sqlite3
+
+# OS generated files
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db
+
+#logs 
+*.log
+logs/*.*
+!logs/.gitkeep
+
+# local files
+files/
+!files/.gitkeep

state_manager/.python-version

@@ -0,0 +1 @@
+3.12

state_manager/Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.12-slim-bookworm
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+WORKDIR /api-server
+
+COPY pyproject.toml uv.lock ./
+
+RUN uv sync --locked
+
+COPY . .
+
+EXPOSE 8000
+
+CMD ["uv", "run", "run.py", "--mode", "production", "--workers", "4"]

state_manager/app/config/__init__.py

@@ -0,0 +1 @@
+# Config module for State Manager

state_manager/app/config/cors.py

@@ -0,0 +1,47 @@
+"""
+CORS configuration for the State Manager API
+"""
+import os
+from typing import List
+
+def get_cors_origins() -> List[str]:
+    """
+    Get CORS origins from environment variables or use defaults
+    """
+    # Get origins from environment variable
+    cors_origins = os.getenv("CORS_ORIGINS", "")
+
+    if cors_origins:
+        # Split by comma and strip whitespace
+        return [origin.strip() for origin in cors_origins.split(",") if origin.strip()]
+
+    # Default origins for development
+    return [
+        "http://localhost:3000",  # Next.js frontend
+        "http://localhost:3001",  # Alternative frontend port
+        "http://127.0.0.1:3000",  # Alternative localhost
+        "http://127.0.0.1:3001",  # Alternative localhost port
+    ]
+
+def get_cors_config():
+    """
+    Get CORS configuration
+    """
+    return {
+        "allow_origins": get_cors_origins(),
+        "allow_credentials": True,
+        "allow_methods": ["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"],
+        "allow_headers": [
+            "Accept",
+            "Accept-Language",
+            "Content-Language",
+            "Content-Type",
+            "X-API-Key",
+            "Authorization",
+            "X-Requested-With",
+            "X-Exosphere-Request-ID",
+        ],
+        "expose_headers": [  
+            "X-Exosphere-Request-ID",  
+        ],
+    }

state_manager/app/config/settings.py

@@ -0,0 +1,48 @@
+import os
+from pydantic import BaseModel, Field
+from dotenv import load_dotenv
+
+load_dotenv()
+
+class Settings(BaseModel):
+    """Application settings loaded from environment variables."""
+
+    # MongoDB Configuration
+    mongo_uri: str = Field(..., description="MongoDB connection URI")
+    mongo_database_name: str = Field(default="exosphere-state-manager", description="MongoDB database name")
+    state_manager_secret: str = Field(..., description="Secret key for API authentication")
+    secrets_encryption_key: str = Field(..., description="Key for encrypting secrets")
+    trigger_workers: int = Field(default=1, description="Number of workers to run the trigger cron")
+
+    # Cleanup / Retention Configs
+    trigger_retention_days: int = Field(default=30, description="How many days to retain old triggers")
+    cleanup_interval_minutes: int = Field(default=60, description="Interval (minutes) between cleanup runs")
+
+    @classmethod
+    def from_env(cls) -> "Settings":
+        return cls(
+            mongo_uri=os.getenv("MONGO_URI"),  # type: ignore
+            mongo_database_name=os.getenv("MONGO_DATABASE_NAME", "exosphere-state-manager"),  # type: ignore
+            state_manager_secret=os.getenv("STATE_MANAGER_SECRET"),  # type: ignore
+            secrets_encryption_key=os.getenv("SECRETS_ENCRYPTION_KEY"),  # type: ignore
+            trigger_workers=int(os.getenv("TRIGGER_WORKERS", 1)),  # type: ignore
+
+            # NEW CONFIGS
+            trigger_retention_days=int(os.getenv("TRIGGER_RETENTION_DAYS", 30)),  # type: ignore
+            cleanup_interval_minutes=int(os.getenv("CLEANUP_INTERVAL_MINUTES", 60))  # type: ignore
+        )
+
+
+# Global settings instance - will be updated when get_settings() is called
+_settings = None
+
+
+def get_settings() -> Settings:
+    """Get the global settings instance, reloading from environment if needed."""
+    global _settings
+    _settings = Settings.from_env()
+    return _settings
+
+
+# Initialize settings
+settings = get_settings()

state_manager/app/config/test_settings.py

@@ -0,0 +1,13 @@
+from app.config.settings import get_settings
+
+def test_settings():
+    settings = get_settings()
+    print("✅ Loaded Settings:")
+    print("Mongo URI:", settings.mongo_uri)
+    print("Database:", settings.mongo_database_name)
+    print("Trigger Workers:", settings.trigger_workers)
+    print("Trigger Retention Days:", settings.trigger_retention_days)
+    print("Cleanup Interval (minutes):", settings.cleanup_interval_minutes)
+
+if __name__ == "__main__":
+    test_settings()