diff --git a/x-pack/plugins/cases/kibana.json b/x-pack/plugins/cases/kibana.json
index 6d21904177a47..af7cdd7a99bed 100644
--- a/x-pack/plugins/cases/kibana.json
+++ b/x-pack/plugins/cases/kibana.json
@@ -31,7 +31,8 @@
      "triggersActionsUi",
      "management",
      "spaces",
-     "security"
+     "security",
+     "notifications"
   ],
   "requiredBundles": [
     "savedObjects"
diff --git a/x-pack/plugins/cases/server/attachment_framework/mocks.ts b/x-pack/plugins/cases/server/attachment_framework/mocks.ts
index d0c6fdf037f2b..58483b072416f 100644
--- a/x-pack/plugins/cases/server/attachment_framework/mocks.ts
+++ b/x-pack/plugins/cases/server/attachment_framework/mocks.ts
@@ -14,8 +14,13 @@ import type {
   CommentRequestPersistableStateType,
 } from '../../common/api';
 import { ExternalReferenceStorageType } from '../../common/api';
+import { ExternalReferenceAttachmentTypeRegistry } from './external_reference_registry';
 import { PersistableStateAttachmentTypeRegistry } from './persistable_state_registry';
-import type { PersistableStateAttachmentTypeSetup, PersistableStateAttachmentState } from './types';
+import type {
+  PersistableStateAttachmentTypeSetup,
+  PersistableStateAttachmentState,
+  ExternalReferenceAttachmentType,
+} from './types';
 
 export const getPersistableAttachment = (): PersistableStateAttachmentTypeSetup => ({
   id: '.test',
@@ -42,6 +47,10 @@ export const getPersistableAttachment = (): PersistableStateAttachmentTypeSetup
   }),
 });
 
+export const getExternalReferenceAttachment = (): ExternalReferenceAttachmentType => ({
+  id: '.test',
+});
+
 export const externalReferenceAttachmentSO = {
   type: CommentType.externalReference as const,
   externalReferenceId: 'my-id',
@@ -130,10 +139,18 @@ export const externalReferenceAttachmentSOAttributesWithoutRefs = omit(
   'externalReferenceId'
 );
 
-export const getPersistableStateAttachmentTypeRegistry =
+export const createPersistableStateAttachmentTypeRegistryMock =
   (): PersistableStateAttachmentTypeRegistry => {
     const persistableStateAttachmentTypeRegistry = new PersistableStateAttachmentTypeRegistry();
     persistableStateAttachmentTypeRegistry.register(getPersistableAttachment());
 
     return persistableStateAttachmentTypeRegistry;
   };
+
+export const createExternalReferenceAttachmentTypeRegistryMock =
+  (): ExternalReferenceAttachmentTypeRegistry => {
+    const externalReferenceAttachmentTypeRegistry = new ExternalReferenceAttachmentTypeRegistry();
+    externalReferenceAttachmentTypeRegistry.register(getExternalReferenceAttachment());
+
+    return externalReferenceAttachmentTypeRegistry;
+  };
diff --git a/x-pack/plugins/cases/server/attachment_framework/so_references.test.ts b/x-pack/plugins/cases/server/attachment_framework/so_references.test.ts
index cc22e907464cd..7ef8816934059 100644
--- a/x-pack/plugins/cases/server/attachment_framework/so_references.test.ts
+++ b/x-pack/plugins/cases/server/attachment_framework/so_references.test.ts
@@ -7,7 +7,7 @@
 
 import { CommentType, SECURITY_SOLUTION_OWNER } from '../../common';
 import {
-  getPersistableStateAttachmentTypeRegistry,
+  createPersistableStateAttachmentTypeRegistryMock,
   persistableStateAttachment,
   persistableStateAttachmentAttributes,
 } from './mocks';
@@ -19,7 +19,7 @@ import {
 } from './so_references';
 
 describe('Persistable state SO references', () => {
-  const persistableStateAttachmentTypeRegistry = getPersistableStateAttachmentTypeRegistry();
+  const persistableStateAttachmentTypeRegistry = createPersistableStateAttachmentTypeRegistryMock();
   const references = [
     {
       id: 'testRef',
diff --git a/x-pack/plugins/cases/server/client/cases/create.test.ts b/x-pack/plugins/cases/server/client/cases/create.test.ts
new file mode 100644
index 0000000000000..fe814be543e2e
--- /dev/null
+++ b/x-pack/plugins/cases/server/client/cases/create.test.ts
@@ -0,0 +1,80 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { SECURITY_SOLUTION_OWNER } from '../../../common';
+import { CaseSeverity, ConnectorTypes } from '../../../common/api';
+import { mockCases } from '../../mocks';
+import { createCasesClientMockArgs } from '../mocks';
+import { create } from './create';
+
+describe('create', () => {
+  const theCase = {
+    title: 'My Case',
+    tags: [],
+    description: 'testing sir',
+    connector: {
+      id: '.none',
+      name: 'None',
+      type: ConnectorTypes.none,
+      fields: null,
+    },
+    settings: { syncAlerts: true },
+    severity: CaseSeverity.LOW,
+    owner: SECURITY_SOLUTION_OWNER,
+    assignees: [{ uid: '1' }],
+  };
+
+  const caseSO = mockCases[0];
+
+  describe('Assignees', () => {
+    const clientArgs = createCasesClientMockArgs();
+    clientArgs.services.caseService.postNewCase.mockResolvedValue(caseSO);
+
+    beforeEach(() => {
+      jest.clearAllMocks();
+    });
+
+    it('notifies single assignees', async () => {
+      await create(theCase, clientArgs);
+
+      expect(clientArgs.services.notificationService.notifyAssignees).toHaveBeenCalledWith({
+        assignees: theCase.assignees,
+        theCase: caseSO,
+      });
+    });
+
+    it('notifies multiple assignees', async () => {
+      await create({ ...theCase, assignees: [{ uid: '1' }, { uid: '2' }] }, clientArgs);
+
+      expect(clientArgs.services.notificationService.notifyAssignees).toHaveBeenCalledWith({
+        assignees: [{ uid: '1' }, { uid: '2' }],
+        theCase: caseSO,
+      });
+    });
+
+    it('does not notify when there are no assignees', async () => {
+      await create({ ...theCase, assignees: [] }, clientArgs);
+
+      expect(clientArgs.services.notificationService.notifyAssignees).not.toHaveBeenCalled();
+    });
+
+    it('does not notify the current user', async () => {
+      await create(
+        {
+          ...theCase,
+          assignees: [{ uid: '1' }, { uid: 'u_J41Oh6L9ki-Vo2tOogS8WRTENzhHurGtRc87NgEAlkc_0' }],
+        },
+        clientArgs
+      );
+
+      expect(clientArgs.services.notificationService.notifyAssignees).toHaveBeenCalledWith({
+        assignees: [{ uid: '1' }],
+        theCase: caseSO,
+      });
+    });
+  });
+});
diff --git a/x-pack/plugins/cases/server/client/cases/create.ts b/x-pack/plugins/cases/server/client/cases/create.ts
index ba3da0eefe053..7a249400ccf8d 100644
--- a/x-pack/plugins/cases/server/client/cases/create.ts
+++ b/x-pack/plugins/cases/server/client/cases/create.ts
@@ -41,7 +41,7 @@ export const create = async (
 ): Promise<CaseResponse> => {
   const {
     unsecuredSavedObjectsClient,
-    services: { caseService, userActionService, licensingService },
+    services: { caseService, userActionService, licensingService, notificationService },
     user,
     logger,
     authorization: auth,
@@ -116,11 +116,22 @@ export const create = async (
       owner: newCase.attributes.owner,
     });
 
-    return CaseResponseRt.encode(
-      flattenCaseSavedObject({
-        savedObject: newCase,
-      })
-    );
+    const flattenedCase = flattenCaseSavedObject({
+      savedObject: newCase,
+    });
+
+    if (query.assignees && query.assignees.length !== 0) {
+      const assigneesWithoutCurrentUser = query.assignees.filter(
+        (assignee) => assignee.uid !== user.profile_uid
+      );
+
+      await notificationService.notifyAssignees({
+        assignees: assigneesWithoutCurrentUser,
+        theCase: newCase,
+      });
+    }
+
+    return CaseResponseRt.encode(flattenedCase);
   } catch (error) {
     throw createCaseError({ message: `Failed to create case: ${error}`, error, logger });
   }
diff --git a/x-pack/plugins/cases/server/client/cases/get.ts b/x-pack/plugins/cases/server/client/cases/get.ts
index 4f219db2c28f7..db836786b6db3 100644
--- a/x-pack/plugins/cases/server/client/cases/get.ts
+++ b/x-pack/plugins/cases/server/client/cases/get.ts
@@ -9,7 +9,7 @@ import { pipe } from 'fp-ts/lib/pipeable';
 import { fold } from 'fp-ts/lib/Either';
 import { identity } from 'fp-ts/lib/function';
 
-import type { SavedObject, SavedObjectsResolveResponse } from '@kbn/core/server';
+import type { SavedObjectsResolveResponse } from '@kbn/core/server';
 import type {
   CaseResponse,
   CaseResolveResponse,
@@ -37,6 +37,7 @@ import type { CasesClientArgs } from '..';
 import { Operations } from '../../authorization';
 import { combineAuthorizedAndOwnerFilter } from '../utils';
 import { CasesService } from '../../services';
+import type { CaseSavedObject } from '../../common/types';
 
 /**
  * Parameters for finding cases IDs using an alert ID
@@ -182,7 +183,7 @@ export const get = async (
   } = clientArgs;
 
   try {
-    const theCase: SavedObject<CaseAttributes> = await caseService.getCase({
+    const theCase: CaseSavedObject = await caseService.getCase({
       id,
     });
 
diff --git a/x-pack/plugins/cases/server/client/cases/update.test.ts b/x-pack/plugins/cases/server/client/cases/update.test.ts
new file mode 100644
index 0000000000000..ab80618903fa0
--- /dev/null
+++ b/x-pack/plugins/cases/server/client/cases/update.test.ts
@@ -0,0 +1,253 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { mockCases } from '../../mocks';
+import { createCasesClientMockArgs } from '../mocks';
+import { update } from './update';
+
+describe('update', () => {
+  const cases = {
+    cases: [
+      {
+        id: mockCases[0].id,
+        version: mockCases[0].version ?? '',
+        assignees: [{ uid: '1' }],
+      },
+    ],
+  };
+
+  describe('Assignees', () => {
+    const clientArgs = createCasesClientMockArgs();
+
+    beforeEach(() => {
+      jest.clearAllMocks();
+      clientArgs.services.caseService.getCases.mockResolvedValue({ saved_objects: mockCases });
+      clientArgs.services.caseService.getAllCaseComments.mockResolvedValue({
+        saved_objects: [],
+        total: 0,
+        per_page: 10,
+        page: 1,
+      });
+
+      clientArgs.services.caseService.patchCases.mockResolvedValue({
+        saved_objects: [{ ...mockCases[0], attributes: { assignees: cases.cases[0].assignees } }],
+      });
+    });
+
+    it('notifies an assignee', async () => {
+      await update(cases, clientArgs);
+
+      expect(clientArgs.services.notificationService.bulkNotifyAssignees).toHaveBeenCalledWith([
+        {
+          assignees: [{ uid: '1' }],
+          theCase: {
+            ...mockCases[0],
+            attributes: { ...mockCases[0].attributes, assignees: [{ uid: '1' }] },
+          },
+        },
+      ]);
+    });
+
+    it('does not notify if the case does not exist', async () => {
+      expect.assertions(2);
+
+      await expect(
+        update(
+          {
+            cases: [
+              {
+                id: 'not-exists',
+                version: '123',
+                assignees: [{ uid: '1' }],
+              },
+            ],
+          },
+          clientArgs
+        )
+      ).rejects.toThrow(
+        'Failed to update case, ids: [{"id":"not-exists","version":"123"}]: Error: These cases not-exists do not exist. Please check you have the correct ids.'
+      );
+
+      expect(clientArgs.services.notificationService.bulkNotifyAssignees).not.toHaveBeenCalled();
+    });
+
+    it('does not notify if the case is patched with the same assignee', async () => {
+      expect.assertions(2);
+
+      clientArgs.services.caseService.getCases.mockResolvedValue({
+        saved_objects: [
+          {
+            ...mockCases[0],
+            attributes: { ...mockCases[0].attributes, assignees: [{ uid: '1' }] },
+          },
+        ],
+      });
+
+      await expect(update(cases, clientArgs)).rejects.toThrow(
+        'Failed to update case, ids: [{"id":"mock-id-1","version":"WzAsMV0="}]: Error: All update fields are identical to current version.'
+      );
+
+      expect(clientArgs.services.notificationService.bulkNotifyAssignees).not.toHaveBeenCalled();
+    });
+
+    it('notifies only new users', async () => {
+      clientArgs.services.caseService.getCases.mockResolvedValue({
+        saved_objects: [
+          {
+            ...mockCases[0],
+            attributes: { ...mockCases[0].attributes, assignees: [{ uid: '1' }] },
+          },
+        ],
+      });
+
+      clientArgs.services.caseService.patchCases.mockResolvedValue({
+        saved_objects: [
+          {
+            ...mockCases[0],
+            attributes: { assignees: [{ uid: '1' }, { uid: '2' }, { uid: '3' }] },
+          },
+        ],
+      });
+
+      await update(
+        {
+          cases: [
+            {
+              id: mockCases[0].id,
+              version: mockCases[0].version ?? '',
+              assignees: [{ uid: '1' }, { uid: '2' }, { uid: '3' }],
+            },
+          ],
+        },
+        clientArgs
+      );
+
+      expect(clientArgs.services.notificationService.bulkNotifyAssignees).toHaveBeenCalledWith([
+        {
+          assignees: [{ uid: '2' }, { uid: '3' }],
+          theCase: {
+            ...mockCases[0],
+            attributes: {
+              ...mockCases[0].attributes,
+              assignees: [{ uid: '1' }, { uid: '2' }, { uid: '3' }],
+            },
+          },
+        },
+      ]);
+    });
+
+    it('does not notify when removing assignees', async () => {
+      clientArgs.services.caseService.getCases.mockResolvedValue({
+        saved_objects: [
+          {
+            ...mockCases[0],
+            attributes: { ...mockCases[0].attributes, assignees: [{ uid: '1' }, { uid: '2' }] },
+          },
+        ],
+      });
+
+      clientArgs.services.caseService.patchCases.mockResolvedValue({
+        saved_objects: [{ ...mockCases[0], attributes: { assignees: [{ uid: '1' }] } }],
+      });
+
+      await update(
+        {
+          cases: [
+            {
+              id: mockCases[0].id,
+              version: mockCases[0].version ?? '',
+              assignees: [{ uid: '1' }],
+            },
+          ],
+        },
+        clientArgs
+      );
+
+      expect(clientArgs.services.notificationService.bulkNotifyAssignees).toHaveBeenCalledWith([]);
+      expect(clientArgs.services.notificationService.notifyAssignees).not.toHaveBeenCalled();
+    });
+
+    it('does not notify the current user', async () => {
+      clientArgs.services.caseService.getCases.mockResolvedValue({
+        saved_objects: [
+          {
+            ...mockCases[0],
+            attributes: { ...mockCases[0].attributes, assignees: [{ uid: '1' }] },
+          },
+        ],
+      });
+
+      clientArgs.services.caseService.patchCases.mockResolvedValue({
+        saved_objects: [
+          {
+            ...mockCases[0],
+            attributes: {
+              assignees: [{ uid: '2' }, { uid: 'u_J41Oh6L9ki-Vo2tOogS8WRTENzhHurGtRc87NgEAlkc_0' }],
+            },
+          },
+        ],
+      });
+
+      await update(
+        {
+          cases: [
+            {
+              id: mockCases[0].id,
+              version: mockCases[0].version ?? '',
+              assignees: [{ uid: '2' }, { uid: 'u_J41Oh6L9ki-Vo2tOogS8WRTENzhHurGtRc87NgEAlkc_0' }],
+            },
+          ],
+        },
+        clientArgs
+      );
+
+      expect(clientArgs.services.notificationService.bulkNotifyAssignees).toHaveBeenCalledWith([
+        {
+          assignees: [{ uid: '2' }],
+          theCase: {
+            ...mockCases[0],
+            attributes: {
+              ...mockCases[0].attributes,
+              assignees: [{ uid: '2' }, { uid: 'u_J41Oh6L9ki-Vo2tOogS8WRTENzhHurGtRc87NgEAlkc_0' }],
+            },
+          },
+        },
+      ]);
+    });
+
+    it('does not notify when there are no new assignees', async () => {
+      clientArgs.services.caseService.getCases.mockResolvedValue({
+        saved_objects: [
+          {
+            ...mockCases[0],
+            attributes: { ...mockCases[0].attributes, assignees: [{ uid: '1' }] },
+          },
+        ],
+      });
+
+      await update(
+        {
+          cases: [
+            {
+              id: mockCases[0].id,
+              version: mockCases[0].version ?? '',
+              assignees: [{ uid: '1' }, { uid: 'u_J41Oh6L9ki-Vo2tOogS8WRTENzhHurGtRc87NgEAlkc_0' }],
+            },
+          ],
+        },
+        clientArgs
+      );
+
+      /**
+       * Current user is filtered out. Assignee with uid=1 should not be
+       * notified because it was already assigned to the case.
+       */
+      expect(clientArgs.services.notificationService.bulkNotifyAssignees).toHaveBeenCalledWith([]);
+      expect(clientArgs.services.notificationService.notifyAssignees).not.toHaveBeenCalled();
+    });
+  });
+});
diff --git a/x-pack/plugins/cases/server/client/cases/update.ts b/x-pack/plugins/cases/server/client/cases/update.ts
index 1a6f8301ec93f..3d6c595a55e03 100644
--- a/x-pack/plugins/cases/server/client/cases/update.ts
+++ b/x-pack/plugins/cases/server/client/cases/update.ts
@@ -12,19 +12,23 @@ import { identity } from 'fp-ts/lib/function';
 
 import type {
   SavedObject,
+  SavedObjectsBulkUpdateResponse,
   SavedObjectsFindResponse,
   SavedObjectsFindResult,
+  SavedObjectsUpdateResponse,
 } from '@kbn/core/server';
 
 import { nodeBuilder } from '@kbn/es-query';
 
 import { areTotalAssigneesInvalid } from '../../../common/utils/validators';
 import type {
+  CaseAssignees,
+  CaseAttributes,
   CasePatchRequest,
+  CaseResponse,
   CasesPatchRequest,
   CasesResponse,
   CommentAttributes,
-  CaseAttributes,
   User,
 } from '../../../common/api';
 import {
@@ -42,7 +46,7 @@ import {
   MAX_TITLE_LENGTH,
 } from '../../../common/constants';
 
-import { getCaseToUpdate } from '../utils';
+import { arraysDifference, getCaseToUpdate } from '../utils';
 
 import type { AlertService, CasesService } from '../../services';
 import { createCaseError } from '../../common/error';
@@ -58,6 +62,7 @@ import { Operations } from '../../authorization';
 import { dedupAssignees, getClosedInfoForUpdate, getDurationForUpdate } from './utils';
 import { LICENSING_CASE_ASSIGNMENT_FEATURE } from '../../common/constants';
 import type { LicensingService } from '../../services/licensing';
+import type { CaseSavedObject } from '../../common/types';
 
 /**
  * Throws an error if any of the requests attempt to update the owner of a case.
@@ -252,7 +257,7 @@ async function updateAlerts({
 }
 
 function partitionPatchRequest(
-  casesMap: Map<string, SavedObject<CaseAttributes>>,
+  casesMap: Map<string, CaseSavedObject>,
   patchReqCases: CasePatchRequest[]
 ): {
   nonExistingCases: CasePatchRequest[];
@@ -287,7 +292,7 @@ function partitionPatchRequest(
 
 interface UpdateRequestWithOriginalCase {
   updateReq: CasePatchRequest;
-  originalCase: SavedObject<CaseAttributes>;
+  originalCase: CaseSavedObject;
 }
 
 /**
@@ -301,11 +306,18 @@ export const update = async (
 ): Promise<CasesResponse> => {
   const {
     unsecuredSavedObjectsClient,
-    services: { caseService, userActionService, alertsService, licensingService },
+    services: {
+      caseService,
+      userActionService,
+      alertsService,
+      licensingService,
+      notificationService,
+    },
     user,
     logger,
     authorization,
   } = clientArgs;
+
   const query = pipe(
     excess(CasesPatchRequestRt).decode(cases),
     fold(throwErrors(Boom.badRequest), identity)
@@ -316,10 +328,15 @@ export const update = async (
       caseIds: query.cases.map((q) => q.id),
     });
 
+    /**
+     * Warning: The code below assumes that the
+     * casesMap is immutable. It should be used
+     * only for read.
+     */
     const casesMap = myCases.saved_objects.reduce((acc, so) => {
       acc.set(so.id, so);
       return acc;
-    }, new Map<string, SavedObject<CaseAttributes>>());
+    }, new Map<string, CaseSavedObject>());
 
     const { nonExistingCases, conflictedCases, casesToAuthorize } = partitionPatchRequest(
       casesMap,
@@ -347,7 +364,7 @@ export const update = async (
       );
     }
 
-    const updateCases: UpdateRequestWithOriginalCase[] = query.cases.reduce(
+    const casesToUpdate: UpdateRequestWithOriginalCase[] = query.cases.reduce(
       (acc: UpdateRequestWithOriginalCase[], updateCase) => {
         const originalCase = casesMap.get(updateCase.id);
 
@@ -368,24 +385,24 @@ export const update = async (
       []
     );
 
-    if (updateCases.length <= 0) {
+    if (casesToUpdate.length <= 0) {
       throw Boom.notAcceptable('All update fields are identical to current version.');
     }
 
     const hasPlatinumLicense = await licensingService.isAtLeastPlatinum();
 
-    throwIfUpdateOwner(updateCases);
-    throwIfTitleIsInvalid(updateCases);
-    throwIfUpdateAssigneesWithoutValidLicense(updateCases, hasPlatinumLicense);
-    throwIfTotalAssigneesAreInvalid(updateCases);
+    throwIfUpdateOwner(casesToUpdate);
+    throwIfTitleIsInvalid(casesToUpdate);
+    throwIfUpdateAssigneesWithoutValidLicense(casesToUpdate, hasPlatinumLicense);
+    throwIfTotalAssigneesAreInvalid(casesToUpdate);
 
-    notifyPlatinumUsage(licensingService, updateCases);
+    notifyPlatinumUsage(licensingService, casesToUpdate);
 
-    const updatedCases = await patchCases({ caseService, user, casesToUpdate: updateCases });
+    const updatedCases = await patchCases({ caseService, user, casesToUpdate });
 
     // If a status update occurred and the case is synced then we need to update all alerts' status
     // attached to the case to the new status.
-    const casesWithStatusChangedAndSynced = updateCases.filter(({ updateReq, originalCase }) => {
+    const casesWithStatusChangedAndSynced = casesToUpdate.filter(({ updateReq, originalCase }) => {
       return (
         originalCase != null &&
         updateReq.status != null &&
@@ -396,7 +413,7 @@ export const update = async (
 
     // If syncAlerts setting turned on we need to update all alerts' status
     // attached to the case to the current status.
-    const casesWithSyncSettingChangedToOn = updateCases.filter(({ updateReq, originalCase }) => {
+    const casesWithSyncSettingChangedToOn = casesToUpdate.filter(({ updateReq, originalCase }) => {
       return (
         originalCase != null &&
         updateReq.settings?.syncAlerts != null &&
@@ -413,22 +430,20 @@ export const update = async (
       alertsService,
     });
 
-    const returnUpdatedCase = myCases.saved_objects
-      .filter((myCase) =>
-        updatedCases.saved_objects.some((updatedCase) => updatedCase.id === myCase.id)
-      )
-      .map((myCase) => {
-        const updatedCase = updatedCases.saved_objects.find((c) => c.id === myCase.id);
-        return flattenCaseSavedObject({
-          savedObject: {
-            ...myCase,
-            ...updatedCase,
-            attributes: { ...myCase.attributes, ...updatedCase?.attributes },
-            references: myCase.references,
-            version: updatedCase?.version ?? myCase.version,
-          },
-        });
-      });
+    const returnUpdatedCase = updatedCases.saved_objects.reduce((flattenCases, updatedCase) => {
+      const originalCase = casesMap.get(updatedCase.id);
+
+      if (!originalCase) {
+        return flattenCases;
+      }
+
+      return [
+        ...flattenCases,
+        flattenCaseSavedObject({
+          savedObject: mergeOriginalSOWithUpdatedSO(originalCase, updatedCase),
+        }),
+      ];
+    }, [] as CaseResponse[]);
 
     await userActionService.bulkCreateUpdateCase({
       unsecuredSavedObjectsClient,
@@ -437,6 +452,14 @@ export const update = async (
       user,
     });
 
+    const casesAndAssigneesToNotifyForAssignment = getCasesAndAssigneesToNotifyForAssignment(
+      updatedCases,
+      casesMap,
+      user
+    );
+
+    await notificationService.bulkNotifyAssignees(casesAndAssigneesToNotifyForAssignment);
+
     return CasesResponseRt.encode(returnUpdatedCase);
   } catch (error) {
     const idVersions = cases.cases.map((caseInfo) => ({
@@ -497,3 +520,50 @@ const patchCases = async ({
 
   return updatedCases;
 };
+
+const getCasesAndAssigneesToNotifyForAssignment = (
+  updatedCases: SavedObjectsBulkUpdateResponse<CaseAttributes>,
+  casesMap: Map<string, CaseSavedObject>,
+  user: CasesClientArgs['user']
+) => {
+  return updatedCases.saved_objects.reduce<
+    Array<{ assignees: CaseAssignees; theCase: CaseSavedObject }>
+  >((acc, updatedCase) => {
+    const originalCaseSO = casesMap.get(updatedCase.id);
+
+    if (!originalCaseSO) {
+      return acc;
+    }
+
+    const alreadyAssignedToCase = originalCaseSO.attributes.assignees;
+    const comparedAssignees = arraysDifference(
+      alreadyAssignedToCase,
+      updatedCase.attributes.assignees ?? []
+    );
+
+    if (comparedAssignees && comparedAssignees.addedItems.length > 0) {
+      const theCase = mergeOriginalSOWithUpdatedSO(originalCaseSO, updatedCase);
+
+      const assigneesWithoutCurrentUser = comparedAssignees.addedItems.filter(
+        (assignee) => assignee.uid !== user.profile_uid
+      );
+
+      acc.push({ theCase, assignees: assigneesWithoutCurrentUser });
+    }
+
+    return acc;
+  }, []);
+};
+
+const mergeOriginalSOWithUpdatedSO = (
+  originalSO: CaseSavedObject,
+  updatedSO: SavedObjectsUpdateResponse<CaseAttributes>
+): CaseSavedObject => {
+  return {
+    ...originalSO,
+    ...updatedSO,
+    attributes: { ...originalSO.attributes, ...updatedSO?.attributes },
+    references: updatedSO.references ?? originalSO.references,
+    version: updatedSO?.version ?? updatedSO.version,
+  };
+};
diff --git a/x-pack/plugins/cases/server/client/cases/utils.test.ts b/x-pack/plugins/cases/server/client/cases/utils.test.ts
index d9ea8e83fa91b..9f34a52dae6a4 100644
--- a/x-pack/plugins/cases/server/client/cases/utils.test.ts
+++ b/x-pack/plugins/cases/server/client/cases/utils.test.ts
@@ -5,8 +5,6 @@
  * 2.0.
  */
 
-import { mockCases } from '../../routes/api/__fixtures__';
-
 import {
   comment as commentObj,
   userActions,
@@ -36,6 +34,7 @@ import { flattenCaseSavedObject } from '../../common/utils';
 import { SECURITY_SOLUTION_OWNER } from '../../../common/constants';
 import { casesConnectors } from '../../connectors';
 import { userProfiles, userProfilesMap } from '../user_profiles.mock';
+import { mockCases } from '../../mocks';
 
 const allComments = [
   commentObj,
diff --git a/x-pack/plugins/cases/server/client/factory.ts b/x-pack/plugins/cases/server/client/factory.ts
index 9d1dc2db7b8ac..245a34c81b8fd 100644
--- a/x-pack/plugins/cases/server/client/factory.ts
+++ b/x-pack/plugins/cases/server/client/factory.ts
@@ -19,6 +19,7 @@ import type { PluginStartContract as ActionsPluginStart } from '@kbn/actions-plu
 import type { LensServerPluginSetup } from '@kbn/lens-plugin/server';
 import type { SpacesPluginStart } from '@kbn/spaces-plugin/server';
 import type { LicensingPluginStart } from '@kbn/licensing-plugin/server';
+import type { NotificationsPluginStart } from '@kbn/notifications-plugin/server';
 import { SAVED_OBJECT_TYPES } from '../../common/constants';
 import { Authorization } from '../authorization/authorization';
 import {
@@ -37,6 +38,7 @@ import type { PersistableStateAttachmentTypeRegistry } from '../attachment_frame
 import type { ExternalReferenceAttachmentTypeRegistry } from '../attachment_framework/external_reference_registry';
 import type { CasesServices } from './types';
 import { LicensingService } from '../services/licensing';
+import { EmailNotificationService } from '../services/notifications/email_notification_service';
 
 interface CasesClientFactoryArgs {
   securityPluginSetup: SecurityPluginSetup;
@@ -49,6 +51,7 @@ interface CasesClientFactoryArgs {
   persistableStateAttachmentTypeRegistry: PersistableStateAttachmentTypeRegistry;
   externalReferenceAttachmentTypeRegistry: ExternalReferenceAttachmentTypeRegistry;
   publicBaseUrl?: IBasePath['publicBaseUrl'];
+  notifications: NotificationsPluginStart;
 }
 
 /**
@@ -114,6 +117,7 @@ export class CasesClientFactory {
     const services = this.createServices({
       unsecuredSavedObjectsClient,
       esClient: scopedClusterClient,
+      request,
     });
 
     const userInfo = await this.getUserInfo(request);
@@ -143,9 +147,11 @@ export class CasesClientFactory {
   private createServices({
     unsecuredSavedObjectsClient,
     esClient,
+    request,
   }: {
     unsecuredSavedObjectsClient: SavedObjectsClientContract;
     esClient: ElasticsearchClient;
+    request: KibanaRequest;
   }): CasesServices {
     this.validateInitialization();
 
@@ -165,6 +171,19 @@ export class CasesClientFactory {
       this.options.licensingPluginStart.featureUsage.notifyUsage
     );
 
+    /**
+     * The notifications plugins only exports the EmailService.
+     * We do the same. If in the future we use other means
+     * of notifications we can refactor to use a factory.
+     */
+    const notificationService = new EmailNotificationService({
+      logger: this.logger,
+      notifications: this.options.notifications,
+      security: this.options.securityPluginStart,
+      publicBaseUrl: this.options.publicBaseUrl,
+      spaceId: this.options.spacesPluginStart.spacesService.getSpaceId(request),
+    });
+
     return {
       alertsService: new AlertService(esClient, this.logger),
       caseService,
@@ -176,6 +195,7 @@ export class CasesClientFactory {
       ),
       attachmentService,
       licensingService,
+      notificationService,
     };
   }
 
diff --git a/x-pack/plugins/cases/server/client/metrics/get_case_metrics.test.ts b/x-pack/plugins/cases/server/client/metrics/get_case_metrics.test.ts
index 61c11ad4c7adc..5f31ef8a8ff06 100644
--- a/x-pack/plugins/cases/server/client/metrics/get_case_metrics.test.ts
+++ b/x-pack/plugins/cases/server/client/metrics/get_case_metrics.test.ts
@@ -6,10 +6,9 @@
  */
 
 import { loggingSystemMock, savedObjectsClientMock } from '@kbn/core/server/mocks';
-import type { SavedObject } from '@kbn/core/server';
 
 import { getCaseMetrics } from './get_case_metrics';
-import type { CaseAttributes, CaseResponse } from '../../../common/api';
+import type { CaseResponse } from '../../../common/api';
 import { CaseStatuses } from '../../../common/api';
 import type { CasesClientMock } from '../mocks';
 import { createCasesClientMock } from '../mocks';
@@ -22,6 +21,7 @@ import {
 } from '../../services/mocks';
 import { mockAlertsService } from './test_utils/alerts';
 import { createStatusChangeSavedObject } from './test_utils/lifespan';
+import type { CaseSavedObject } from '../../common/types';
 
 describe('getCaseMetrics', () => {
   const inProgressStatusChangeTimestamp = new Date('2021-11-23T20:00:43Z');
@@ -194,7 +194,7 @@ function createMockClientArgs() {
       attributes: {
         owner: 'security',
       },
-    } as unknown as SavedObject<CaseAttributes>;
+    } as unknown as CaseSavedObject;
   });
 
   const alertsService = mockAlertsService();
diff --git a/x-pack/plugins/cases/server/client/mocks.ts b/x-pack/plugins/cases/server/client/mocks.ts
index cfaf5437f5c7f..79e307e98f8f7 100644
--- a/x-pack/plugins/cases/server/client/mocks.ts
+++ b/x-pack/plugins/cases/server/client/mocks.ts
@@ -6,14 +6,33 @@
  */
 
 import type { PublicContract, PublicMethodsOf } from '@kbn/utility-types';
+import { loggingSystemMock, savedObjectsClientMock } from '@kbn/core/server/mocks';
+import { securityMock } from '@kbn/security-plugin/server/mocks';
 
+import { actionsClientMock } from '@kbn/actions-plugin/server/actions_client.mock';
+import { makeLensEmbeddableFactory } from '@kbn/lens-plugin/server/embeddable/make_lens_embeddable_factory';
 import type { CasesClient } from '.';
+import { createAuthorizationMock } from '../authorization/mock';
+import {
+  connectorMappingsServiceMock,
+  createAlertServiceMock,
+  createAttachmentServiceMock,
+  createCaseServiceMock,
+  createConfigureServiceMock,
+  createLicensingServiceMock,
+  createUserActionServiceMock,
+  createNotificationServiceMock,
+} from '../services/mocks';
 import type { AttachmentsSubClient } from './attachments/client';
 import type { CasesSubClient } from './cases/client';
 import type { ConfigureSubClient } from './configure/client';
 import type { CasesClientFactory } from './factory';
 import type { MetricsSubClient } from './metrics/client';
 import type { UserActionsSubClient } from './user_actions/client';
+import {
+  createExternalReferenceAttachmentTypeRegistryMock,
+  createPersistableStateAttachmentTypeRegistryMock,
+} from '../attachment_framework/mocks';
 
 type CasesSubClientMock = jest.Mocked<CasesSubClient>;
 
@@ -104,3 +123,39 @@ export const createCasesClientFactory = (): CasesClientFactoryMock => {
 
   return factory as unknown as CasesClientFactoryMock;
 };
+
+export const createCasesClientMockArgs = () => {
+  return {
+    services: {
+      alertsService: createAlertServiceMock(),
+      attachmentService: createAttachmentServiceMock(),
+      caseService: createCaseServiceMock(),
+      caseConfigureService: createConfigureServiceMock(),
+      connectorMappingsService: connectorMappingsServiceMock(),
+      userActionService: createUserActionServiceMock(),
+      licensingService: createLicensingServiceMock(),
+      notificationService: createNotificationServiceMock(),
+    },
+    authorization: createAuthorizationMock(),
+    logger: loggingSystemMock.createLogger(),
+    unsecuredSavedObjectsClient: savedObjectsClientMock.create(),
+    actionsClient: actionsClientMock.create(),
+    user: {
+      username: 'damaged_raccoon',
+      email: 'damaged_raccoon@elastic.co',
+      full_name: 'Damaged Raccoon',
+      profile_uid: 'u_J41Oh6L9ki-Vo2tOogS8WRTENzhHurGtRc87NgEAlkc_0',
+    },
+    spaceId: 'default',
+    externalReferenceAttachmentTypeRegistry: createExternalReferenceAttachmentTypeRegistryMock(),
+    persistableStateAttachmentTypeRegistry: createPersistableStateAttachmentTypeRegistryMock(),
+    securityStartPlugin: securityMock.createStart(),
+    lensEmbeddableFactory: jest.fn().mockReturnValue(
+      makeLensEmbeddableFactory(
+        () => ({}),
+        () => ({}),
+        {}
+      )
+    ),
+  };
+};
diff --git a/x-pack/plugins/cases/server/client/types.ts b/x-pack/plugins/cases/server/client/types.ts
index 3309d977ef972..7fded9a5d1a45 100644
--- a/x-pack/plugins/cases/server/client/types.ts
+++ b/x-pack/plugins/cases/server/client/types.ts
@@ -25,6 +25,7 @@ import type {
 import type { PersistableStateAttachmentTypeRegistry } from '../attachment_framework/persistable_state_registry';
 import type { ExternalReferenceAttachmentTypeRegistry } from '../attachment_framework/external_reference_registry';
 import type { LicensingService } from '../services/licensing';
+import type { NotificationService } from '../services/notifications/types';
 
 export interface CasesServices {
   alertsService: AlertService;
@@ -34,6 +35,7 @@ export interface CasesServices {
   userActionService: CaseUserActionService;
   attachmentService: AttachmentService;
   licensingService: LicensingService;
+  notificationService: NotificationService;
 }
 
 /**
diff --git a/x-pack/plugins/cases/server/common/models/case_with_comments.ts b/x-pack/plugins/cases/server/common/models/case_with_comments.ts
index 2e5de37c621e2..5bff0757af718 100644
--- a/x-pack/plugins/cases/server/common/models/case_with_comments.ts
+++ b/x-pack/plugins/cases/server/common/models/case_with_comments.ts
@@ -18,7 +18,6 @@ import type {
   CommentPatchRequest,
   CommentRequest,
   CommentRequestUserType,
-  CaseAttributes,
   CommentRequestAlertType,
 } from '../../../common/api';
 import {
@@ -36,6 +35,7 @@ import {
 import type { CasesClientArgs } from '../../client';
 import type { RefreshSetting } from '../../services/types';
 import { createCaseError } from '../error';
+import type { CaseSavedObject } from '../types';
 import {
   countAlertsForID,
   flattenCommentSavedObjects,
@@ -53,9 +53,9 @@ const ALERT_LIMIT_MSG = `Case has reached the maximum allowed number (${MAX_ALER
  */
 export class CaseCommentModel {
   private readonly params: CaseCommentModelParams;
-  private readonly caseInfo: SavedObject<CaseAttributes>;
+  private readonly caseInfo: CaseSavedObject;
 
-  private constructor(caseInfo: SavedObject<CaseAttributes>, params: CaseCommentModelParams) {
+  private constructor(caseInfo: CaseSavedObject, params: CaseCommentModelParams) {
     this.caseInfo = caseInfo;
     this.params = params;
   }
@@ -71,7 +71,7 @@ export class CaseCommentModel {
     return new CaseCommentModel(savedObject, options);
   }
 
-  public get savedObject(): SavedObject<CaseAttributes> {
+  public get savedObject(): CaseSavedObject {
     return this.caseInfo;
   }
 
@@ -171,7 +171,7 @@ export class CaseCommentModel {
     }
   }
 
-  private newObjectWithInfo(caseInfo: SavedObject<CaseAttributes>): CaseCommentModel {
+  private newObjectWithInfo(caseInfo: CaseSavedObject): CaseCommentModel {
     return new CaseCommentModel(caseInfo, this.params);
   }
 
diff --git a/x-pack/plugins/cases/server/common/types.ts b/x-pack/plugins/cases/server/common/types.ts
index bb5c0d77b8201..8ae038992b28f 100644
--- a/x-pack/plugins/cases/server/common/types.ts
+++ b/x-pack/plugins/cases/server/common/types.ts
@@ -5,8 +5,9 @@
  * 2.0.
  */
 
+import type { SavedObject } from '@kbn/core-saved-objects-common';
 import type { KueryNode } from '@kbn/es-query';
-import type { SavedObjectFindOptions } from '../../common/api';
+import type { CaseAttributes, SavedObjectFindOptions } from '../../common/api';
 
 /**
  * This structure holds the alert ID and index from an alert comment
@@ -19,3 +20,5 @@ export interface AlertInfo {
 export type SavedObjectFindOptionsKueryNode = Omit<SavedObjectFindOptions, 'filter'> & {
   filter?: KueryNode;
 };
+
+export type CaseSavedObject = SavedObject<CaseAttributes>;
diff --git a/x-pack/plugins/cases/server/common/utils.test.ts b/x-pack/plugins/cases/server/common/utils.test.ts
index 0bf4695cbc407..de7a96614b296 100644
--- a/x-pack/plugins/cases/server/common/utils.test.ts
+++ b/x-pack/plugins/cases/server/common/utils.test.ts
@@ -16,7 +16,6 @@ import type {
   CommentRequestUserType,
 } from '../../common/api';
 import { CaseSeverity, CommentType, ConnectorTypes } from '../../common/api';
-import { mockCaseComments, mockCases } from '../routes/api/__fixtures__/mock_saved_objects';
 import {
   flattenCaseSavedObject,
   transformNewComment,
@@ -36,6 +35,7 @@ import {
 } from './utils';
 import { newCase } from '../routes/api/__mocks__/request_responses';
 import { CASE_VIEW_PAGE_TABS } from '../../common/types';
+import { mockCases, mockCaseComments } from '../mocks';
 
 interface CommentReference {
   ids: string[];
diff --git a/x-pack/plugins/cases/server/common/utils.ts b/x-pack/plugins/cases/server/common/utils.ts
index 5d826b0d323f4..508b98775c619 100644
--- a/x-pack/plugins/cases/server/common/utils.ts
+++ b/x-pack/plugins/cases/server/common/utils.ts
@@ -24,7 +24,7 @@ import {
   OWNER_INFO,
 } from '../../common/constants';
 import type { CASE_VIEW_PAGE_TABS } from '../../common/types';
-import type { AlertInfo } from './types';
+import type { AlertInfo, CaseSavedObject } from './types';
 
 import type {
   CaseAttributes,
@@ -118,7 +118,7 @@ export const flattenCaseSavedObject = ({
   totalComment = comments.length,
   totalAlerts = 0,
 }: {
-  savedObject: SavedObject<CaseAttributes>;
+  savedObject: CaseSavedObject;
   comments?: Array<SavedObject<CommentAttributes>>;
   totalComment?: number;
   totalAlerts?: number;
diff --git a/x-pack/plugins/cases/server/routes/api/__fixtures__/mock_saved_objects.ts b/x-pack/plugins/cases/server/mocks.ts
similarity index 94%
rename from x-pack/plugins/cases/server/routes/api/__fixtures__/mock_saved_objects.ts
rename to x-pack/plugins/cases/server/mocks.ts
index d2eda3d6c9d73..c68606bad456e 100644
--- a/x-pack/plugins/cases/server/routes/api/__fixtures__/mock_saved_objects.ts
+++ b/x-pack/plugins/cases/server/mocks.ts
@@ -6,11 +6,12 @@
  */
 
 import type { SavedObject } from '@kbn/core/server';
-import type { CaseAttributes, CommentAttributes } from '../../../../common/api';
-import { CaseSeverity, CaseStatuses, CommentType, ConnectorTypes } from '../../../../common/api';
-import { SECURITY_SOLUTION_OWNER } from '../../../../common/constants';
+import type { CaseSavedObject } from './common/types';
+import type { CasePostRequest, CommentAttributes } from '../common/api';
+import { CaseSeverity, CaseStatuses, CommentType, ConnectorTypes } from '../common/api';
+import { SECURITY_SOLUTION_OWNER } from '../common/constants';
 
-export const mockCases: Array<SavedObject<CaseAttributes>> = [
+export const mockCases: CaseSavedObject[] = [
   {
     type: 'cases',
     id: 'mock-id-1',
@@ -400,3 +401,19 @@ export const mockCaseComments: Array<SavedObject<CommentAttributes>> = [
     version: 'WzYsMV0=',
   },
 ];
+
+export const newCase: CasePostRequest = {
+  title: 'My new case',
+  description: 'A description',
+  tags: ['new', 'case'],
+  connector: {
+    id: 'none',
+    name: 'none',
+    type: ConnectorTypes.none,
+    fields: null,
+  },
+  settings: {
+    syncAlerts: true,
+  },
+  owner: SECURITY_SOLUTION_OWNER,
+};
diff --git a/x-pack/plugins/cases/server/plugin.ts b/x-pack/plugins/cases/server/plugin.ts
index 7f06f34210c4e..ac26cfc7b81a0 100644
--- a/x-pack/plugins/cases/server/plugin.ts
+++ b/x-pack/plugins/cases/server/plugin.ts
@@ -31,8 +31,9 @@ import type {
 } from '@kbn/task-manager-plugin/server';
 import type { UsageCollectionSetup } from '@kbn/usage-collection-plugin/server';
 import type { LicensingPluginSetup, LicensingPluginStart } from '@kbn/licensing-plugin/server';
-import { APP_ID } from '../common/constants';
+import type { NotificationsPluginStart } from '@kbn/notifications-plugin/server';
 
+import { APP_ID } from '../common/constants';
 import {
   createCaseCommentSavedObjectType,
   caseConfigureSavedObjectType,
@@ -72,6 +73,7 @@ export interface PluginsStart {
   taskManager?: TaskManagerStartContract;
   security: SecurityPluginStart;
   spaces: SpacesPluginStart;
+  notifications: NotificationsPluginStart;
 }
 
 export class CasePlugin {
@@ -199,6 +201,7 @@ export class CasePlugin {
       persistableStateAttachmentTypeRegistry: this.persistableStateAttachmentTypeRegistry,
       externalReferenceAttachmentTypeRegistry: this.externalReferenceAttachmentTypeRegistry,
       publicBaseUrl: core.http.basePath.publicBaseUrl,
+      notifications: plugins.notifications,
     });
 
     const client = core.elasticsearch.client;
diff --git a/x-pack/plugins/cases/server/routes/api/__fixtures__/index.ts b/x-pack/plugins/cases/server/routes/api/__fixtures__/index.ts
deleted file mode 100644
index f5c37f96260a4..0000000000000
--- a/x-pack/plugins/cases/server/routes/api/__fixtures__/index.ts
+++ /dev/null
@@ -1,8 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License
- * 2.0; you may not use this file except in compliance with the Elastic License
- * 2.0.
- */
-
-export * from './mock_saved_objects';
diff --git a/x-pack/plugins/cases/server/services/attachments/index.test.ts b/x-pack/plugins/cases/server/services/attachments/index.test.ts
index 49a7badd91769..e1dce7877a58d 100644
--- a/x-pack/plugins/cases/server/services/attachments/index.test.ts
+++ b/x-pack/plugins/cases/server/services/attachments/index.test.ts
@@ -14,7 +14,7 @@ import {
   externalReferenceAttachmentSO,
   externalReferenceAttachmentSOAttributes,
   externalReferenceAttachmentSOAttributesWithoutRefs,
-  getPersistableStateAttachmentTypeRegistry,
+  createPersistableStateAttachmentTypeRegistryMock,
   persistableStateAttachment,
   persistableStateAttachmentAttributes,
   persistableStateAttachmentAttributesWithoutInjectedId,
@@ -23,7 +23,7 @@ import {
 describe('CasesService', () => {
   const unsecuredSavedObjectsClient = savedObjectsClientMock.create();
   const mockLogger = loggerMock.create();
-  const persistableStateAttachmentTypeRegistry = getPersistableStateAttachmentTypeRegistry();
+  const persistableStateAttachmentTypeRegistry = createPersistableStateAttachmentTypeRegistryMock();
   let service: AttachmentService;
 
   beforeEach(() => {
diff --git a/x-pack/plugins/cases/server/services/cases/index.test.ts b/x-pack/plugins/cases/server/services/cases/index.test.ts
index 1043d456f5cae..71f2c96cc2cbc 100644
--- a/x-pack/plugins/cases/server/services/cases/index.test.ts
+++ b/x-pack/plugins/cases/server/services/cases/index.test.ts
@@ -43,6 +43,7 @@ import {
 import type { ESCaseAttributes } from './types';
 import { AttachmentService } from '../attachments';
 import { PersistableStateAttachmentTypeRegistry } from '../../attachment_framework/persistable_state_registry';
+import type { CaseSavedObject } from '../../common/types';
 
 const createUpdateSOResponse = ({
   connector,
@@ -147,7 +148,7 @@ describe('CasesService', () => {
         await service.patchCase({
           caseId: '1',
           updatedAttributes: createCasePostParams(createJiraConnector(), createExternalService()),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         const {
@@ -196,7 +197,7 @@ describe('CasesService', () => {
         await service.patchCase({
           caseId: '1',
           updatedAttributes: createCasePostParams(createJiraConnector(), createExternalService()),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         const { connector } = unsecuredSavedObjectsClient.update.mock
@@ -227,7 +228,7 @@ describe('CasesService', () => {
         await service.patchCase({
           caseId: '1',
           updatedAttributes: createCasePostParams(createJiraConnector(), createExternalService()),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         const { connector } = unsecuredSavedObjectsClient.update.mock
@@ -262,7 +263,7 @@ describe('CasesService', () => {
         await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(createJiraConnector()),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         const updateAttributes = unsecuredSavedObjectsClient.update.mock
@@ -290,7 +291,7 @@ describe('CasesService', () => {
         await service.patchCase({
           caseId: '1',
           updatedAttributes: createCasePostParams(getNoneCaseConnector(), createExternalService()),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         const updateAttributes = unsecuredSavedObjectsClient.update.mock
@@ -320,7 +321,7 @@ describe('CasesService', () => {
           updatedAttributes: createCasePostParams(createJiraConnector(), createExternalService()),
           originalCase: {
             references: [{ id: 'a', name: 'awesome', type: 'hello' }],
-          } as SavedObject<CaseAttributes>,
+          } as CaseSavedObject,
         });
 
         const updateOptions = unsecuredSavedObjectsClient.update.mock
@@ -358,7 +359,7 @@ describe('CasesService', () => {
             references: [
               { id: '1', name: CONNECTOR_ID_REFERENCE_NAME, type: ACTION_SAVED_OBJECT_TYPE },
             ],
-          } as SavedObject<CaseAttributes>,
+          } as CaseSavedObject,
         });
 
         const updateOptions = unsecuredSavedObjectsClient.update.mock
@@ -387,7 +388,7 @@ describe('CasesService', () => {
         await service.patchCase({
           caseId: '1',
           updatedAttributes: createCasePostParams(getNoneCaseConnector(), createExternalService()),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         const updateAttributes = unsecuredSavedObjectsClient.update.mock
@@ -416,7 +417,7 @@ describe('CasesService', () => {
         await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(unsecuredSavedObjectsClient.update.mock.calls[0][2]).toMatchInlineSnapshot(
@@ -435,7 +436,7 @@ describe('CasesService', () => {
         await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(getNoneCaseConnector()),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(unsecuredSavedObjectsClient.update.mock.calls[0][2]).toMatchInlineSnapshot(`
@@ -665,7 +666,7 @@ describe('CasesService', () => {
                 createJiraConnector(),
                 createExternalService()
               ),
-              originalCase: {} as SavedObject<CaseAttributes>,
+              originalCase: {} as CaseSavedObject,
             },
           ],
         });
@@ -713,7 +714,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res.attributes).toMatchInlineSnapshot(`
@@ -737,7 +738,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res.attributes).toMatchInlineSnapshot(`
@@ -756,7 +757,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res.attributes).toMatchInlineSnapshot(`Object {}`);
@@ -771,7 +772,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res).toMatchInlineSnapshot(`
@@ -802,7 +803,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res.attributes.connector).toMatchInlineSnapshot(`
@@ -832,7 +833,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res.attributes.external_service?.connector_id).toBe('none');
@@ -855,7 +856,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res).toMatchInlineSnapshot(`
@@ -891,7 +892,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res.attributes.connector).toMatchInlineSnapshot(`
@@ -917,7 +918,7 @@ describe('CasesService', () => {
         const res = await service.patchCase({
           caseId: '1',
           updatedAttributes: createCaseUpdateParams(),
-          originalCase: {} as SavedObject<CaseAttributes>,
+          originalCase: {} as CaseSavedObject,
         });
 
         expect(res.attributes.external_service).toMatchInlineSnapshot(`
diff --git a/x-pack/plugins/cases/server/services/cases/index.ts b/x-pack/plugins/cases/server/services/cases/index.ts
index 8359675a0ce0b..1ab431f85ba3c 100644
--- a/x-pack/plugins/cases/server/services/cases/index.ts
+++ b/x-pack/plugins/cases/server/services/cases/index.ts
@@ -7,7 +7,6 @@
 
 import type {
   Logger,
-  SavedObject,
   SavedObjectsClientContract,
   SavedObjectsFindResponse,
   SavedObjectsBulkResponse,
@@ -36,7 +35,7 @@ import type {
   CaseStatuses,
 } from '../../../common/api';
 import { caseStatuses } from '../../../common/api';
-import type { SavedObjectFindOptionsKueryNode } from '../../common/types';
+import type { CaseSavedObject, SavedObjectFindOptionsKueryNode } from '../../common/types';
 import { defaultSortField, flattenCaseSavedObject } from '../../common/utils';
 import { DEFAULT_PAGE, DEFAULT_PER_PAGE } from '../../routes/api';
 import { combineFilters } from '../../client/utils';
@@ -93,7 +92,7 @@ interface PostCaseArgs extends IndexRefresh {
 interface PatchCase extends IndexRefresh {
   caseId: string;
   updatedAttributes: Partial<CaseAttributes & PushedArgs>;
-  originalCase: SavedObject<CaseAttributes>;
+  originalCase: CaseSavedObject;
   version?: string;
 }
 type PatchCaseArgs = PatchCase;
@@ -309,7 +308,7 @@ export class CasesService {
     }
   }
 
-  public async getCase({ id: caseId }: GetCaseArgs): Promise<SavedObject<CaseAttributes>> {
+  public async getCase({ id: caseId }: GetCaseArgs): Promise<CaseSavedObject> {
     try {
       this.log.debug(`Attempting to GET case ${caseId}`);
       const caseSavedObject = await this.unsecuredSavedObjectsClient.get<ESCaseAttributes>(
@@ -543,11 +542,7 @@ export class CasesService {
     }
   }
 
-  public async postNewCase({
-    attributes,
-    id,
-    refresh,
-  }: PostCaseArgs): Promise<SavedObject<CaseAttributes>> {
+  public async postNewCase({ attributes, id, refresh }: PostCaseArgs): Promise<CaseSavedObject> {
     try {
       this.log.debug(`Attempting to POST a new case`);
       const transformedAttributes = transformAttributesToESModel(attributes);
diff --git a/x-pack/plugins/cases/server/services/cases/transform.ts b/x-pack/plugins/cases/server/services/cases/transform.ts
index 1b66ff41df421..1204375a0982e 100644
--- a/x-pack/plugins/cases/server/services/cases/transform.ts
+++ b/x-pack/plugins/cases/server/services/cases/transform.ts
@@ -30,6 +30,7 @@ import {
   transformESConnectorToExternalModel,
 } from '../transform';
 import { ConnectorReferenceHandler } from '../connector_reference_handler';
+import type { CaseSavedObject } from '../../common/types';
 
 export function transformUpdateResponsesToExternalModels(
   response: SavedObjectsBulkUpdateResponse<ESCaseAttributes>
@@ -164,7 +165,7 @@ export function transformFindResponseToExternalModel(
 
 export function transformSavedObjectToExternalModel(
   caseSavedObject: SavedObject<ESCaseAttributes>
-): SavedObject<CaseAttributes> {
+): CaseSavedObject {
   const connector = transformESConnectorOrUseDefault({
     // if the saved object had an error the attributes field will not exist
     connector: caseSavedObject.attributes?.connector,
diff --git a/x-pack/plugins/cases/server/services/mocks.ts b/x-pack/plugins/cases/server/services/mocks.ts
index 6ef9af3f36a1b..46b9cd12830c5 100644
--- a/x-pack/plugins/cases/server/services/mocks.ts
+++ b/x-pack/plugins/cases/server/services/mocks.ts
@@ -14,6 +14,8 @@ import type {
   ConnectorMappingsService,
   AttachmentService,
 } from '.';
+import type { LicensingService } from './licensing';
+import type { EmailNotificationService } from './notifications/email_notification_service';
 
 export type CaseServiceMock = jest.Mocked<CasesService>;
 export type CaseConfigureServiceMock = jest.Mocked<CaseConfigureService>;
@@ -21,9 +23,11 @@ export type ConnectorMappingsServiceMock = jest.Mocked<ConnectorMappingsService>
 export type CaseUserActionServiceMock = jest.Mocked<CaseUserActionService>;
 export type AlertServiceMock = jest.Mocked<AlertService>;
 export type AttachmentServiceMock = jest.Mocked<AttachmentService>;
+export type LicensingServiceMock = jest.Mocked<LicensingService>;
+export type NotificationServiceMock = jest.Mocked<EmailNotificationService>;
 
 export const createCaseServiceMock = (): CaseServiceMock => {
-  const service: PublicMethodsOf<CasesService> = {
+  const service = {
     deleteCase: jest.fn(),
     findCases: jest.fn(),
     getAllCaseComments: jest.fn(),
@@ -118,3 +122,27 @@ export const createAttachmentServiceMock = (): AttachmentServiceMock => {
   // the cast here is required because jest.Mocked tries to include private members and would throw an error
   return service as unknown as AttachmentServiceMock;
 };
+
+export const createLicensingServiceMock = (): LicensingServiceMock => {
+  const service: PublicMethodsOf<LicensingService> = {
+    notifyUsage: jest.fn(),
+    getLicenseInformation: jest.fn(),
+    isAtLeast: jest.fn(),
+    isAtLeastPlatinum: jest.fn().mockReturnValue(true),
+    isAtLeastGold: jest.fn(),
+    isAtLeastEnterprise: jest.fn(),
+  };
+
+  // the cast here is required because jest.Mocked tries to include private members and would throw an error
+  return service as unknown as LicensingServiceMock;
+};
+
+export const createNotificationServiceMock = (): NotificationServiceMock => {
+  const service: PublicMethodsOf<EmailNotificationService> = {
+    notifyAssignees: jest.fn(),
+    bulkNotifyAssignees: jest.fn(),
+  };
+
+  // the cast here is required because jest.Mocked tries to include private members and would throw an error
+  return service as unknown as NotificationServiceMock;
+};
diff --git a/x-pack/plugins/cases/server/services/notifications/email_notification_service.test.ts b/x-pack/plugins/cases/server/services/notifications/email_notification_service.test.ts
new file mode 100644
index 0000000000000..84ac5283067f6
--- /dev/null
+++ b/x-pack/plugins/cases/server/services/notifications/email_notification_service.test.ts
@@ -0,0 +1,270 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { notificationsMock } from '@kbn/notifications-plugin/server/mocks';
+import { createCasesClientMockArgs } from '../../client/mocks';
+import { userProfiles } from '../../client/user_profiles.mock';
+import { mockCases } from '../../mocks';
+import { EmailNotificationService } from './email_notification_service';
+
+describe('EmailNotificationService', () => {
+  const clientArgs = createCasesClientMockArgs();
+  const caseSO = mockCases[0];
+  const assignees = userProfiles.map((userProfile) => ({ uid: userProfile.uid }));
+
+  const notifications = notificationsMock.createStart();
+
+  let emailNotificationService: EmailNotificationService;
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    notifications.isEmailServiceAvailable.mockReturnValue(true);
+    clientArgs.securityStartPlugin.userProfiles.bulkGet.mockResolvedValue(userProfiles);
+
+    emailNotificationService = new EmailNotificationService({
+      logger: clientArgs.logger,
+      security: clientArgs.securityStartPlugin,
+      publicBaseUrl: 'https://example.com',
+      notifications,
+      spaceId: 'default',
+    });
+  });
+
+  it('notifies assignees', async () => {
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: caseSO,
+    });
+
+    expect(notifications.getEmailService().sendPlainTextEmail).toHaveBeenCalledWith({
+      context: {
+        relatedObjects: [
+          {
+            id: 'mock-id-1',
+            namespace: undefined,
+            type: 'cases',
+          },
+        ],
+      },
+      message:
+        'You are assigned to an Elastic Kibana Case.\r\n\r\nTitle: Super Bad Security Issue\r\n\r\nStatus: open\r\n\r\nSeverity: low\r\n\r\nTags: defacement\r\n\r\n\r\n\r\n[View the case details](https://example.com/app/security/cases/mock-id-1)',
+      subject: '[Elastic][Cases] Super Bad Security Issue',
+      to: ['damaged_raccoon@elastic.co', 'physical_dinosaur@elastic.co', 'wet_dingo@elastic.co'],
+    });
+  });
+
+  it('filters out duplicates assignees', async () => {
+    await emailNotificationService.notifyAssignees({
+      assignees: [...assignees, { uid: assignees[0].uid }],
+      theCase: caseSO,
+    });
+
+    expect(notifications.getEmailService().sendPlainTextEmail).toHaveBeenCalledWith({
+      context: {
+        relatedObjects: [
+          {
+            id: 'mock-id-1',
+            namespace: undefined,
+            type: 'cases',
+          },
+        ],
+      },
+      message:
+        'You are assigned to an Elastic Kibana Case.\r\n\r\nTitle: Super Bad Security Issue\r\n\r\nStatus: open\r\n\r\nSeverity: low\r\n\r\nTags: defacement\r\n\r\n\r\n\r\n[View the case details](https://example.com/app/security/cases/mock-id-1)',
+      subject: '[Elastic][Cases] Super Bad Security Issue',
+      to: ['damaged_raccoon@elastic.co', 'physical_dinosaur@elastic.co', 'wet_dingo@elastic.co'],
+    });
+  });
+
+  it('filters out assignees without email', async () => {
+    clientArgs.securityStartPlugin.userProfiles.bulkGet.mockResolvedValue([
+      { ...userProfiles[0], user: { ...userProfiles[0].user, email: undefined } },
+      { ...userProfiles[1] },
+    ]);
+
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: caseSO,
+    });
+
+    expect(notifications.getEmailService().sendPlainTextEmail).toHaveBeenCalledWith({
+      context: {
+        relatedObjects: [
+          {
+            id: 'mock-id-1',
+            namespace: undefined,
+            type: 'cases',
+          },
+        ],
+      },
+      message:
+        'You are assigned to an Elastic Kibana Case.\r\n\r\nTitle: Super Bad Security Issue\r\n\r\nStatus: open\r\n\r\nSeverity: low\r\n\r\nTags: defacement\r\n\r\n\r\n\r\n[View the case details](https://example.com/app/security/cases/mock-id-1)',
+      subject: '[Elastic][Cases] Super Bad Security Issue',
+      to: ['physical_dinosaur@elastic.co'],
+    });
+  });
+
+  it('passes the namespace correctly', async () => {
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: { ...caseSO, namespaces: ['space1'] },
+    });
+
+    expect(notifications.getEmailService().sendPlainTextEmail).toHaveBeenCalledWith({
+      context: {
+        relatedObjects: [
+          {
+            id: 'mock-id-1',
+            namespace: 'space1',
+            type: 'cases',
+          },
+        ],
+      },
+      message:
+        'You are assigned to an Elastic Kibana Case.\r\n\r\nTitle: Super Bad Security Issue\r\n\r\nStatus: open\r\n\r\nSeverity: low\r\n\r\nTags: defacement\r\n\r\n\r\n\r\n[View the case details](https://example.com/app/security/cases/mock-id-1)',
+      subject: '[Elastic][Cases] Super Bad Security Issue',
+      to: ['damaged_raccoon@elastic.co', 'physical_dinosaur@elastic.co', 'wet_dingo@elastic.co'],
+    });
+  });
+
+  it('adds a backlink URL correctly with spaceId', async () => {
+    emailNotificationService = new EmailNotificationService({
+      logger: clientArgs.logger,
+      security: clientArgs.securityStartPlugin,
+      publicBaseUrl: 'https://example.com',
+      notifications,
+      spaceId: 'test-space',
+    });
+
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: caseSO,
+    });
+
+    expect(notifications.getEmailService().sendPlainTextEmail).toHaveBeenCalledWith({
+      context: {
+        relatedObjects: [
+          {
+            id: 'mock-id-1',
+            namespace: undefined,
+            type: 'cases',
+          },
+        ],
+      },
+      message:
+        'You are assigned to an Elastic Kibana Case.\r\n\r\nTitle: Super Bad Security Issue\r\n\r\nStatus: open\r\n\r\nSeverity: low\r\n\r\nTags: defacement\r\n\r\n\r\n\r\n[View the case details](https://example.com/s/test-space/app/security/cases/mock-id-1)',
+      subject: '[Elastic][Cases] Super Bad Security Issue',
+      to: ['damaged_raccoon@elastic.co', 'physical_dinosaur@elastic.co', 'wet_dingo@elastic.co'],
+    });
+  });
+
+  it('does not include the backlink of the publicBaseUrl is not defined', async () => {
+    emailNotificationService = new EmailNotificationService({
+      logger: clientArgs.logger,
+      security: clientArgs.securityStartPlugin,
+      notifications,
+      spaceId: 'default',
+    });
+
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: caseSO,
+    });
+
+    expect(notifications.getEmailService().sendPlainTextEmail).toHaveBeenCalledWith({
+      context: {
+        relatedObjects: [
+          {
+            id: 'mock-id-1',
+            namespace: undefined,
+            type: 'cases',
+          },
+        ],
+      },
+      message:
+        'You are assigned to an Elastic Kibana Case.\r\n\r\nTitle: Super Bad Security Issue\r\n\r\nStatus: open\r\n\r\nSeverity: low\r\n\r\nTags: defacement\r\n\r\n',
+      subject: '[Elastic][Cases] Super Bad Security Issue',
+      to: ['damaged_raccoon@elastic.co', 'physical_dinosaur@elastic.co', 'wet_dingo@elastic.co'],
+    });
+  });
+
+  it('shows multiple tags correctly', async () => {
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: { ...caseSO, attributes: { ...caseSO.attributes, tags: ['one', 'two'] } },
+    });
+
+    expect(notifications.getEmailService().sendPlainTextEmail).toHaveBeenCalledWith({
+      context: {
+        relatedObjects: [
+          {
+            id: 'mock-id-1',
+            namespace: undefined,
+            type: 'cases',
+          },
+        ],
+      },
+      message:
+        'You are assigned to an Elastic Kibana Case.\r\n\r\nTitle: Super Bad Security Issue\r\n\r\nStatus: open\r\n\r\nSeverity: low\r\n\r\nTags: one, two\r\n\r\n\r\n\r\n[View the case details](https://example.com/app/security/cases/mock-id-1)',
+      subject: '[Elastic][Cases] Super Bad Security Issue',
+      to: ['damaged_raccoon@elastic.co', 'physical_dinosaur@elastic.co', 'wet_dingo@elastic.co'],
+    });
+  });
+
+  it('does not show the tags section with empty tags', async () => {
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: { ...caseSO, attributes: { ...caseSO.attributes, tags: [] } },
+    });
+
+    expect(notifications.getEmailService().sendPlainTextEmail).toHaveBeenCalledWith({
+      context: {
+        relatedObjects: [
+          {
+            id: 'mock-id-1',
+            namespace: undefined,
+            type: 'cases',
+          },
+        ],
+      },
+      message:
+        'You are assigned to an Elastic Kibana Case.\r\n\r\nTitle: Super Bad Security Issue\r\n\r\nStatus: open\r\n\r\nSeverity: low\r\n\r\n\r\n\r\n[View the case details](https://example.com/app/security/cases/mock-id-1)',
+      subject: '[Elastic][Cases] Super Bad Security Issue',
+      to: ['damaged_raccoon@elastic.co', 'physical_dinosaur@elastic.co', 'wet_dingo@elastic.co'],
+    });
+  });
+
+  it('logs a warning and not notify assignees when the email service is not available', async () => {
+    notifications.isEmailServiceAvailable.mockReturnValue(false);
+
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: caseSO,
+    });
+
+    expect(clientArgs.logger.warn).toHaveBeenCalledWith(
+      'Could not notifying assignees. Email service is not available.'
+    );
+    expect(notifications.getEmailService().sendPlainTextEmail).not.toHaveBeenCalled();
+  });
+
+  it('logs a warning and not notify assignees on error', async () => {
+    clientArgs.securityStartPlugin.userProfiles.bulkGet.mockRejectedValue(
+      new Error('Cannot get user profiles')
+    );
+
+    await emailNotificationService.notifyAssignees({
+      assignees,
+      theCase: caseSO,
+    });
+
+    expect(clientArgs.logger.warn).toHaveBeenCalledWith(
+      'Error notifying assignees: Cannot get user profiles'
+    );
+    expect(notifications.getEmailService().sendPlainTextEmail).not.toHaveBeenCalled();
+  });
+});
diff --git a/x-pack/plugins/cases/server/services/notifications/email_notification_service.ts b/x-pack/plugins/cases/server/services/notifications/email_notification_service.ts
new file mode 100644
index 0000000000000..fc7458c7453ba
--- /dev/null
+++ b/x-pack/plugins/cases/server/services/notifications/email_notification_service.ts
@@ -0,0 +1,143 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import pMap from 'p-map';
+import type { IBasePath, Logger } from '@kbn/core/server';
+import type { NotificationsPluginStart } from '@kbn/notifications-plugin/server';
+import type { SecurityPluginStart } from '@kbn/security-plugin/server';
+import type { UserProfileUserInfo } from '@kbn/user-profile-components';
+import { CASE_SAVED_OBJECT, MAX_CONCURRENT_SEARCHES } from '../../../common/constants';
+import type { CaseSavedObject } from '../../common/types';
+import { getCaseViewPath } from '../../common/utils';
+import type { NotificationService, NotifyArgs } from './types';
+
+type WithRequiredProperty<T, K extends keyof T> = T & Required<Pick<T, K>>;
+
+type UserProfileUserInfoWithEmail = WithRequiredProperty<UserProfileUserInfo, 'email'>;
+
+export class EmailNotificationService implements NotificationService {
+  private readonly logger: Logger;
+  private readonly notifications: NotificationsPluginStart;
+  private readonly security: SecurityPluginStart;
+  private readonly spaceId: string;
+  private readonly publicBaseUrl?: IBasePath['publicBaseUrl'];
+
+  constructor({
+    logger,
+    notifications,
+    security,
+    publicBaseUrl,
+    spaceId,
+  }: {
+    logger: Logger;
+    notifications: NotificationsPluginStart;
+    security: SecurityPluginStart;
+    spaceId: string;
+    publicBaseUrl?: IBasePath['publicBaseUrl'];
+  }) {
+    this.logger = logger;
+    this.notifications = notifications;
+    this.security = security;
+    this.spaceId = spaceId;
+    this.publicBaseUrl = publicBaseUrl;
+  }
+
+  private static getTitle(theCase: CaseSavedObject) {
+    return `[Elastic][Cases] ${theCase.attributes.title}`;
+  }
+
+  private static getMessage(
+    theCase: CaseSavedObject,
+    spaceId: string,
+    publicBaseUrl?: IBasePath['publicBaseUrl']
+  ) {
+    const lineBreak = '\r\n\r\n';
+    let message = `You are assigned to an Elastic Kibana Case.${lineBreak}`;
+    message = `${message}Title: ${theCase.attributes.title}${lineBreak}`;
+    message = `${message}Status: ${theCase.attributes.status}${lineBreak}`;
+    message = `${message}Severity: ${theCase.attributes.severity}${lineBreak}`;
+
+    if (theCase.attributes.tags.length > 0) {
+      message = `${message}Tags: ${theCase.attributes.tags.join(', ')}${lineBreak}`;
+    }
+
+    if (publicBaseUrl) {
+      const caseUrl = getCaseViewPath({
+        publicBaseUrl,
+        caseId: theCase.id,
+        owner: theCase.attributes.owner,
+        spaceId,
+      });
+
+      message = `${message}${lineBreak}[View the case details](${caseUrl})`;
+    }
+
+    return message;
+  }
+
+  public async notifyAssignees({ assignees, theCase }: NotifyArgs) {
+    try {
+      if (!this.notifications.isEmailServiceAvailable()) {
+        this.logger.warn('Could not notifying assignees. Email service is not available.');
+        return;
+      }
+
+      const uids = new Set(assignees.map((assignee) => assignee.uid));
+      const userProfiles = await this.security.userProfiles.bulkGet({ uids });
+      const users = userProfiles.map((profile) => profile.user);
+
+      const to = users
+        .filter((user): user is UserProfileUserInfoWithEmail => user.email != null)
+        .map((user) => user.email);
+
+      const subject = EmailNotificationService.getTitle(theCase);
+      const message = EmailNotificationService.getMessage(
+        theCase,
+        this.spaceId,
+        this.publicBaseUrl
+      );
+
+      await this.notifications.getEmailService().sendPlainTextEmail({
+        to,
+        subject,
+        message,
+        context: {
+          relatedObjects: [
+            {
+              id: theCase.id,
+              type: CASE_SAVED_OBJECT,
+              /**
+               * Cases are not shareable at the moment from the UI
+               * The namespaces should be either undefined or contain
+               * only one item, the space the case got created. If we decide
+               * in the future to share cases in multiple spaces we need
+               * to change the logic.
+               */
+              namespace: theCase.namespaces?.[0],
+            },
+          ],
+        },
+      });
+    } catch (error) {
+      this.logger.warn(`Error notifying assignees: ${error.message}`);
+    }
+  }
+
+  public async bulkNotifyAssignees(casesAndAssigneesToNotifyForAssignment: NotifyArgs[]) {
+    if (casesAndAssigneesToNotifyForAssignment.length === 0) {
+      return;
+    }
+
+    await pMap(
+      casesAndAssigneesToNotifyForAssignment,
+      (args: NotifyArgs) => this.notifyAssignees(args),
+      {
+        concurrency: MAX_CONCURRENT_SEARCHES,
+      }
+    );
+  }
+}
diff --git a/x-pack/plugins/cases/server/services/notifications/types.ts b/x-pack/plugins/cases/server/services/notifications/types.ts
new file mode 100644
index 0000000000000..0efc326cf7835
--- /dev/null
+++ b/x-pack/plugins/cases/server/services/notifications/types.ts
@@ -0,0 +1,19 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { CaseAssignees } from '../../../common/api';
+import type { CaseSavedObject } from '../../common/types';
+
+export interface NotifyArgs {
+  assignees: CaseAssignees;
+  theCase: CaseSavedObject;
+}
+
+export interface NotificationService {
+  notifyAssignees: (args: NotifyArgs) => Promise<void>;
+  bulkNotifyAssignees: (args: NotifyArgs[]) => Promise<void>;
+}
diff --git a/x-pack/plugins/cases/server/services/so_references.test.ts b/x-pack/plugins/cases/server/services/so_references.test.ts
index 1f7a5d3f135b5..4b349a90c4c92 100644
--- a/x-pack/plugins/cases/server/services/so_references.test.ts
+++ b/x-pack/plugins/cases/server/services/so_references.test.ts
@@ -9,7 +9,7 @@ import {
   externalReferenceAttachmentESAttributes,
   externalReferenceAttachmentSOAttributes,
   externalReferenceAttachmentSOAttributesWithoutRefs,
-  getPersistableStateAttachmentTypeRegistry,
+  createPersistableStateAttachmentTypeRegistryMock,
   persistableStateAttachmentAttributes,
   persistableStateAttachmentAttributesWithoutInjectedId,
 } from '../attachment_framework/mocks';
@@ -21,7 +21,7 @@ import {
 } from './so_references';
 
 describe('so_references', () => {
-  const persistableStateAttachmentTypeRegistry = getPersistableStateAttachmentTypeRegistry();
+  const persistableStateAttachmentTypeRegistry = createPersistableStateAttachmentTypeRegistryMock();
   const references = [
     {
       id: 'testRef',
diff --git a/x-pack/plugins/cases/server/services/user_actions/builder_factory.test.ts b/x-pack/plugins/cases/server/services/user_actions/builder_factory.test.ts
index 1815b5abe7491..8a3fbcc03d2a1 100644
--- a/x-pack/plugins/cases/server/services/user_actions/builder_factory.test.ts
+++ b/x-pack/plugins/cases/server/services/user_actions/builder_factory.test.ts
@@ -17,14 +17,14 @@ import {
 import {
   externalReferenceAttachmentES,
   externalReferenceAttachmentSO,
-  getPersistableStateAttachmentTypeRegistry,
+  createPersistableStateAttachmentTypeRegistryMock,
   persistableStateAttachment,
 } from '../../attachment_framework/mocks';
 import { BuilderFactory } from './builder_factory';
 import { casePayload, externalService } from './mocks';
 
 describe('UserActionBuilder', () => {
-  const persistableStateAttachmentTypeRegistry = getPersistableStateAttachmentTypeRegistry();
+  const persistableStateAttachmentTypeRegistry = createPersistableStateAttachmentTypeRegistryMock();
   const builderFactory = new BuilderFactory({ persistableStateAttachmentTypeRegistry });
   const commonArgs = {
     caseId: '123',
diff --git a/x-pack/plugins/cases/server/services/user_actions/index.test.ts b/x-pack/plugins/cases/server/services/user_actions/index.test.ts
index 732b4e0c924f3..a14e16a87db92 100644
--- a/x-pack/plugins/cases/server/services/user_actions/index.test.ts
+++ b/x-pack/plugins/cases/server/services/user_actions/index.test.ts
@@ -59,7 +59,7 @@ import { CaseUserActionService, transformFindResponseToExternalModel } from '.';
 import type { PersistableStateAttachmentTypeRegistry } from '../../attachment_framework/persistable_state_registry';
 import {
   externalReferenceAttachmentSO,
-  getPersistableStateAttachmentTypeRegistry,
+  createPersistableStateAttachmentTypeRegistryMock,
   persistableStateAttachment,
 } from '../../attachment_framework/mocks';
 
@@ -305,7 +305,7 @@ const testConnectorId = (
 };
 
 describe('CaseUserActionService', () => {
-  const persistableStateAttachmentTypeRegistry = getPersistableStateAttachmentTypeRegistry();
+  const persistableStateAttachmentTypeRegistry = createPersistableStateAttachmentTypeRegistryMock();
 
   beforeAll(() => {
     jest.useFakeTimers('modern');
diff --git a/x-pack/plugins/cases/server/services/user_actions/index.ts b/x-pack/plugins/cases/server/services/user_actions/index.ts
index c27c313faaae7..37642218364b9 100644
--- a/x-pack/plugins/cases/server/services/user_actions/index.ts
+++ b/x-pack/plugins/cases/server/services/user_actions/index.ts
@@ -68,6 +68,7 @@ import type { PersistableStateAttachmentTypeRegistry } from '../../attachment_fr
 import { injectPersistableReferencesToSO } from '../../attachment_framework/so_references';
 import type { IndexRefresh } from '../types';
 import { isAssigneesArray, isStringArray } from './type_guards';
+import type { CaseSavedObject } from '../../common/types';
 
 interface GetCaseUserActionArgs extends ClientArgs {
   caseId: string;
@@ -106,7 +107,7 @@ interface TypedUserActionDiffedItems<T> extends GetUserActionItemByDifference {
 }
 
 interface BulkCreateBulkUpdateCaseUserActions extends ClientArgs, IndexRefresh {
-  originalCases: Array<SavedObject<CaseAttributes>>;
+  originalCases: CaseSavedObject[];
   updatedCases: Array<SavedObjectsUpdateResponse<CaseAttributes>>;
   user: User;
 }
diff --git a/x-pack/plugins/cases/tsconfig.json b/x-pack/plugins/cases/tsconfig.json
index 0237880148358..3bc66fd8bca3b 100644
--- a/x-pack/plugins/cases/tsconfig.json
+++ b/x-pack/plugins/cases/tsconfig.json
@@ -24,6 +24,7 @@
     { "path": "../rule_registry/tsconfig.json" },
     { "path": "../triggers_actions_ui/tsconfig.json"},
     { "path": "../stack_connectors/tsconfig.json"},
+    { "path": "../notifications/tsconfig.json" },
     { "path": "../../../src/plugins/es_ui_shared/tsconfig.json" },
     { "path": "../../../src/plugins/kibana_react/tsconfig.json" },
     { "path": "../../../src/plugins/kibana_utils/tsconfig.json" },