[PM-17563] Security task background synchronization (#14086)

* [PM-17563] Implement listenForTaskNotifications in default-task.service.ts

* [PM-17563] Update syncService to include userId in syncCompleted message payload

* [PM-17563] Update default-task.service to react to both pending task notifications and completed syncs

* [PM-17563] Add unit tests around task notification listening

* [PM-17563] Only check for at risk password tasks if tasks are enabled

* [PM-17563] Make userId required even if undefined

* [PM-17563] Use abstract TaskService instead of default implementation in MainBackground

* [PM-17563] Cleanup userId filtering

libs/common/src/vault/tasks/abstractions/task.service.ts

@@ -1,4 +1,4 @@
-import { Observable } from "rxjs";
+import { Observable, Subscription } from "rxjs";
 
 import { SecurityTaskId, UserId } from "@bitwarden/common/types/guid";
 
@@ -43,4 +43,9 @@ export abstract class TaskService {
    * @param userId - The user who is completing the task.
    */
   abstract markAsComplete(taskId: SecurityTaskId, userId: UserId): Promise<void>;
+
+  /**
+   * Creates a subscription for pending security task notifications or completed syncs for unlocked users.
+   */
+  abstract listenForTaskNotifications(): Subscription;
 }

libs/common/src/vault/tasks/services/default-task.service.spec.ts

@@ -1,9 +1,15 @@
-import { BehaviorSubject, firstValueFrom } from "rxjs";
+import { BehaviorSubject, firstValueFrom, Subject } from "rxjs";
 
 import { ApiService } from "@bitwarden/common/abstractions/api.service";
 import { OrganizationService } from "@bitwarden/common/admin-console/abstractions/organization/organization.service.abstraction";
 import { Organization } from "@bitwarden/common/admin-console/models/domain/organization";
+import { AuthService } from "@bitwarden/common/auth/abstractions/auth.service";
+import { AuthenticationStatus } from "@bitwarden/common/auth/enums/authentication-status";
+import { NotificationType } from "@bitwarden/common/enums";
+import { NotificationResponse } from "@bitwarden/common/models/response/notification.response";
 import { ConfigService } from "@bitwarden/common/platform/abstractions/config/config.service";
+import { Message, MessageListener } from "@bitwarden/common/platform/messaging";
+import { NotificationsService } from "@bitwarden/common/platform/notifications";
 import { SecurityTaskId, UserId } from "@bitwarden/common/types/guid";
 
 import { FakeStateProvider, mockAccountServiceWith } from "../../../../spec";
@@ -16,10 +22,13 @@ import { DefaultTaskService } from "./default-task.service";
 describe("Default task service", () => {
   let fakeStateProvider: FakeStateProvider;
 
+  const userId = "user-id" as UserId;
   const mockApiSend = jest.fn();
   const mockGetAllOrgs$ = jest.fn();
   const mockGetFeatureFlag$ = jest.fn();
-
+  const mockAuthStatuses$ = new BehaviorSubject<Record<UserId, AuthenticationStatus>>({});
+  const mockNotifications$ = new Subject<readonly [NotificationResponse, UserId]>();
+  const mockMessages$ = new Subject<Message<Record<string, unknown>>>();
   let service: DefaultTaskService;
 
   beforeEach(async () => {
@@ -27,12 +36,15 @@ describe("Default task service", () => {
     mockGetAllOrgs$.mockClear();
     mockGetFeatureFlag$.mockClear();
 
-    fakeStateProvider = new FakeStateProvider(mockAccountServiceWith("user-id" as UserId));
+    fakeStateProvider = new FakeStateProvider(mockAccountServiceWith(userId));
     service = new DefaultTaskService(
       fakeStateProvider,
       { send: mockApiSend } as unknown as ApiService,
       { organizations$: mockGetAllOrgs$ } as unknown as OrganizationService,
       { getFeatureFlag$: mockGetFeatureFlag$ } as unknown as ConfigService,
+      { authStatuses$: mockAuthStatuses$.asObservable() } as unknown as AuthService,
+      { notifications$: mockNotifications$.asObservable() } as unknown as NotificationsService,
+      { allMessages$: mockMessages$.asObservable() } as unknown as MessageListener,
     );
   });
 
@@ -257,4 +269,235 @@ describe("Default task service", () => {
       ]);
     });
   });
+
+  describe("listenForTaskNotifications()", () => {
+    it("should not subscribe to notifications when there are no unlocked users", () => {
+      mockAuthStatuses$.next({
+        [userId]: AuthenticationStatus.Locked,
+      });
+
+      service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+      const notificationHelper$ = (service["securityTaskNotifications$"] = jest.fn());
+      const syncCompletedHelper$ = (service["syncCompletedMessage$"] = jest.fn());
+
+      const subscription = service.listenForTaskNotifications();
+
+      expect(notificationHelper$).not.toHaveBeenCalled();
+      expect(syncCompletedHelper$).not.toHaveBeenCalled();
+      subscription.unsubscribe();
+    });
+
+    it("should not subscribe to notifications when no users have tasks enabled", () => {
+      mockAuthStatuses$.next({
+        [userId]: AuthenticationStatus.Unlocked,
+      });
+
+      service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(false));
+      const notificationHelper$ = (service["securityTaskNotifications$"] = jest.fn());
+      const syncCompletedHelper$ = (service["syncCompletedMessage$"] = jest.fn());
+
+      const subscription = service.listenForTaskNotifications();
+
+      expect(notificationHelper$).not.toHaveBeenCalled();
+      expect(syncCompletedHelper$).not.toHaveBeenCalled();
+      subscription.unsubscribe();
+    });
+
+    it("should subscribe to notifications when there are unlocked users with tasks enabled", () => {
+      mockAuthStatuses$.next({
+        [userId]: AuthenticationStatus.Unlocked,
+      });
+      service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+
+      const notificationHelper$ = (service["securityTaskNotifications$"] = jest.fn());
+      const syncCompletedHelper$ = (service["syncCompletedMessage$"] = jest.fn());
+
+      const subscription = service.listenForTaskNotifications();
+
+      expect(notificationHelper$).toHaveBeenCalled();
+      expect(syncCompletedHelper$).toHaveBeenCalled();
+      subscription.unsubscribe();
+    });
+
+    describe("notification handling", () => {
+      it("should refresh tasks when a notification is received for an allowed user", async () => {
+        mockAuthStatuses$.next({
+          [userId]: AuthenticationStatus.Unlocked,
+        });
+        service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+
+        const syncCompletedHelper$ = (service["syncCompletedMessage$"] = jest.fn(
+          () => new Subject(),
+        ));
+        const refreshTasks = jest.spyOn(service, "refreshTasks");
+
+        const subscription = service.listenForTaskNotifications();
+
+        const notification = {
+          type: NotificationType.PendingSecurityTasks,
+        } as NotificationResponse;
+        mockNotifications$.next([notification, userId]);
+
+        await new Promise(process.nextTick);
+
+        expect(syncCompletedHelper$).toHaveBeenCalled();
+        expect(refreshTasks).toHaveBeenCalledWith(userId);
+        subscription.unsubscribe();
+      });
+
+      it("should ignore notifications for other users", async () => {
+        mockAuthStatuses$.next({
+          [userId]: AuthenticationStatus.Unlocked,
+        });
+        service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+
+        const syncCompletedHelper$ = (service["syncCompletedMessage$"] = jest.fn(
+          () => new Subject(),
+        ));
+        const refreshTasks = jest.spyOn(service, "refreshTasks");
+
+        const subscription = service.listenForTaskNotifications();
+
+        const notification = {
+          type: NotificationType.PendingSecurityTasks,
+        } as NotificationResponse;
+        mockNotifications$.next([notification, "other-user-id" as UserId]);
+
+        await new Promise(process.nextTick);
+
+        expect(syncCompletedHelper$).toHaveBeenCalled();
+        expect(refreshTasks).not.toHaveBeenCalledWith(userId);
+        subscription.unsubscribe();
+      });
+
+      it("should ignore other notifications types", async () => {
+        mockAuthStatuses$.next({
+          [userId]: AuthenticationStatus.Unlocked,
+        });
+        service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+
+        const syncCompletedHelper$ = (service["syncCompletedMessage$"] = jest.fn(
+          () => new Subject(),
+        ));
+        const refreshTasks = jest.spyOn(service, "refreshTasks");
+
+        const subscription = service.listenForTaskNotifications();
+
+        const notification = {
+          type: NotificationType.SyncSettings,
+        } as NotificationResponse;
+        mockNotifications$.next([notification, userId]);
+
+        await new Promise(process.nextTick);
+
+        expect(syncCompletedHelper$).toHaveBeenCalled();
+        expect(refreshTasks).not.toHaveBeenCalledWith(userId);
+        subscription.unsubscribe();
+      });
+    });
+
+    describe("sync completed handling", () => {
+      it("should refresh tasks when a sync completed message is received for an allowed user", async () => {
+        mockAuthStatuses$.next({
+          [userId]: AuthenticationStatus.Unlocked,
+        });
+        service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+
+        const notificationHelper$ = (service["securityTaskNotifications$"] = jest.fn(
+          () => new Subject(),
+        ));
+        const refreshTasks = jest.spyOn(service, "refreshTasks");
+
+        const subscription = service.listenForTaskNotifications();
+
+        mockMessages$.next({
+          command: "syncCompleted",
+          userId,
+          successfully: true,
+        });
+
+        await new Promise(process.nextTick);
+
+        expect(notificationHelper$).toHaveBeenCalled();
+        expect(refreshTasks).toHaveBeenCalledWith(userId);
+        subscription.unsubscribe();
+      });
+
+      it("should ignore non syncCompleted messages", async () => {
+        mockAuthStatuses$.next({
+          [userId]: AuthenticationStatus.Unlocked,
+        });
+        service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+
+        const notificationHelper$ = (service["securityTaskNotifications$"] = jest.fn(
+          () => new Subject(),
+        ));
+        const refreshTasks = jest.spyOn(service, "refreshTasks");
+
+        const subscription = service.listenForTaskNotifications();
+
+        mockMessages$.next({
+          command: "other-command",
+        });
+
+        await new Promise(process.nextTick);
+
+        expect(notificationHelper$).toHaveBeenCalled();
+        expect(refreshTasks).not.toHaveBeenCalledWith(userId);
+        subscription.unsubscribe();
+      });
+
+      it("should ignore failed sync messages", async () => {
+        mockAuthStatuses$.next({
+          [userId]: AuthenticationStatus.Unlocked,
+        });
+        service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+
+        const notificationHelper$ = (service["securityTaskNotifications$"] = jest.fn(
+          () => new Subject(),
+        ));
+        const refreshTasks = jest.spyOn(service, "refreshTasks");
+
+        const subscription = service.listenForTaskNotifications();
+
+        mockMessages$.next({
+          command: "syncCompleted",
+          userId,
+          successfully: false,
+        });
+
+        await new Promise(process.nextTick);
+
+        expect(notificationHelper$).toHaveBeenCalled();
+        expect(refreshTasks).not.toHaveBeenCalledWith(userId);
+        subscription.unsubscribe();
+      });
+
+      it("should ignore sync messages for other users", async () => {
+        mockAuthStatuses$.next({
+          [userId]: AuthenticationStatus.Unlocked,
+        });
+        service.tasksEnabled$ = jest.fn(() => new BehaviorSubject(true));
+
+        const notificationHelper$ = (service["securityTaskNotifications$"] = jest.fn(
+          () => new Subject(),
+        ));
+        const refreshTasks = jest.spyOn(service, "refreshTasks");
+
+        const subscription = service.listenForTaskNotifications();
+
+        mockMessages$.next({
+          command: "syncCompleted",
+          userId: "other-user-id" as UserId,
+          successfully: true,
+        });
+
+        await new Promise(process.nextTick);
+
+        expect(notificationHelper$).toHaveBeenCalled();
+        expect(refreshTasks).not.toHaveBeenCalledWith(userId);
+        subscription.unsubscribe();
+      });
+    });
+  });
 });

libs/common/src/vault/tasks/services/default-task.service.ts

@@ -1,10 +1,15 @@
-import { combineLatest, map, switchMap } from "rxjs";
+import { combineLatest, filter, map, merge, Observable, of, Subscription, switchMap } from "rxjs";
 
 import { ApiService } from "@bitwarden/common/abstractions/api.service";
 import { OrganizationService } from "@bitwarden/common/admin-console/abstractions/organization/organization.service.abstraction";
+import { AuthService } from "@bitwarden/common/auth/abstractions/auth.service";
+import { AuthenticationStatus } from "@bitwarden/common/auth/enums/authentication-status";
+import { NotificationType } from "@bitwarden/common/enums";
 import { FeatureFlag } from "@bitwarden/common/enums/feature-flag.enum";
 import { ListResponse } from "@bitwarden/common/models/response/list.response";
 import { ConfigService } from "@bitwarden/common/platform/abstractions/config/config.service";
+import { MessageListener } from "@bitwarden/common/platform/messaging";
+import { NotificationsService } from "@bitwarden/common/platform/notifications";
 import { StateProvider } from "@bitwarden/common/platform/state";
 import { SecurityTaskId, UserId } from "@bitwarden/common/types/guid";
 
@@ -14,12 +19,21 @@ import { SecurityTaskStatus } from "../enums";
 import { SecurityTask, SecurityTaskData, SecurityTaskResponse } from "../models";
 import { SECURITY_TASKS } from "../state/security-task.state";
 
+const getUnlockedUserIds = map<Record<UserId, AuthenticationStatus>, UserId[]>((authStatuses) =>
+  Object.entries(authStatuses ?? {})
+    .filter(([, status]) => status >= AuthenticationStatus.Unlocked)
+    .map(([userId]) => userId as UserId),
+);
+
 export class DefaultTaskService implements TaskService {
   constructor(
     private stateProvider: StateProvider,
     private apiService: ApiService,
     private organizationService: OrganizationService,
     private configService: ConfigService,
+    private authService: AuthService,
+    private notificationService: NotificationsService,
+    private messageListener: MessageListener,
   ) {}
 
   tasksEnabled$ = perUserCache$((userId) => {
@@ -36,6 +50,7 @@ export class DefaultTaskService implements TaskService {
       switchMap(async (tasks) => {
         if (tasks == null) {
           await this.fetchTasksFromApi(userId);
+          return null;
         }
         return tasks;
       }),
@@ -97,4 +112,66 @@ export class DefaultTaskService implements TaskService {
   ): Promise<SecurityTaskData[] | null> {
     return this.taskState(userId).update(() => tasks);
   }
+
+  /**
+   * Helper observable that filters the list of unlocked user IDs to only those with tasks enabled.
+   * @private
+   */
+  private getOnlyTaskEnabledUsers = switchMap<UserId[], Observable<UserId[]>>((unlockedUserIds) => {
+    if (unlockedUserIds.length === 0) {
+      return of([]);
+    }
+
+    return combineLatest(
+      unlockedUserIds.map((userId) =>
+        this.tasksEnabled$(userId).pipe(map((enabled) => (enabled ? userId : null))),
+      ),
+    ).pipe(map((userIds) => userIds.filter((userId) => userId !== null) as UserId[]));
+  });
+
+  /**
+   * Helper observable that emits whenever a security task notification is received for a user in the provided list.
+   * @private
+   */
+  private securityTaskNotifications$(filterByUserIds: UserId[]) {
+    return this.notificationService.notifications$.pipe(
+      filter(
+        ([notification, userId]) =>
+          notification.type === NotificationType.PendingSecurityTasks &&
+          filterByUserIds.includes(userId),
+      ),
+      map(([, userId]) => userId),
+    );
+  }
+
+  /**
+   * Helper observable that emits whenever a sync is completed for a user in the provided list.
+   */
+  private syncCompletedMessage$(filterByUserIds: UserId[]) {
+    return this.messageListener.allMessages$.pipe(
+      filter((msg) => msg.command === "syncCompleted" && !!msg.successfully && !!msg.userId),
+      map((msg) => msg.userId as UserId),
+      filter((userId) => filterByUserIds.includes(userId)),
+    );
+  }
+
+  /**
+   * Creates a subscription for pending security task notifications or completed syncs for unlocked users.
+   */
+  listenForTaskNotifications(): Subscription {
+    return this.authService.authStatuses$
+      .pipe(
+        getUnlockedUserIds,
+        this.getOnlyTaskEnabledUsers,
+        filter((allowedUserIds) => allowedUserIds.length > 0),
+        switchMap((allowedUserIds) =>
+          merge(
+            this.securityTaskNotifications$(allowedUserIds),
+            this.syncCompletedMessage$(allowedUserIds),
+          ),
+        ),
+        switchMap((userId) => this.refreshTasks(userId)),
+      )
+      .subscribe();
+  }
 }