diff --git a/keep/api/core/db.py b/keep/api/core/db.py
index 0d1c0e3c5..b63f68820 100644
--- a/keep/api/core/db.py
+++ b/keep/api/core/db.py
@@ -3248,6 +3248,71 @@ def remove_alerts_to_incident_by_incident_id(
         return deleted
 
 
+class DestinationIncidentNotFound(Exception):
+    pass
+
+
+def merge_incidents_to_id(
+    tenant_id: str,
+    source_incident_ids: List[UUID],
+    destination_incident_id: UUID,
+    merged_by: str | None = None,
+) -> Optional[Incident]:
+    with Session(engine) as session:
+        destination_incident = session.exec(
+            select(Incident)
+            .where(
+                Incident.tenant_id == tenant_id, Incident.id == destination_incident_id
+            )
+            .options(joinedload(Incident.alerts))
+        ).first()
+
+        if not destination_incident:
+            # TODO: maybe allow to create a new incident if the destination incident does not exist
+            raise DestinationIncidentNotFound(
+                f"Destination incident with id {destination_incident_id} not found"
+            )
+
+        source_incidents = session.exec(
+            select(Incident).filter(
+                Incident.tenant_id == tenant_id,
+                Incident.id.in_(source_incident_ids),
+            )
+        ).all()
+
+        alerts_to_add_ids = []
+        for source_incident in source_incidents:
+            alerts_to_add_ids.extend([alert.id for alert in source_incident.alerts])
+            source_incident.merged_into_id = destination_incident.id
+            source_incident.merged_at = datetime.now(tz=timezone.utc)
+            source_incident.status = IncidentStatus.MERGED.value
+            source_incident.merged_by = merged_by
+            try:
+                # TODO: optimize, process in bulk
+                remove_alerts_to_incident_by_incident_id(
+                    tenant_id,
+                    source_incident.id,
+                    [alert.id for alert in source_incident.alerts],
+                )
+            except OperationalError as e:
+                logger.error(
+                    f"Error removing alerts to incident {source_incident.id}: {e}"
+                )
+
+        try:
+            add_alerts_to_incident(
+                tenant_id, destination_incident, alerts_to_add_ids, session=session
+            )
+        except OperationalError as e:
+            logger.error(
+                f"Error adding alerts to incident {destination_incident.id}: {e}"
+            )
+
+        session.commit()
+        session.refresh(destination_incident)
+        return destination_incident
+
+
 def get_alerts_count(
     tenant_id: str,
 ) -> int:
diff --git a/keep/api/models/alert.py b/keep/api/models/alert.py
index c49a89bfb..910f16573 100644
--- a/keep/api/models/alert.py
+++ b/keep/api/models/alert.py
@@ -108,6 +108,8 @@ class IncidentStatus(Enum):
     RESOLVED = "resolved"
     # Incident has been acknowledged but not resolved
     ACKNOWLEDGED = "acknowledged"
+    # Incident was merged with another incident
+    MERGED = "merged"
 
 
 class IncidentSeverity(SeverityBaseInterface):
@@ -490,6 +492,11 @@ def from_db_incident(cls, db_incident):
         return dto
 
 
+class MergeIncidentsCommandDto(BaseModel):
+    source_incident_ids: list[UUID]
+    destination_incident_id: UUID
+
+
 class DeduplicationRuleDto(BaseModel):
     id: str | None  # UUID
     name: str
diff --git a/keep/api/models/db/alert.py b/keep/api/models/db/alert.py
index b46690734..5f05c8245 100644
--- a/keep/api/models/db/alert.py
+++ b/keep/api/models/db/alert.py
@@ -126,7 +126,7 @@ class Incident(SQLModel, table=True):
     rule_id: UUID | None = Field(
         sa_column=Column(
             UUIDType(binary=False),
-            ForeignKey("rule.id", use_alter=False, ondelete="CASCADE"),
+            ForeignKey("rule.id", ondelete="CASCADE"),
             nullable=True,
         ),
     )
@@ -137,7 +137,7 @@ class Incident(SQLModel, table=True):
     same_incident_in_the_past_id: UUID | None = Field(
         sa_column=Column(
             UUIDType(binary=False),
-            ForeignKey("incident.id", use_alter=False, ondelete="SET NULL"),
+            ForeignKey("incident.id", ondelete="SET NULL"),
             nullable=True,
         ),
     )
@@ -146,11 +146,38 @@ class Incident(SQLModel, table=True):
         back_populates="same_incidents_in_the_future",
         sa_relationship_kwargs=dict(
             remote_side="Incident.id",
+            foreign_keys="[Incident.same_incident_in_the_past_id]",
         ),
     )
 
-    same_incidents_in_the_future: list["Incident"] = Relationship(
+    same_incidents_in_the_future: List["Incident"] = Relationship(
         back_populates="same_incident_in_the_past",
+        sa_relationship_kwargs=dict(
+            foreign_keys="[Incident.same_incident_in_the_past_id]",
+        ),
+    )
+
+    merged_into_id: UUID | None = Field(
+        sa_column=Column(
+            UUIDType(binary=False),
+            ForeignKey("incident.id", ondelete="SET NULL"),
+            nullable=True,
+        ),
+    )
+    merged_at: datetime | None = Field(default=None)
+    merged_by: str | None = Field(default=None)
+    merged_into: Optional["Incident"] = Relationship(
+        back_populates="merged_incidents",
+        sa_relationship_kwargs=dict(
+            remote_side="Incident.id",
+            foreign_keys="[Incident.merged_into_id]",
+        ),
+    )
+    merged_incidents: List["Incident"] = Relationship(
+        back_populates="merged_into",
+        sa_relationship_kwargs=dict(
+            foreign_keys="[Incident.merged_into_id]",
+        ),
     )
 
     def __init__(self, **kwargs):
diff --git a/keep/api/models/db/migrations/versions/2024-10-21-20-48_89b4d3905d26.py b/keep/api/models/db/migrations/versions/2024-10-21-20-48_89b4d3905d26.py
new file mode 100644
index 000000000..f601ebb2b
--- /dev/null
+++ b/keep/api/models/db/migrations/versions/2024-10-21-20-48_89b4d3905d26.py
@@ -0,0 +1,44 @@
+"""Merge Incidents
+
+Revision ID: 89b4d3905d26
+Revises: 83c1020be97d
+Create Date: 2024-10-21 20:48:40.151171
+
+"""
+
+import sqlalchemy as sa
+import sqlalchemy_utils
+import sqlmodel
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "89b4d3905d26"
+down_revision = "83c1020be97d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("incident", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "merged_into_id",
+                sqlalchemy_utils.types.uuid.UUIDType(binary=False),
+                nullable=True,
+            )
+        )
+        batch_op.add_column(sa.Column("merged_at", sa.DateTime(), nullable=True))
+        batch_op.add_column(
+            sa.Column("merged_by", sqlmodel.sql.sqltypes.AutoString(), nullable=True)
+        )
+        batch_op.create_foreign_key(
+            None, "incident", ["merged_into_id"], ["id"], ondelete="SET NULL"
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("incident", schema=None) as batch_op:
+        batch_op.drop_constraint(None, type_="foreignkey")
+        batch_op.drop_column("merged_by")
+        batch_op.drop_column("merged_at")
+        batch_op.drop_column("merged_into_id")
diff --git a/keep/api/routes/incidents.py b/keep/api/routes/incidents.py
index bf2921bd8..0ab3c2ca1 100644
--- a/keep/api/routes/incidents.py
+++ b/keep/api/routes/incidents.py
@@ -28,6 +28,9 @@
     get_workflow_executions_for_incident_or_alert,
     remove_alerts_to_incident_by_incident_id,
     update_incident_from_dto_by_id,
+    get_incidents_meta_for_tenant,
+    merge_incidents_to_id,
+    DestinationIncidentNotFound,
 )
 from keep.api.core.dependencies import get_pusher_client
 from keep.api.core.elastic import ElasticClient
@@ -37,6 +40,7 @@
     IncidentDto,
     IncidentDtoIn,
     IncidentListFilterParamsDto,
+    MergeIncidentsCommandDto,
     IncidentSeverity,
     IncidentSorting,
     IncidentStatus,
@@ -348,6 +352,36 @@ def delete_incident(
     return Response(status_code=202)
 
 
+@router.post("/merge", description="Merge incidents", response_model=IncidentDto)
+def merge_incidents(
+    command: MergeIncidentsCommandDto,
+    authenticated_entity: AuthenticatedEntity = Depends(
+        IdentityManagerFactory.get_auth_verifier(["write:incident"])
+    ),
+) -> Response:
+    tenant_id = authenticated_entity.tenant_id
+    logger.info(
+        "Merging incidents",
+        extra={
+            "source_incident_ids": command.source_incident_ids,
+            "destination_incident_id": command.destination_incident_id,
+            "tenant_id": tenant_id,
+        },
+    )
+
+    try:
+        updated_incident = merge_incidents_to_id(
+            tenant_id,
+            command.source_incident_ids,
+            command.destination_incident_id,
+            authenticated_entity.email,
+        )
+        updated_incident_dto = IncidentDto.from_db_incident(updated_incident)
+        return updated_incident_dto
+    except DestinationIncidentNotFound as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
 @router.get(
     "/{incident_id}/alerts",
     description="Get incident alerts by incident incident id",