feat: sending email when workflow fails (#588)

Author: shahargl

.gitignore

@@ -177,7 +177,7 @@ e1faa321-35df-486b-8fa8-3601ee714011*
 # sqlite db
 *.sqlite3
 state/*
-
+.terraform*
 examples/alerts/dd.yml
 keep-ui/node_modules
 keep-ui/node_modules/*

keep/api/core/db.py

@@ -684,6 +684,11 @@ def get_api_key(api_key: str):
     return tenant_api_key
 
 
+def get_user_by_api_key(api_key: str):
+    api_key = get_api_key(api_key)
+    return api_key.created_by
+
+
 # this is only for single tenant
 def get_user(username, password, update_sign_in=True):
     from keep.api.core.dependencies import SINGLE_TENANT_UUID
@@ -770,3 +775,19 @@ def get_workflow_id_by_name(tenant_id, workflow_name):
 
         if workflow:
             return workflow.id
+
+
+def get_previous_execution_id(tenant_id, workflow_id, workflow_execution_id):
+    with Session(engine) as session:
+        previous_execution = session.exec(
+            select(WorkflowExecution)
+            .where(WorkflowExecution.tenant_id == tenant_id)
+            .where(WorkflowExecution.workflow_id == workflow_id)
+            .where(WorkflowExecution.id != workflow_execution_id)
+            .order_by(WorkflowExecution.started.desc())
+            .limit(1)
+        ).first()
+        if previous_execution:
+            return previous_execution
+        else:
+            return None

keep/api/core/dependencies.py

@@ -15,7 +15,7 @@
 from sqlmodel import Session
 
 from keep.api.core.config import AuthenticationType
-from keep.api.core.db import get_api_key, get_session
+from keep.api.core.db import get_api_key, get_session, get_user_by_api_key
 
 logger = logging.getLogger(__name__)
 
@@ -39,7 +39,8 @@ def get_user_email(request: Request) -> str | None:
         decoded_token = jwt.decode(token, options={"verify_signature": False})
         return decoded_token.get("email")
     elif "x-api-key" in request.headers:
-        return "apikey@keephq.dev"
+        username = get_user_by_api_key(request.headers["x-api-key"])
+        return username
     else:
         raise HTTPException(
             status_code=401, detail="Invalid authentication credentials"

keep/api/models/db/tenant.py

@@ -18,6 +18,7 @@ class TenantApiKey(SQLModel, table=True):
     tenant: Tenant = Relationship()
     is_system: bool = False
     system_description: Optional[str] = None
+    created_by: str
 
     class Config:
         orm_mode = True

keep/api/routes/alerts.py

@@ -7,6 +7,7 @@
 from pusher import Pusher
 from sqlmodel import Session
 
+from keep.api.core.config import config
 from keep.api.core.db import enrich_alert as enrich_alert_db
 from keep.api.core.db import get_alerts as get_alerts_from_db
 from keep.api.core.db import get_enrichments as get_enrichments_from_db
@@ -20,6 +21,7 @@
 )
 from keep.api.models.alert import AlertDto, DeleteRequestBody, EnrichAlertRequestBody
 from keep.api.models.db.alert import Alert
+from keep.api.utils.email_utils import EmailTemplates, send_email
 from keep.contextmanager.contextmanager import ContextManager
 from keep.providers.providers_factory import ProvidersFactory
 from keep.workflowmanager.workflowmanager import WorkflowManager
@@ -300,6 +302,29 @@ def assign_alert(
         enrichments={"assignee": user_email if not unassign else None},
     )
 
+    try:
+        logger.info("Sending assign alert email to user")
+        # TODO: this should be changed to dynamic url but we don't know what's the frontend URL
+        keep_platform_url = config(
+            "KEEP_PLATFORM_URL", default="https://platform.keephq.dev"
+        )
+        url = f"{keep_platform_url}/alerts?fingerprint={fingerprint}"
+        send_email(
+            to_email=user_email,
+            template_id=EmailTemplates.ALERT_ASSIGNED_TO_USER,
+            url=url,
+        )
+        logger.info("Sent assign alert email to user")
+    except Exception as e:
+        logger.exception(
+            "Failed to send email to user",
+            extra={
+                "error": str(e),
+                "tenant_id": tenant_id,
+                "user_email": user_email,
+            },
+        )
+
     logger.info(
         "Assigned alert successfully",
         extra={

keep/api/routes/providers.py

@@ -675,6 +675,7 @@ def install_provider_webhook(
     webhook_api_key = get_or_create_api_key(
         session=session,
         tenant_id=tenant_id,
+        created_by="system",
         unique_api_key_id="webhook",
         system_description="Webhooks API key",
     )
@@ -702,6 +703,7 @@ def get_webhook_settings(
     webhook_api_key = get_or_create_api_key(
         session=session,
         tenant_id=tenant_id,
+        created_by="system",
         unique_api_key_id="webhook",
         system_description="Webhooks API key",
     )

keep/api/routes/settings.py

@@ -15,7 +15,7 @@
 from keep.api.core.db import delete_user as delete_user_from_db
 from keep.api.core.db import get_session
 from keep.api.core.db import get_users as get_users_from_db
-from keep.api.core.dependencies import verify_bearer_token
+from keep.api.core.dependencies import get_user_email, verify_bearer_token
 from keep.api.models.alert import AlertDto
 from keep.api.models.smtp import SMTPSettings
 from keep.api.models.user import User
@@ -41,6 +41,7 @@ def webhook_settings(
     webhook_api_key = get_or_create_api_key(
         session=session,
         tenant_id=tenant_id,
+        created_by="system",
         unique_api_key_id="webhook",
         system_description="Webhooks API key",
     )
@@ -275,11 +276,13 @@ def _print_debug(self, *args):
 def get_api_key(
     tenant_id: str = Depends(verify_bearer_token),
     session: Session = Depends(get_session),
+    user_name: str = Depends(get_user_email),
 ):
     # get the api key for the CLI
     api_key = get_or_create_api_key(
         session=session,
         tenant_id=tenant_id,
+        created_by=user_name,
         unique_api_key_id="cli",
         system_description="API key",
     )

keep/api/utils/email_utils.py

@@ -0,0 +1,62 @@
+import enum
+import logging
+
+from sendgrid import SendGridAPIClient
+from sendgrid.helpers.mail import Mail
+
+from keep.api.core.config import config
+
+# TODO
+# This is beta code. It will be changed in the future.
+
+# Sending emails is currently done via SendGrid. It doesnt fit well with OSS, so we need to:
+# 1. Add EmailManager that will support more than just SendGrid
+# 2. Add support for templates/html
+# 3. Add support for SMTP (how will it work with templates?)
+
+
+# In the OSS - you can overwrite the template ids
+class EmailTemplates(enum.Enum):
+    WORKFLOW_RUN_FAILED = config(
+        "WORKFLOW_FAILED_EMAIL_TEMPLATE_ID",
+        default="d-bb1b3bb30ce8460cbe6ed008701affb1",
+    )
+    ALERT_ASSIGNED_TO_USER = config(
+        "ALERT_ASSIGNED_TO_USER_EMAIL_TEMPLATE_ID",
+        default="d-58ec64ed781e4c359e18da7ad97ac750",
+    )
+
+
+logger = logging.getLogger(__name__)
+
+# CONSTS
+FROM_EMAIL = config("SENDGRID_FROM_EMAIL", default="platform@keephq.dev")
+API_KEY = config("SENDGRID_API_KEY", default=None)
+CC = config("SENDGRID_CC", default="founders@keephq.dev")
+
+
+def send_email(
+    to_email: str,
+    template_id: EmailTemplates,
+    **kwargs,
+):
+    # that's ok on OSS
+    if not API_KEY:
+        logger.debug("No SendGrid API key, skipping sending email")
+        return
+
+    message = Mail(from_email=FROM_EMAIL, to_emails=to_email)
+    message.template_id = template_id.value
+    # TODO: validate the kwargs and the template parameters are the same
+    message.dynamic_template_data = kwargs
+    # send the email
+    try:
+        logger.info(f"Sending email to {to_email} with template {template_id}")
+        sg = SendGridAPIClient(API_KEY)
+        sg.send(message)
+        logger.info(f"Email sent to {to_email} with template {template_id}")
+    except Exception as e:
+        logger.error(
+            f"Failed to send email to {to_email} with template {template_id}: {e}"
+        )
+        raise

keep/api/utils/tenant_utils.py

@@ -17,6 +17,7 @@ def create_api_key(
     tenant_id: str,
     unique_api_key_id: str,
     is_system: bool,
+    created_by: str,
     commit: bool = True,
     system_description: Optional[str] = None,
 ) -> str:
@@ -54,6 +55,7 @@ def create_api_key(
         key_hash=hashed_api_key,
         is_system=is_system,
         system_description=system_description,
+        created_by=created_by,
     )
     session.add(new_installation_api_key)
     if commit:
@@ -68,6 +70,7 @@ def create_api_key(
 def get_or_create_api_key(
     session: Session,
     tenant_id: str,
+    created_by: str,
     unique_api_key_id: str,
     system_description: Optional[str] = None,
 ) -> str:
@@ -98,6 +101,7 @@ def get_or_create_api_key(
             session,
             tenant_id,
             unique_api_key_id,
+            created_by=created_by,
             is_system=True,
             system_description=system_description,
         )

keep/contextmanager/contextmanager.py

@@ -56,6 +56,7 @@ def api_key(self):
             session = next(get_session())
             self._api_key = get_or_create_api_key(
                 session=session,
+                created_by="system",
                 tenant_id=self.tenant_id,
                 unique_api_key_id="webhook",
             )