Add Web Push Support (#11346)

* WIP: PoC with lots of terrible code with web push

* fix service worker building

* Work on WebPush Tailored to Browser

* Clean Up Web And MV2

* Fix Merge Conflicts

* Prettier

* Use Unsupported for MV2

* Add Doc Comments

* Remove Permission Button

* Fix Type Test

* Write Time In More Readable Format

* Add SignalR Logger

* `sheduleReconnect` -> `scheduleReconnect`

Co-authored-by: Matt Gibson <mgibson@bitwarden.com>

* Capture Support Context In Connector

* Remove Unneeded CSP Change

* Fix Build

* Simplify `getOrCreateSubscription`

* Add More Docs to Matrix

* Update libs/common/src/platform/notifications/internal/worker-webpush-connection.service.ts

Co-authored-by: Matt Gibson <mgibson@bitwarden.com>

* Move API Service Into Notifications Folder

* Allow Connection When Account Is Locked

* Add Comments to NotificationsService

* Only Change Support Status If Public Key Changes

* Move Service Choice Out To Method

* Use Named Constant For Disabled Notification Url

* Add Test & Cleanup

* Flatten

* Move Tests into `beforeEach` & `afterEach`

* Add Tests

* Test `distinctUntilChanged`'s Operators More

* Make Helper And Cleanup Chain

* Add Back Cast

* Add extra safety to incoming config check

* Put data through response object

* Apply TS Strict Rules

* Finish PushTechnology comment

* Use `instanceof` check

* Do Safer Worker Based Registration for MV3

* Remove TODO

* Switch to SignalR on any WebPush Error

* Fix Manifest Permissions

* Add Back `webNavigation`

* Sorry, Remove `webNavigation`

* Fixed merge conflicts.

---------

Co-authored-by: Matt Gibson <mgibson@bitwarden.com>
Co-authored-by: Todd Martin <tmartin@bitwarden.com>
Co-authored-by: Todd Martin <106564991+trmartin4@users.noreply.github.com>

libs/common/src/platform/abstractions/config/server-config.ts

@@ -3,6 +3,7 @@
 import { Jsonify } from "type-fest";
 
 import { AllowedFeatureFlagTypes } from "../../../enums/feature-flag.enum";
+import { PushTechnology } from "../../../enums/push-technology.enum";
 import {
   ServerConfigData,
   ThirdPartyServerConfigData,
@@ -10,6 +11,11 @@ import {
 } from "../../models/data/server-config.data";
 import { ServerSettings } from "../../models/domain/server-settings";
 
+type PushConfig =
+  | { pushTechnology: PushTechnology.SignalR }
+  | { pushTechnology: PushTechnology.WebPush; vapidPublicKey: string }
+  | undefined;
+
 const dayInMilliseconds = 24 * 3600 * 1000;
 
 export class ServerConfig {
@@ -19,6 +25,7 @@ export class ServerConfig {
   environment?: EnvironmentServerConfigData;
   utcDate: Date;
   featureStates: { [key: string]: AllowedFeatureFlagTypes } = {};
+  push: PushConfig;
   settings: ServerSettings;
 
   constructor(serverConfigData: ServerConfigData) {
@@ -28,6 +35,15 @@ export class ServerConfig {
     this.utcDate = new Date(serverConfigData.utcDate);
     this.environment = serverConfigData.environment;
     this.featureStates = serverConfigData.featureStates;
+    this.push =
+      serverConfigData.push == null
+        ? {
+            pushTechnology: PushTechnology.SignalR,
+          }
+        : {
+            pushTechnology: serverConfigData.push.pushTechnology,
+            vapidPublicKey: serverConfigData.push.vapidPublicKey,
+          };
     this.settings = serverConfigData.settings;
 
     if (this.server?.name == null && this.server?.url == null) {

libs/common/src/platform/misc/support-status.ts

@@ -0,0 +1,48 @@
+import { ObservableInput, OperatorFunction, switchMap } from "rxjs";
+
+/**
+ * Indicates that the given set of actions is not supported and there is
+ * not anything the user can do to make it supported. The reason property
+ * should contain a documented and machine readable string so more in
+ * depth details can be shown to the user.
+ */
+export type NotSupported = { type: "not-supported"; reason: string };
+
+/**
+ * Indicates that the given set of actions does not currently work but
+ * could be supported if configuration, either inside Bitwarden or outside,
+ * is done. The reason property should contain a documented and
+ * machine readable string so further instruction can be supplied to the caller.
+ */
+export type NeedsConfiguration = { type: "needs-configuration"; reason: string };
+
+/**
+ * Indicates that the actions in the service property are supported.
+ */
+export type Supported<T> = { type: "supported"; service: T };
+
+/**
+ * A type encapsulating the status of support for a service.
+ */
+export type SupportStatus<T> = Supported<T> | NeedsConfiguration | NotSupported;
+
+/**
+ * Projects each source value to one of the given projects defined in `selectors`.
+ *
+ * @param selectors.supported The function to run when the given item reports that it is supported
+ * @param selectors.notSupported The function to run when the given item reports that it is either not-supported
+ *   or needs-configuration.
+ * @returns A function that returns an Observable that emits the result of one of the given projection functions.
+ */
+export function supportSwitch<TService, TSupported, TNotSupported>(selectors: {
+  supported: (service: TService, index: number) => ObservableInput<TSupported>;
+  notSupported: (reason: string, index: number) => ObservableInput<TNotSupported>;
+}): OperatorFunction<SupportStatus<TService>, TSupported | TNotSupported> {
+  return switchMap((supportStatus, index) => {
+    if (supportStatus.type === "supported") {
+      return selectors.supported(supportStatus.service, index);
+    }
+
+    return selectors.notSupported(supportStatus.reason, index);
+  });
+}

libs/common/src/platform/models/data/server-config.data.spec.ts

@@ -1,3 +1,4 @@
+import { PushTechnology } from "../../../enums/push-technology.enum";
 import { Region } from "../../abstractions/environment.service";
 
 import {
@@ -29,6 +30,9 @@ describe("ServerConfigData", () => {
         },
         utcDate: "2020-01-01T00:00:00.000Z",
         featureStates: { feature: "state" },
+        push: {
+          pushTechnology: PushTechnology.SignalR,
+        },
       };
       const serverConfigData = ServerConfigData.fromJSON(json);
 

libs/common/src/platform/models/data/server-config.data.ts

@@ -9,6 +9,7 @@ import {
   ServerConfigResponse,
   ThirdPartyServerConfigResponse,
   EnvironmentServerConfigResponse,
+  PushSettingsConfigResponse,
 } from "../response/server-config.response";
 
 export class ServerConfigData {
@@ -18,6 +19,7 @@ export class ServerConfigData {
   environment?: EnvironmentServerConfigData;
   utcDate: string;
   featureStates: { [key: string]: AllowedFeatureFlagTypes } = {};
+  push: PushSettingsConfigData;
   settings: ServerSettings;
 
   constructor(serverConfigResponse: Partial<ServerConfigResponse>) {
@@ -32,6 +34,9 @@ export class ServerConfigData {
       : null;
     this.featureStates = serverConfigResponse?.featureStates;
     this.settings = new ServerSettings(serverConfigResponse.settings);
+    this.push = serverConfigResponse?.push
+      ? new PushSettingsConfigData(serverConfigResponse.push)
+      : null;
   }
 
   static fromJSON(obj: Jsonify<ServerConfigData>): ServerConfigData {
@@ -42,6 +47,20 @@ export class ServerConfigData {
   }
 }
 
+export class PushSettingsConfigData {
+  pushTechnology: number;
+  vapidPublicKey?: string;
+
+  constructor(response: Partial<PushSettingsConfigResponse>) {
+    this.pushTechnology = response.pushTechnology;
+    this.vapidPublicKey = response.vapidPublicKey;
+  }
+
+  static fromJSON(obj: Jsonify<PushSettingsConfigData>): PushSettingsConfigData {
+    return Object.assign(new PushSettingsConfigData({}), obj);
+  }
+}
+
 export class ThirdPartyServerConfigData {
   name: string;
   url: string;

libs/common/src/platform/models/response/server-config.response.ts

@@ -11,6 +11,7 @@ export class ServerConfigResponse extends BaseResponse {
   server: ThirdPartyServerConfigResponse;
   environment: EnvironmentServerConfigResponse;
   featureStates: { [key: string]: AllowedFeatureFlagTypes } = {};
+  push: PushSettingsConfigResponse;
   settings: ServerSettings;
 
   constructor(response: any) {
@@ -25,10 +26,27 @@ export class ServerConfigResponse extends BaseResponse {
     this.server = new ThirdPartyServerConfigResponse(this.getResponseProperty("Server"));
     this.environment = new EnvironmentServerConfigResponse(this.getResponseProperty("Environment"));
     this.featureStates = this.getResponseProperty("FeatureStates");
+    this.push = new PushSettingsConfigResponse(this.getResponseProperty("Push"));
     this.settings = new ServerSettings(this.getResponseProperty("Settings"));
   }
 }
 
+export class PushSettingsConfigResponse extends BaseResponse {
+  pushTechnology: number;
+  vapidPublicKey: string;
+
+  constructor(data: any = null) {
+    super(data);
+
+    if (data == null) {
+      return;
+    }
+
+    this.pushTechnology = this.getResponseProperty("PushTechnology");
+    this.vapidPublicKey = this.getResponseProperty("VapidPublicKey");
+  }
+}
+
 export class EnvironmentServerConfigResponse extends BaseResponse {
   cloudRegion: Region;
   vault: string;

libs/common/src/platform/notifications/index.ts

@@ -0,0 +1 @@
+export { NotificationsService } from "./notifications.service";

libs/common/src/platform/notifications/internal/default-notifications.service.spec.ts

@@ -0,0 +1,316 @@
+import { mock, MockProxy } from "jest-mock-extended";
+import { BehaviorSubject, bufferCount, firstValueFrom, ObservedValueOf, Subject } from "rxjs";
+
+import { LogoutReason } from "@bitwarden/auth/common";
+
+import { awaitAsync } from "../../../../spec";
+import { Matrix } from "../../../../spec/matrix";
+import { AccountService } from "../../../auth/abstractions/account.service";
+import { AuthService } from "../../../auth/abstractions/auth.service";
+import { AuthenticationStatus } from "../../../auth/enums/authentication-status";
+import { NotificationType } from "../../../enums";
+import { NotificationResponse } from "../../../models/response/notification.response";
+import { UserId } from "../../../types/guid";
+import { AppIdService } from "../../abstractions/app-id.service";
+import { Environment, EnvironmentService } from "../../abstractions/environment.service";
+import { LogService } from "../../abstractions/log.service";
+import { MessageSender } from "../../messaging";
+import { SupportStatus } from "../../misc/support-status";
+import { SyncService } from "../../sync";
+
+import {
+  DefaultNotificationsService,
+  DISABLED_NOTIFICATIONS_URL,
+} from "./default-notifications.service";
+import { SignalRNotification, SignalRConnectionService } from "./signalr-connection.service";
+import { WebPushConnectionService, WebPushConnector } from "./webpush-connection.service";
+import { WorkerWebPushConnectionService } from "./worker-webpush-connection.service";
+
+describe("NotificationsService", () => {
+  let syncService: MockProxy<SyncService>;
+  let appIdService: MockProxy<AppIdService>;
+  let environmentService: MockProxy<EnvironmentService>;
+  let logoutCallback: jest.Mock<Promise<void>, [logoutReason: LogoutReason]>;
+  let messagingService: MockProxy<MessageSender>;
+  let accountService: MockProxy<AccountService>;
+  let signalRNotificationConnectionService: MockProxy<SignalRConnectionService>;
+  let authService: MockProxy<AuthService>;
+  let webPushNotificationConnectionService: MockProxy<WebPushConnectionService>;
+
+  let activeAccount: BehaviorSubject<ObservedValueOf<AccountService["activeAccount$"]>>;
+
+  let environment: BehaviorSubject<ObservedValueOf<EnvironmentService["environment$"]>>;
+
+  let authStatusGetter: (userId: UserId) => BehaviorSubject<AuthenticationStatus>;
+
+  let webPushSupportGetter: (userId: UserId) => BehaviorSubject<SupportStatus<WebPushConnector>>;
+
+  let signalrNotificationGetter: (
+    userId: UserId,
+    notificationsUrl: string,
+  ) => Subject<SignalRNotification>;
+
+  let sut: DefaultNotificationsService;
+
+  beforeEach(() => {
+    syncService = mock<SyncService>();
+    appIdService = mock<AppIdService>();
+    environmentService = mock<EnvironmentService>();
+    logoutCallback = jest.fn<Promise<void>, [logoutReason: LogoutReason]>();
+    messagingService = mock<MessageSender>();
+    accountService = mock<AccountService>();
+    signalRNotificationConnectionService = mock<SignalRConnectionService>();
+    authService = mock<AuthService>();
+    webPushNotificationConnectionService = mock<WorkerWebPushConnectionService>();
+
+    activeAccount = new BehaviorSubject<ObservedValueOf<AccountService["activeAccount$"]>>(null);
+    accountService.activeAccount$ = activeAccount.asObservable();
+
+    environment = new BehaviorSubject<ObservedValueOf<EnvironmentService["environment$"]>>({
+      getNotificationsUrl: () => "https://notifications.bitwarden.com",
+    } as Environment);
+
+    environmentService.environment$ = environment;
+
+    authStatusGetter = Matrix.autoMockMethod(
+      authService.authStatusFor$,
+      () => new BehaviorSubject<AuthenticationStatus>(AuthenticationStatus.LoggedOut),
+    );
+
+    webPushSupportGetter = Matrix.autoMockMethod(
+      webPushNotificationConnectionService.supportStatus$,
+      () =>
+        new BehaviorSubject<SupportStatus<WebPushConnector>>({
+          type: "not-supported",
+          reason: "test",
+        }),
+    );
+
+    signalrNotificationGetter = Matrix.autoMockMethod(
+      signalRNotificationConnectionService.connect$,
+      () => new Subject<SignalRNotification>(),
+    );
+
+    sut = new DefaultNotificationsService(
+      syncService,
+      appIdService,
+      environmentService,
+      logoutCallback,
+      messagingService,
+      accountService,
+      signalRNotificationConnectionService,
+      authService,
+      webPushNotificationConnectionService,
+      mock<LogService>(),
+    );
+  });
+
+  const mockUser1 = "user1" as UserId;
+  const mockUser2 = "user2" as UserId;
+
+  function emitActiveUser(userId: UserId) {
+    if (userId == null) {
+      activeAccount.next(null);
+    } else {
+      activeAccount.next({ id: userId, email: "email", name: "Test Name", emailVerified: true });
+    }
+  }
+
+  function emitNotificationUrl(url: string) {
+    environment.next({
+      getNotificationsUrl: () => url,
+    } as Environment);
+  }
+
+  const expectNotification = (
+    notification: readonly [NotificationResponse, UserId],
+    expectedUser: UserId,
+    expectedType: NotificationType,
+  ) => {
+    const [actualNotification, actualUser] = notification;
+    expect(actualUser).toBe(expectedUser);
+    expect(actualNotification.type).toBe(expectedType);
+  };
+
+  it("emits notifications through WebPush when supported", async () => {
+    const notificationsPromise = firstValueFrom(sut.notifications$.pipe(bufferCount(2)));
+
+    emitActiveUser(mockUser1);
+    emitNotificationUrl("http://test.example.com");
+    authStatusGetter(mockUser1).next(AuthenticationStatus.Unlocked);
+
+    const webPush = mock<WebPushConnector>();
+    const webPushSubject = new Subject<NotificationResponse>();
+    webPush.notifications$ = webPushSubject;
+
+    webPushSupportGetter(mockUser1).next({ type: "supported", service: webPush });
+    webPushSubject.next(new NotificationResponse({ type: NotificationType.SyncFolderCreate }));
+    webPushSubject.next(new NotificationResponse({ type: NotificationType.SyncFolderDelete }));
+
+    const notifications = await notificationsPromise;
+    expectNotification(notifications[0], mockUser1, NotificationType.SyncFolderCreate);
+    expectNotification(notifications[1], mockUser1, NotificationType.SyncFolderDelete);
+  });
+
+  it("switches to SignalR when web push is not supported.", async () => {
+    const notificationsPromise = firstValueFrom(sut.notifications$.pipe(bufferCount(2)));
+
+    emitActiveUser(mockUser1);
+    emitNotificationUrl("http://test.example.com");
+    authStatusGetter(mockUser1).next(AuthenticationStatus.Unlocked);
+
+    const webPush = mock<WebPushConnector>();
+    const webPushSubject = new Subject<NotificationResponse>();
+    webPush.notifications$ = webPushSubject;
+
+    webPushSupportGetter(mockUser1).next({ type: "supported", service: webPush });
+    webPushSubject.next(new NotificationResponse({ type: NotificationType.SyncFolderCreate }));
+
+    emitActiveUser(mockUser2);
+    authStatusGetter(mockUser2).next(AuthenticationStatus.Unlocked);
+    // Second user does not support web push
+    webPushSupportGetter(mockUser2).next({ type: "not-supported", reason: "test" });
+
+    signalrNotificationGetter(mockUser2, "http://test.example.com").next({
+      type: "ReceiveMessage",
+      message: new NotificationResponse({ type: NotificationType.SyncCipherUpdate }),
+    });
+
+    const notifications = await notificationsPromise;
+    expectNotification(notifications[0], mockUser1, NotificationType.SyncFolderCreate);
+    expectNotification(notifications[1], mockUser2, NotificationType.SyncCipherUpdate);
+  });
+
+  it("switches to WebPush when it becomes supported.", async () => {
+    const notificationsPromise = firstValueFrom(sut.notifications$.pipe(bufferCount(2)));
+
+    emitActiveUser(mockUser1);
+    emitNotificationUrl("http://test.example.com");
+    authStatusGetter(mockUser1).next(AuthenticationStatus.Unlocked);
+    webPushSupportGetter(mockUser1).next({ type: "not-supported", reason: "test" });
+
+    signalrNotificationGetter(mockUser1, "http://test.example.com").next({
+      type: "ReceiveMessage",
+      message: new NotificationResponse({ type: NotificationType.AuthRequest }),
+    });
+
+    const webPush = mock<WebPushConnector>();
+    const webPushSubject = new Subject<NotificationResponse>();
+    webPush.notifications$ = webPushSubject;
+
+    webPushSupportGetter(mockUser1).next({ type: "supported", service: webPush });
+    webPushSubject.next(new NotificationResponse({ type: NotificationType.SyncLoginDelete }));
+
+    const notifications = await notificationsPromise;
+    expectNotification(notifications[0], mockUser1, NotificationType.AuthRequest);
+    expectNotification(notifications[1], mockUser1, NotificationType.SyncLoginDelete);
+  });
+
+  it("does not emit SignalR heartbeats", async () => {
+    const notificationsPromise = firstValueFrom(sut.notifications$.pipe(bufferCount(1)));
+
+    emitActiveUser(mockUser1);
+    emitNotificationUrl("http://test.example.com");
+    authStatusGetter(mockUser1).next(AuthenticationStatus.Unlocked);
+    webPushSupportGetter(mockUser1).next({ type: "not-supported", reason: "test" });
+
+    signalrNotificationGetter(mockUser1, "http://test.example.com").next({ type: "Heartbeat" });
+    signalrNotificationGetter(mockUser1, "http://test.example.com").next({
+      type: "ReceiveMessage",
+      message: new NotificationResponse({ type: NotificationType.AuthRequestResponse }),
+    });
+
+    const notifications = await notificationsPromise;
+    expectNotification(notifications[0], mockUser1, NotificationType.AuthRequestResponse);
+  });
+
+  it.each([
+    { initialStatus: AuthenticationStatus.Locked, updatedStatus: AuthenticationStatus.Unlocked },
+    { initialStatus: AuthenticationStatus.Unlocked, updatedStatus: AuthenticationStatus.Locked },
+    { initialStatus: AuthenticationStatus.Locked, updatedStatus: AuthenticationStatus.Locked },
+    { initialStatus: AuthenticationStatus.Unlocked, updatedStatus: AuthenticationStatus.Unlocked },
+  ])(
+    "does not re-connect when the user transitions from $initialStatus to $updatedStatus",
+    async ({ initialStatus, updatedStatus }) => {
+      emitActiveUser(mockUser1);
+      emitNotificationUrl("http://test.example.com");
+      authStatusGetter(mockUser1).next(initialStatus);
+      webPushSupportGetter(mockUser1).next({ type: "not-supported", reason: "test" });
+
+      const notificationsSubscriptions = sut.notifications$.subscribe();
+      await awaitAsync(1);
+
+      authStatusGetter(mockUser1).next(updatedStatus);
+      await awaitAsync(1);
+
+      expect(signalRNotificationConnectionService.connect$).toHaveBeenCalledTimes(1);
+      expect(signalRNotificationConnectionService.connect$).toHaveBeenCalledWith(
+        mockUser1,
+        "http://test.example.com",
+      );
+      notificationsSubscriptions.unsubscribe();
+    },
+  );
+
+  it.each([AuthenticationStatus.Locked, AuthenticationStatus.Unlocked])(
+    "connects when a user transitions from logged out to %s",
+    async (newStatus: AuthenticationStatus) => {
+      emitActiveUser(mockUser1);
+      emitNotificationUrl("http://test.example.com");
+      authStatusGetter(mockUser1).next(AuthenticationStatus.LoggedOut);
+      webPushSupportGetter(mockUser1).next({ type: "not-supported", reason: "test" });
+
+      const notificationsSubscriptions = sut.notifications$.subscribe();
+      await awaitAsync(1);
+
+      authStatusGetter(mockUser1).next(newStatus);
+      await awaitAsync(1);
+
+      expect(signalRNotificationConnectionService.connect$).toHaveBeenCalledTimes(1);
+      expect(signalRNotificationConnectionService.connect$).toHaveBeenCalledWith(
+        mockUser1,
+        "http://test.example.com",
+      );
+      notificationsSubscriptions.unsubscribe();
+    },
+  );
+
+  it("does not connect to any notification stream when notifications are disabled through special url", () => {
+    const subscription = sut.notifications$.subscribe();
+    emitActiveUser(mockUser1);
+    emitNotificationUrl(DISABLED_NOTIFICATIONS_URL);
+
+    expect(signalRNotificationConnectionService.connect$).not.toHaveBeenCalled();
+    expect(webPushNotificationConnectionService.supportStatus$).not.toHaveBeenCalled();
+
+    subscription.unsubscribe();
+  });
+
+  it("does not connect to any notification stream when there is no active user", () => {
+    const subscription = sut.notifications$.subscribe();
+    emitActiveUser(null);
+
+    expect(signalRNotificationConnectionService.connect$).not.toHaveBeenCalled();
+    expect(webPushNotificationConnectionService.supportStatus$).not.toHaveBeenCalled();
+
+    subscription.unsubscribe();
+  });
+
+  it("does not reconnect if the same notification url is emitted", async () => {
+    const subscription = sut.notifications$.subscribe();
+
+    emitActiveUser(mockUser1);
+    emitNotificationUrl("http://test.example.com");
+    authStatusGetter(mockUser1).next(AuthenticationStatus.Unlocked);
+
+    await awaitAsync(1);
+
+    expect(webPushNotificationConnectionService.supportStatus$).toHaveBeenCalledTimes(1);
+    emitNotificationUrl("http://test.example.com");
+
+    await awaitAsync(1);
+
+    expect(webPushNotificationConnectionService.supportStatus$).toHaveBeenCalledTimes(1);
+    subscription.unsubscribe();
+  });
+});

libs/common/src/platform/notifications/internal/default-notifications.service.ts

@@ -0,0 +1,238 @@
+import {
+  BehaviorSubject,
+  catchError,
+  distinctUntilChanged,
+  EMPTY,
+  filter,
+  map,
+  mergeMap,
+  Observable,
+  switchMap,
+} from "rxjs";
+
+import { LogoutReason } from "@bitwarden/auth/common";
+
+import { AccountService } from "../../../auth/abstractions/account.service";
+import { AuthService } from "../../../auth/abstractions/auth.service";
+import { AuthenticationStatus } from "../../../auth/enums/authentication-status";
+import { NotificationType } from "../../../enums";
+import {
+  NotificationResponse,
+  SyncCipherNotification,
+  SyncFolderNotification,
+  SyncSendNotification,
+} from "../../../models/response/notification.response";
+import { UserId } from "../../../types/guid";
+import { SyncService } from "../../../vault/abstractions/sync/sync.service.abstraction";
+import { AppIdService } from "../../abstractions/app-id.service";
+import { EnvironmentService } from "../../abstractions/environment.service";
+import { LogService } from "../../abstractions/log.service";
+import { MessagingService } from "../../abstractions/messaging.service";
+import { supportSwitch } from "../../misc/support-status";
+import { NotificationsService as NotificationsServiceAbstraction } from "../notifications.service";
+
+import { ReceiveMessage, SignalRConnectionService } from "./signalr-connection.service";
+import { WebPushConnectionService } from "./webpush-connection.service";
+
+export const DISABLED_NOTIFICATIONS_URL = "http://-";
+
+export class DefaultNotificationsService implements NotificationsServiceAbstraction {
+  notifications$: Observable<readonly [NotificationResponse, UserId]>;
+
+  private activitySubject = new BehaviorSubject<"active" | "inactive">("active");
+
+  constructor(
+    private syncService: SyncService,
+    private appIdService: AppIdService,
+    private environmentService: EnvironmentService,
+    private logoutCallback: (logoutReason: LogoutReason, userId: UserId) => Promise<void>,
+    private messagingService: MessagingService,
+    private readonly accountService: AccountService,
+    private readonly signalRConnectionService: SignalRConnectionService,
+    private readonly authService: AuthService,
+    private readonly webPushConnectionService: WebPushConnectionService,
+    private readonly logService: LogService,
+  ) {
+    this.notifications$ = this.accountService.activeAccount$.pipe(
+      map((account) => account?.id),
+      distinctUntilChanged(),
+      switchMap((activeAccountId) => {
+        if (activeAccountId == null) {
+          // We don't emit notifications for inactive accounts currently
+          return EMPTY;
+        }
+
+        return this.userNotifications$(activeAccountId).pipe(
+          map((notification) => [notification, activeAccountId] as const),
+        );
+      }),
+    );
+  }
+
+  /**
+   * Retrieves a stream of push notifications for the given user.
+   * @param userId The user id of the user to get the push notifications for.
+   */
+  private userNotifications$(userId: UserId) {
+    return this.environmentService.environment$.pipe(
+      map((env) => env.getNotificationsUrl()),
+      distinctUntilChanged(),
+      switchMap((notificationsUrl) => {
+        if (notificationsUrl === DISABLED_NOTIFICATIONS_URL) {
+          return EMPTY;
+        }
+
+        return this.userNotificationsHelper$(userId, notificationsUrl);
+      }),
+    );
+  }
+
+  private userNotificationsHelper$(userId: UserId, notificationsUrl: string) {
+    return this.hasAccessToken$(userId).pipe(
+      switchMap((hasAccessToken) => {
+        if (!hasAccessToken) {
+          return EMPTY;
+        }
+
+        return this.activitySubject;
+      }),
+      switchMap((activityStatus) => {
+        if (activityStatus === "inactive") {
+          return EMPTY;
+        }
+
+        return this.webPushConnectionService.supportStatus$(userId);
+      }),
+      supportSwitch({
+        supported: (service) =>
+          service.notifications$.pipe(
+            catchError((err: unknown) => {
+              this.logService.warning("Issue with web push, falling back to SignalR", err);
+              return this.connectSignalR$(userId, notificationsUrl);
+            }),
+          ),
+        notSupported: () => this.connectSignalR$(userId, notificationsUrl),
+      }),
+    );
+  }
+
+  private connectSignalR$(userId: UserId, notificationsUrl: string) {
+    return this.signalRConnectionService.connect$(userId, notificationsUrl).pipe(
+      filter((n) => n.type === "ReceiveMessage"),
+      map((n) => (n as ReceiveMessage).message),
+    );
+  }
+
+  private hasAccessToken$(userId: UserId) {
+    return this.authService.authStatusFor$(userId).pipe(
+      map(
+        (authStatus) =>
+          authStatus === AuthenticationStatus.Locked ||
+          authStatus === AuthenticationStatus.Unlocked,
+      ),
+      distinctUntilChanged(),
+    );
+  }
+
+  private async processNotification(notification: NotificationResponse, userId: UserId) {
+    const appId = await this.appIdService.getAppId();
+    if (notification == null || notification.contextId === appId) {
+      return;
+    }
+
+    const payloadUserId = notification.payload?.userId || notification.payload?.UserId;
+    if (payloadUserId != null && payloadUserId !== userId) {
+      return;
+    }
+
+    switch (notification.type) {
+      case NotificationType.SyncCipherCreate:
+      case NotificationType.SyncCipherUpdate:
+        await this.syncService.syncUpsertCipher(
+          notification.payload as SyncCipherNotification,
+          notification.type === NotificationType.SyncCipherUpdate,
+        );
+        break;
+      case NotificationType.SyncCipherDelete:
+      case NotificationType.SyncLoginDelete:
+        await this.syncService.syncDeleteCipher(notification.payload as SyncCipherNotification);
+        break;
+      case NotificationType.SyncFolderCreate:
+      case NotificationType.SyncFolderUpdate:
+        await this.syncService.syncUpsertFolder(
+          notification.payload as SyncFolderNotification,
+          notification.type === NotificationType.SyncFolderUpdate,
+          userId,
+        );
+        break;
+      case NotificationType.SyncFolderDelete:
+        await this.syncService.syncDeleteFolder(
+          notification.payload as SyncFolderNotification,
+          userId,
+        );
+        break;
+      case NotificationType.SyncVault:
+      case NotificationType.SyncCiphers:
+      case NotificationType.SyncSettings:
+        await this.syncService.fullSync(false);
+
+        break;
+      case NotificationType.SyncOrganizations:
+        // An organization update may not have bumped the user's account revision date, so force a sync
+        await this.syncService.fullSync(true);
+        break;
+      case NotificationType.SyncOrgKeys:
+        await this.syncService.fullSync(true);
+        this.activitySubject.next("inactive"); // Force a disconnect
+        this.activitySubject.next("active"); // Allow a reconnect
+        break;
+      case NotificationType.LogOut:
+        this.logService.info("[Notifications Service] Received logout notification");
+        await this.logoutCallback("logoutNotification", userId);
+        break;
+      case NotificationType.SyncSendCreate:
+      case NotificationType.SyncSendUpdate:
+        await this.syncService.syncUpsertSend(
+          notification.payload as SyncSendNotification,
+          notification.type === NotificationType.SyncSendUpdate,
+        );
+        break;
+      case NotificationType.SyncSendDelete:
+        await this.syncService.syncDeleteSend(notification.payload as SyncSendNotification);
+        break;
+      case NotificationType.AuthRequest:
+        {
+          this.messagingService.send("openLoginApproval", {
+            notificationId: notification.payload.id,
+          });
+        }
+        break;
+      case NotificationType.SyncOrganizationStatusChanged:
+        await this.syncService.fullSync(true);
+        break;
+      case NotificationType.SyncOrganizationCollectionSettingChanged:
+        await this.syncService.fullSync(true);
+        break;
+      default:
+        break;
+    }
+  }
+
+  startListening() {
+    return this.notifications$
+      .pipe(
+        mergeMap(async ([notification, userId]) => this.processNotification(notification, userId)),
+      )
+      .subscribe({
+        error: (e: unknown) => this.logService.warning("Error in notifications$ observable", e),
+      });
+  }
+
+  reconnectFromActivity(): void {
+    this.activitySubject.next("active");
+  }
+
+  disconnectFromInactivity(): void {
+    this.activitySubject.next("inactive");
+  }
+}

libs/common/src/platform/notifications/internal/index.ts

@@ -0,0 +1,8 @@
+export * from "./worker-webpush-connection.service";
+export * from "./signalr-connection.service";
+export * from "./default-notifications.service";
+export * from "./noop-notifications.service";
+export * from "./unsupported-webpush-connection.service";
+export * from "./webpush-connection.service";
+export * from "./websocket-webpush-connection.service";
+export * from "./web-push-notifications-api.service";

libs/common/src/platform/notifications/internal/noop-notifications.service.ts

@@ -0,0 +1,23 @@
+import { Subscription } from "rxjs";
+
+import { LogService } from "../../abstractions/log.service";
+import { NotificationsService } from "../notifications.service";
+
+export class NoopNotificationsService implements NotificationsService {
+  constructor(private logService: LogService) {}
+
+  startListening(): Subscription {
+    this.logService.info(
+      "Initializing no-op notification service, no push notifications will be received",
+    );
+    return Subscription.EMPTY;
+  }
+
+  reconnectFromActivity(): void {
+    this.logService.info("Reconnecting notification service from activity");
+  }
+
+  disconnectFromInactivity(): void {
+    this.logService.info("Disconnecting notification service from inactivity");
+  }
+}

libs/common/src/platform/notifications/internal/signalr-connection.service.ts

@@ -0,0 +1,125 @@
+import {
+  HttpTransportType,
+  HubConnectionBuilder,
+  HubConnectionState,
+  ILogger,
+  LogLevel,
+} from "@microsoft/signalr";
+import { MessagePackHubProtocol } from "@microsoft/signalr-protocol-msgpack";
+import { Observable, Subscription } from "rxjs";
+
+import { ApiService } from "../../../abstractions/api.service";
+import { NotificationResponse } from "../../../models/response/notification.response";
+import { UserId } from "../../../types/guid";
+import { LogService } from "../../abstractions/log.service";
+
+// 2 Minutes
+const MIN_RECONNECT_TIME = 2 * 60 * 1000;
+// 5 Minutes
+const MAX_RECONNECT_TIME = 5 * 60 * 1000;
+
+export type Heartbeat = { type: "Heartbeat" };
+export type ReceiveMessage = { type: "ReceiveMessage"; message: NotificationResponse };
+
+export type SignalRNotification = Heartbeat | ReceiveMessage;
+
+class SignalRLogger implements ILogger {
+  constructor(private readonly logService: LogService) {}
+
+  log(logLevel: LogLevel, message: string): void {
+    switch (logLevel) {
+      case LogLevel.Critical:
+        this.logService.error(message);
+        break;
+      case LogLevel.Error:
+        this.logService.error(message);
+        break;
+      case LogLevel.Warning:
+        this.logService.warning(message);
+        break;
+      case LogLevel.Information:
+        this.logService.info(message);
+        break;
+      case LogLevel.Debug:
+        this.logService.debug(message);
+        break;
+    }
+  }
+}
+
+export class SignalRConnectionService {
+  constructor(
+    private readonly apiService: ApiService,
+    private readonly logService: LogService,
+  ) {}
+
+  connect$(userId: UserId, notificationsUrl: string) {
+    return new Observable<SignalRNotification>((subsciber) => {
+      const connection = new HubConnectionBuilder()
+        .withUrl(notificationsUrl + "/hub", {
+          accessTokenFactory: () => this.apiService.getActiveBearerToken(),
+          skipNegotiation: true,
+          transport: HttpTransportType.WebSockets,
+        })
+        .withHubProtocol(new MessagePackHubProtocol())
+        .configureLogging(new SignalRLogger(this.logService))
+        .build();
+
+      connection.on("ReceiveMessage", (data: any) => {
+        subsciber.next({ type: "ReceiveMessage", message: new NotificationResponse(data) });
+      });
+
+      connection.on("Heartbeat", () => {
+        subsciber.next({ type: "Heartbeat" });
+      });
+
+      let reconnectSubscription: Subscription | null = null;
+
+      // Create schedule reconnect function
+      const scheduleReconnect = (): Subscription => {
+        if (
+          connection == null ||
+          connection.state !== HubConnectionState.Disconnected ||
+          (reconnectSubscription != null && !reconnectSubscription.closed)
+        ) {
+          return Subscription.EMPTY;
+        }
+
+        const randomTime = this.random();
+        const timeoutHandler = setTimeout(() => {
+          connection
+            .start()
+            .then(() => (reconnectSubscription = null))
+            .catch(() => {
+              reconnectSubscription = scheduleReconnect();
+            });
+        }, randomTime);
+
+        return new Subscription(() => clearTimeout(timeoutHandler));
+      };
+
+      connection.onclose((error) => {
+        reconnectSubscription = scheduleReconnect();
+      });
+
+      // Start connection
+      connection.start().catch(() => {
+        reconnectSubscription = scheduleReconnect();
+      });
+
+      return () => {
+        connection?.stop().catch((error) => {
+          this.logService.error("Error while stopping SignalR connection", error);
+          // TODO: Does calling stop call `onclose`?
+          reconnectSubscription?.unsubscribe();
+        });
+      };
+    });
+  }
+
+  private random() {
+    return (
+      Math.floor(Math.random() * (MAX_RECONNECT_TIME - MIN_RECONNECT_TIME + 1)) + MIN_RECONNECT_TIME
+    );
+  }
+}

libs/common/src/platform/notifications/internal/unsupported-webpush-connection.service.ts

@@ -0,0 +1,15 @@
+import { Observable, of } from "rxjs";
+
+import { UserId } from "../../../types/guid";
+import { SupportStatus } from "../../misc/support-status";
+
+import { WebPushConnectionService, WebPushConnector } from "./webpush-connection.service";
+
+/**
+ * An implementation of {@see WebPushConnectionService} for clients that do not have support for WebPush
+ */
+export class UnsupportedWebPushConnectionService implements WebPushConnectionService {
+  supportStatus$(userId: UserId): Observable<SupportStatus<WebPushConnector>> {
+    return of({ type: "not-supported", reason: "client-not-supported" });
+  }
+}

libs/common/src/platform/notifications/internal/web-push-notifications-api.service.ts

@@ -0,0 +1,25 @@
+import { ApiService } from "../../../abstractions/api.service";
+import { AppIdService } from "../../abstractions/app-id.service";
+
+import { WebPushRequest } from "./web-push.request";
+
+export class WebPushNotificationsApiService {
+  constructor(
+    private readonly apiService: ApiService,
+    private readonly appIdService: AppIdService,
+  ) {}
+
+  /**
+   * Posts a device-user association to the server and ensures it's installed for push notifications
+   */
+  async putSubscription(pushSubscription: PushSubscriptionJSON): Promise<void> {
+    const request = WebPushRequest.from(pushSubscription);
+    await this.apiService.send(
+      "POST",
+      `/devices/identifier/${await this.appIdService.getAppId()}/web-push-auth`,
+      request,
+      true,
+      false,
+    );
+  }
+}

libs/common/src/platform/notifications/internal/web-push.request.ts

@@ -0,0 +1,13 @@
+export class WebPushRequest {
+  endpoint: string | undefined;
+  p256dh: string | undefined;
+  auth: string | undefined;
+
+  static from(pushSubscription: PushSubscriptionJSON): WebPushRequest {
+    const result = new WebPushRequest();
+    result.endpoint = pushSubscription.endpoint;
+    result.p256dh = pushSubscription.keys?.p256dh;
+    result.auth = pushSubscription.keys?.auth;
+    return result;
+  }
+}

libs/common/src/platform/notifications/internal/webpush-connection.service.ts

@@ -0,0 +1,13 @@
+import { Observable } from "rxjs";
+
+import { NotificationResponse } from "../../../models/response/notification.response";
+import { UserId } from "../../../types/guid";
+import { SupportStatus } from "../../misc/support-status";
+
+export interface WebPushConnector {
+  notifications$: Observable<NotificationResponse>;
+}
+
+export abstract class WebPushConnectionService {
+  abstract supportStatus$(userId: UserId): Observable<SupportStatus<WebPushConnector>>;
+}

libs/common/src/platform/notifications/internal/websocket-webpush-connection.service.ts

@@ -0,0 +1,12 @@
+import { Observable, of } from "rxjs";
+
+import { UserId } from "../../../types/guid";
+import { SupportStatus } from "../../misc/support-status";
+
+import { WebPushConnectionService, WebPushConnector } from "./webpush-connection.service";
+
+export class WebSocketWebPushConnectionService implements WebPushConnectionService {
+  supportStatus$(userId: UserId): Observable<SupportStatus<WebPushConnector>> {
+    return of({ type: "not-supported", reason: "work-in-progress" });
+  }
+}

libs/common/src/platform/notifications/internal/worker-webpush-connection.service.ts

@@ -0,0 +1,168 @@
+import {
+  concat,
+  concatMap,
+  defer,
+  distinctUntilChanged,
+  fromEvent,
+  map,
+  Observable,
+  Subject,
+  Subscription,
+  switchMap,
+} from "rxjs";
+
+import { PushTechnology } from "../../../enums/push-technology.enum";
+import { NotificationResponse } from "../../../models/response/notification.response";
+import { UserId } from "../../../types/guid";
+import { ConfigService } from "../../abstractions/config/config.service";
+import { SupportStatus } from "../../misc/support-status";
+import { Utils } from "../../misc/utils";
+
+import { WebPushNotificationsApiService } from "./web-push-notifications-api.service";
+import { WebPushConnectionService, WebPushConnector } from "./webpush-connection.service";
+
+// Ref: https://w3c.github.io/push-api/#the-pushsubscriptionchange-event
+interface PushSubscriptionChangeEvent {
+  readonly newSubscription?: PushSubscription;
+  readonly oldSubscription?: PushSubscription;
+}
+
+// Ref: https://developer.mozilla.org/en-US/docs/Web/API/PushMessageData
+interface PushMessageData {
+  json(): any;
+}
+
+// Ref: https://developer.mozilla.org/en-US/docs/Web/API/PushEvent
+interface PushEvent {
+  data: PushMessageData;
+}
+
+/**
+ * An implementation for connecting to web push based notifications running in a Worker.
+ */
+export class WorkerWebPushConnectionService implements WebPushConnectionService {
+  private pushEvent = new Subject<PushEvent>();
+  private pushChangeEvent = new Subject<PushSubscriptionChangeEvent>();
+
+  constructor(
+    private readonly configService: ConfigService,
+    private readonly webPushApiService: WebPushNotificationsApiService,
+    private readonly serviceWorkerRegistration: ServiceWorkerRegistration,
+  ) {}
+
+  start(): Subscription {
+    const subscription = new Subscription(() => {
+      this.pushEvent.complete();
+      this.pushChangeEvent.complete();
+      this.pushEvent = new Subject<PushEvent>();
+      this.pushChangeEvent = new Subject<PushSubscriptionChangeEvent>();
+    });
+
+    const pushEventSubscription = fromEvent<PushEvent>(self, "push").subscribe(this.pushEvent);
+
+    const pushChangeEventSubscription = fromEvent<PushSubscriptionChangeEvent>(
+      self,
+      "pushsubscriptionchange",
+    ).subscribe(this.pushChangeEvent);
+
+    subscription.add(pushEventSubscription);
+    subscription.add(pushChangeEventSubscription);
+
+    return subscription;
+  }
+
+  supportStatus$(userId: UserId): Observable<SupportStatus<WebPushConnector>> {
+    // Check the server config to see if it supports sending WebPush notifications
+    // FIXME: get config of server for the specified userId, once ConfigService supports it
+    return this.configService.serverConfig$.pipe(
+      map((config) =>
+        config?.push?.pushTechnology === PushTechnology.WebPush ? config.push.vapidPublicKey : null,
+      ),
+      // No need to re-emit when there is new server config if the vapidPublicKey is still there and the exact same
+      distinctUntilChanged(),
+      map((publicKey) => {
+        if (publicKey == null) {
+          return {
+            type: "not-supported",
+            reason: "server-not-configured",
+          } satisfies SupportStatus<WebPushConnector>;
+        }
+
+        return {
+          type: "supported",
+          service: new MyWebPushConnector(
+            publicKey,
+            userId,
+            this.webPushApiService,
+            this.serviceWorkerRegistration,
+            this.pushEvent,
+            this.pushChangeEvent,
+          ),
+        } satisfies SupportStatus<WebPushConnector>;
+      }),
+    );
+  }
+}
+
+class MyWebPushConnector implements WebPushConnector {
+  notifications$: Observable<NotificationResponse>;
+
+  constructor(
+    private readonly vapidPublicKey: string,
+    private readonly userId: UserId,
+    private readonly webPushApiService: WebPushNotificationsApiService,
+    private readonly serviceWorkerRegistration: ServiceWorkerRegistration,
+    private readonly pushEvent$: Observable<PushEvent>,
+    private readonly pushChangeEvent$: Observable<PushSubscriptionChangeEvent>,
+  ) {
+    this.notifications$ = this.getOrCreateSubscription$(this.vapidPublicKey).pipe(
+      concatMap((subscription) => {
+        return defer(() => {
+          if (subscription == null) {
+            throw new Error("Expected a non-null subscription.");
+          }
+          return this.webPushApiService.putSubscription(subscription.toJSON());
+        }).pipe(
+          switchMap(() => this.pushEvent$),
+          map((e) => new NotificationResponse(e.data.json().data)),
+        );
+      }),
+    );
+  }
+
+  private async pushManagerSubscribe(key: string) {
+    return await this.serviceWorkerRegistration.pushManager.subscribe({
+      userVisibleOnly: true,
+      applicationServerKey: key,
+    });
+  }
+
+  private getOrCreateSubscription$(key: string) {
+    return concat(
+      defer(async () => {
+        const existingSubscription =
+          await this.serviceWorkerRegistration.pushManager.getSubscription();
+
+        if (existingSubscription == null) {
+          return await this.pushManagerSubscribe(key);
+        }
+
+        const subscriptionKey = Utils.fromBufferToUrlB64(
+          // REASON: `Utils.fromBufferToUrlB64` handles null by returning null back to it.
+          // its annotation should be updated and then this assertion can be removed.
+          // eslint-disable-next-line @typescript-eslint/no-non-null-asserted-optional-chain
+          existingSubscription.options?.applicationServerKey!,
+        );
+
+        if (subscriptionKey !== key) {
+          // There is a subscription, but it's not for the current server, unsubscribe and then make a new one
+          await existingSubscription.unsubscribe();
+          return await this.pushManagerSubscribe(key);
+        }
+
+        return existingSubscription;
+      }),
+      this.pushChangeEvent$.pipe(map((event) => event.newSubscription)),
+    );
+  }
+}

libs/common/src/platform/notifications/notifications.service.ts

@@ -0,0 +1,18 @@
+import { Subscription } from "rxjs";
+
+/**
+ * A service offering abilities to interact with push notifications from the server.
+ */
+export abstract class NotificationsService {
+  /**
+   * Starts automatic listening and processing of notifications, should only be called once per application,
+   * or you will risk notifications being processed multiple times.
+   */
+  abstract startListening(): Subscription;
+  // TODO: Delete this method in favor of an `ActivityService` that notifications can depend on.
+  // https://bitwarden.atlassian.net/browse/PM-14264
+  abstract reconnectFromActivity(): void;
+  // TODO: Delete this method in favor of an `ActivityService` that notifications can depend on.
+  // https://bitwarden.atlassian.net/browse/PM-14264
+  abstract disconnectFromInactivity(): void;
+}

libs/common/src/platform/services/noop-notifications.service.ts

@@ -1,28 +0,0 @@
-import { NotificationsService as NotificationsServiceAbstraction } from "../../abstractions/notifications.service";
-import { LogService } from "../abstractions/log.service";
-
-export class NoopNotificationsService implements NotificationsServiceAbstraction {
-  constructor(private logService: LogService) {}
-
-  init(): Promise<void> {
-    this.logService.info(
-      "Initializing no-op notification service, no push notifications will be received",
-    );
-    return Promise.resolve();
-  }
-
-  updateConnection(sync?: boolean): Promise<void> {
-    this.logService.info("Updating notification service connection");
-    return Promise.resolve();
-  }
-
-  reconnectFromActivity(): Promise<void> {
-    this.logService.info("Reconnecting notification service from activity");
-    return Promise.resolve();
-  }
-
-  disconnectFromInactivity(): Promise<void> {
-    this.logService.info("Disconnecting notification service from inactivity");
-    return Promise.resolve();
-  }
-}