feat(core): add notification list (#10480)

packages/frontend/core/src/modules/notification/index.ts

@@ -0,0 +1,26 @@
+export { NotificationCountService } from './services/count';
+export { NotificationListService } from './services/list';
+export type { Notification, NotificationBody } from './stores/notification';
+export { NotificationType } from './stores/notification';
+import type { Framework } from '@toeverything/infra';
+
+import { GraphQLService, ServerScope, ServerService } from '../cloud';
+import { GlobalSessionState } from '../storage';
+import { NotificationCountService } from './services/count';
+import { NotificationListService } from './services/list';
+import { NotificationStore } from './stores/notification';
+
+export function configureNotificationModule(framework: Framework) {
+  framework
+    .scope(ServerScope)
+    .service(NotificationCountService, [NotificationStore])
+    .service(NotificationListService, [
+      NotificationStore,
+      NotificationCountService,
+    ])
+    .store(NotificationStore, [
+      GraphQLService,
+      ServerService,
+      GlobalSessionState,
+    ]);
+}

packages/frontend/core/src/modules/notification/services/count.ts

@@ -0,0 +1,68 @@
+import {
+  catchErrorInto,
+  effect,
+  exhaustMapWithTrailing,
+  fromPromise,
+  LiveData,
+  onComplete,
+  OnEvent,
+  onStart,
+  Service,
+  smartRetry,
+} from '@toeverything/infra';
+import { EMPTY, mergeMap, switchMap, timer } from 'rxjs';
+
+import { ServerStarted } from '../../cloud/events/server-started';
+import { ApplicationFocused } from '../../lifecycle';
+import type { NotificationStore } from '../stores/notification';
+
+@OnEvent(ApplicationFocused, s => s.handleApplicationFocused)
+@OnEvent(ServerStarted, s => s.handleServerStarted)
+export class NotificationCountService extends Service {
+  constructor(private readonly store: NotificationStore) {
+    super();
+  }
+
+  readonly count$ = LiveData.from(this.store.watchNotificationCountCache(), 0);
+  readonly isLoading$ = new LiveData(false);
+  readonly error$ = new LiveData<any>(null);
+
+  revalidate = effect(
+    switchMap(() => {
+      return timer(0, 30000); // revalidate every 30 seconds
+    }),
+    exhaustMapWithTrailing(() => {
+      return fromPromise(signal =>
+        this.store.getNotificationCount(signal)
+      ).pipe(
+        mergeMap(result => {
+          this.setCount(result ?? 0);
+          return EMPTY;
+        }),
+        smartRetry(),
+        catchErrorInto(this.error$),
+        onStart(() => {
+          this.isLoading$.setValue(true);
+        }),
+        onComplete(() => this.isLoading$.setValue(false))
+      );
+    })
+  );
+
+  handleApplicationFocused() {
+    this.revalidate();
+  }
+
+  handleServerStarted() {
+    this.revalidate();
+  }
+
+  setCount(count: number) {
+    this.store.setNotificationCountCache(count);
+  }
+
+  override dispose(): void {
+    super.dispose();
+    this.revalidate.unsubscribe();
+  }
+}

packages/frontend/core/src/modules/notification/services/list.ts

@@ -0,0 +1,93 @@
+import {
+  catchErrorInto,
+  effect,
+  fromPromise,
+  LiveData,
+  onComplete,
+  onStart,
+  Service,
+  smartRetry,
+} from '@toeverything/infra';
+import { EMPTY, exhaustMap, mergeMap } from 'rxjs';
+
+import type { Notification, NotificationStore } from '../stores/notification';
+import type { NotificationCountService } from './count';
+
+export class NotificationListService extends Service {
+  isLoading$ = new LiveData(false);
+  notifications$ = new LiveData<Notification[]>([]);
+  nextCursor$ = new LiveData<string | undefined>(undefined);
+  hasMore$ = new LiveData(true);
+  error$ = new LiveData<any>(null);
+
+  readonly PAGE_SIZE = 8;
+
+  constructor(
+    private readonly store: NotificationStore,
+    private readonly notificationCount: NotificationCountService
+  ) {
+    super();
+  }
+
+  readonly loadMore = effect(
+    exhaustMap(() => {
+      if (!this.hasMore$.value) {
+        return EMPTY;
+      }
+      return fromPromise(signal =>
+        this.store.listNotification(
+          {
+            first: this.PAGE_SIZE,
+            after: this.nextCursor$.value,
+          },
+          signal
+        )
+      ).pipe(
+        mergeMap(result => {
+          if (!result) {
+            // If the user is not logged in, we just ignore the result.
+            return EMPTY;
+          }
+          const { edges, pageInfo, totalCount } = result;
+          this.notifications$.next([
+            ...this.notifications$.value,
+            ...edges.map(edge => edge.node),
+          ]);
+
+          // keep the notification count in sync
+          this.notificationCount.setCount(totalCount);
+
+          this.hasMore$.next(pageInfo.hasNextPage);
+          this.nextCursor$.next(pageInfo.endCursor ?? undefined);
+
+          return EMPTY;
+        }),
+        smartRetry(),
+        catchErrorInto(this.error$),
+        onStart(() => {
+          this.isLoading$.setValue(true);
+        }),
+        onComplete(() => this.isLoading$.setValue(false))
+      );
+    })
+  );
+
+  reset() {
+    this.notifications$.setValue([]);
+    this.hasMore$.setValue(true);
+    this.nextCursor$.setValue(undefined);
+    this.isLoading$.setValue(false);
+    this.error$.setValue(null);
+    this.loadMore.reset();
+  }
+
+  async readNotification(id: string) {
+    await this.store.readNotification(id);
+    this.notifications$.next(
+      this.notifications$.value.filter(notification => notification.id !== id)
+    );
+    this.notificationCount.setCount(
+      Math.max(this.notificationCount.count$.value - 1, 0)
+    );
+  }
+}

packages/frontend/core/src/modules/notification/stores/notification.ts

@@ -0,0 +1,85 @@
+import {
+  type ListNotificationsQuery,
+  listNotificationsQuery,
+  notificationCountQuery,
+  type PaginationInput,
+  readNotificationMutation,
+  type UnionNotificationBodyType,
+} from '@affine/graphql';
+import { Store } from '@toeverything/infra';
+import { map } from 'rxjs';
+
+import type { GraphQLService, ServerService } from '../../cloud';
+import type { GlobalSessionState } from '../../storage';
+
+export type Notification = NonNullable<
+  ListNotificationsQuery['currentUser']
+>['notifications']['edges'][number]['node'];
+
+export type NotificationBody = UnionNotificationBodyType;
+
+export { NotificationType } from '@affine/graphql';
+
+export class NotificationStore extends Store {
+  constructor(
+    private readonly gqlService: GraphQLService,
+    private readonly serverService: ServerService,
+    private readonly globalSessionState: GlobalSessionState
+  ) {
+    super();
+  }
+
+  watchNotificationCountCache() {
+    return this.globalSessionState
+      .watch('notification-count:' + this.serverService.server.id)
+      .pipe(
+        map(count => {
+          if (typeof count === 'number') {
+            return count;
+          }
+          return 0;
+        })
+      );
+  }
+
+  setNotificationCountCache(count: number) {
+    this.globalSessionState.set(
+      'notification-count:' + this.serverService.server.id,
+      count
+    );
+  }
+
+  async getNotificationCount(signal?: AbortSignal) {
+    const result = await this.gqlService.gql({
+      query: notificationCountQuery,
+      context: {
+        signal,
+      },
+    });
+
+    return result.currentUser?.notificationCount;
+  }
+
+  async listNotification(pagination: PaginationInput, signal?: AbortSignal) {
+    const result = await this.gqlService.gql({
+      query: listNotificationsQuery,
+      variables: {
+        pagination: pagination,
+      },
+      context: {
+        signal,
+      },
+    });
+
+    return result.currentUser?.notifications;
+  }
+
+  readNotification(id: string) {
+    return this.gqlService.gql({
+      query: readNotificationMutation,
+      variables: {
+        id,
+      },
+    });
+  }
+}