relatica/lib/riverpod_controllers/notification_services.dart

import 'package:logging/logging.dart';
import 'package:relatica/models/exec_error.dart';
import 'package:relatica/models/user_notification.dart';
import 'package:relatica/riverpod_controllers/direct_message_services.dart';
import 'package:relatica/riverpod_controllers/settings_services.dart';
import 'package:relatica/serializers/mastodon/follow_request_mastodon_extensions.dart';
import 'package:result_monad/result_monad.dart';
import 'package:riverpod_annotation/riverpod_annotation.dart';

import '../friendica_client/friendica_client.dart';
import '../friendica_client/paged_response.dart';
import '../friendica_client/pages_manager.dart';
import '../friendica_client/paging_data.dart';
import '../globals.dart';
import '../models/auth/profile.dart';
import '../services/feature_version_checker.dart';
import '../services/follow_requests_manager.dart';
import '../services/network_status_service.dart';
import '../utils/active_profile_selector.dart';

part 'notification_services.g.dart';

const _itemsPerQuery = 50;
const _minimumDmsAndCrsUpdateDuration = Duration(seconds: 30);
final _logger = Logger('NotificationManager');

@Riverpod(keepAlive: true)
class NotificationsManager extends _$NotificationsManager {
  late Profile userProfile;
  final dms = <UserNotification>[];
  final connectionRequests = <UserNotification>[];
  final unread = <UserNotification>[];
  final read = <UserNotification>[];
  var lastDmsUpdate = DateTime(1900);
  var lastCrUpdate = DateTime(1900);
  var initialized = false;

  bool get hasNotifications =>
      dms.isNotEmpty ||
      connectionRequests.isNotEmpty ||
      unread.isNotEmpty ||
      read.isNotEmpty;

  @override
  Future<Result<List<UserNotification>, ExecError>> build(
      Profile profile) async {
    _logger.fine('Building');
    userProfile = profile;
    if (!initialized) {
      _logger.fine('Initializing');
      await loadUnreadNotifications(true);
      initialized = true;
      _logger.fine('Initialized');
    }

    return Result.ok([...connectionRequests, ...dms, ...unread, ...read]);
  }

  void refreshNotifications() async {
    clear(withListenerNotification: true);
  }

  void clear({bool withListenerNotification = true}) {
    dms.clear();
    connectionRequests.clear();
    unread.clear();
    read.clear();
    initialized = false;
    if (withListenerNotification) {
      ref.invalidateSelf();
    }
  }

  FutureResult<List<UserNotification>, ExecError> loadUnreadNotifications(
      bool withListenerNotification) async {
    final notificationsFromRefresh = <UserNotification>[];

    final pm = _buildPageManager(profile, false);
    final useActualRequests = getIt<FriendicaVersionChecker>()
        .canUseFeature(RelaticaFeatures.usingActualFollowRequests);
    var hasMore = true;
    var first = true;
    const maxCalls = 3;
    var count = 0;
    while (hasMore && count < maxCalls) {
      final result =
          first ? await pm.initialize(_itemsPerQuery) : await pm.nextFromEnd();

      first = false;
      result.match(
          onSuccess: (nd) =>
              _logger.fine('Got ${nd.data.length} notifications'),
          onError: (e) => _logger.severe('Error getting notification: $e'));
      final response = result.getValueOrElse(() => PagedResponse([]));
      response.data
          .where((n) =>
              !useActualRequests || n.type != NotificationType.follow_request)
          .forEach(notificationsFromRefresh.add);
      hasMore = response.next != null;
      count++;
    }

    // filter out connection requests if going to use the real service for that when doing the query
    // get earliest and latest notification ID from unread notifications

    // query all notifications over that in page increments of 25
    // query unread notifications in increments of 25 after the latest ID
    return await _postFetchOperations(
      notificationsFromRefresh,
      withListenerNotification,
    );
  }

  FutureResult<List<UserNotification>, ExecError> _postFetchOperations(
    List<UserNotification> notificationsFromRefresh,
    bool withListenerNotification,
  ) async {
    getIt<NetworkStatusService>().startNotificationUpdate();
    if (DateTime.now().difference(lastDmsUpdate) >
        _minimumDmsAndCrsUpdateDuration) {
      await ref
          .read(directMessageThreadIdsProvider(userProfile).notifier)
          .update();
      lastDmsUpdate = DateTime.now();
    }

    final useActualRequests = getIt<FriendicaVersionChecker>()
        .canUseFeature(RelaticaFeatures.usingActualFollowRequests);

    if (useActualRequests) {
      if (DateTime.now().difference(lastCrUpdate) >
          _minimumDmsAndCrsUpdateDuration) {
        await getIt<ActiveProfileSelector<FollowRequestsManager>>()
            .getForProfile(profile)
            .transformAsync((fm) async => fm.update());
        lastCrUpdate = DateTime.now();
      }
    }

    final notifications = <String, UserNotification>{};

    notificationsFromRefresh.removeWhere((n) =>
        n.type == NotificationType.direct_message ||
        (useActualRequests && n.type == NotificationType.follow_request));
    for (final n in notificationsFromRefresh) {
      notifications[n.id] = n;
    }

    getIt<NetworkStatusService>().finishNotificationUpdate();
    for (final n in buildUnreadMessageNotifications(useActualRequests)) {
      notifications[n.id] = n;
    }

    _processNewNotifications(notifications.values);

    if (withListenerNotification) {
      ref.invalidateSelf();
    }
    return Result.ok(notifications.values.toList());
  }

  FutureResult<List<UserNotification>, ExecError> loadNewerNotifications({
    bool withListenerNotification = true,
  }) async {
    final (_, highestId) =
        unread.isNotEmpty ? calcLowHigh(unread) : calcLowHigh(read);
    final pm = _buildPageManager(
      profile,
      true,
      initialPages: read.isEmpty && unread.isEmpty
          ? []
          : [
              PagedResponse(
                <String>[],
                previous: PagingData(
                  minId: highestId,
                  limit: _itemsPerQuery,
                ),
              )
            ],
    );

    final result = await (unread.isEmpty && read.isEmpty
            ? pm.initialize(_itemsPerQuery)
            : pm.previousFromBeginning())
        .andThenAsync(
          (page) async =>
              await _postFetchOperations(page.data, withListenerNotification),
        )
        .withError(
            (error) => _logger.info('Error getting more updates: $error'));
    return result.execErrorCast();
  }

  FutureResult<List<UserNotification>, ExecError> loadOlderNotifications(
      {bool withListenerNotification = true}) async {
    if (unread.isNotEmpty) {
      final result =
          await _loadOlderUnreadNotifications(withListenerNotification);
      final nonDmAndConnectionNotifications = result
          .getValueOrElse(() => [])
          .where((n) =>
              n.type != NotificationType.follow_request &&
              n.type != NotificationType.direct_message)
          .toList();
      if (nonDmAndConnectionNotifications.isNotEmpty) {
        return result;
      }
    }

    return _loadOlderReadAndUnreadNotifications(withListenerNotification);
  }

  FutureResult<bool, ExecError> markSeen(UserNotification notification) async {
    final result = await NotificationsClient(profile)
        .clearNotification(notification)
        .withResult((_) {
      unread.remove(notification);
      read.add(notification.copy(dismissed: true));
      read.sort();
      ref.invalidateSelf();
    });

    return result.execErrorCast();
  }

  FutureResult<bool, ExecError> markAllAsRead() async {
    final result = await NotificationsClient(userProfile)
        .clearNotifications()
        .withResult((_) {
      unread.map((n) => n.copy(dismissed: true)).forEach(read.add);
      unread.clear();
      read.sort();
      ref.invalidateSelf();
    });

    return result.execErrorCast();
  }

  List<UserNotification> buildUnreadMessageNotifications(
      bool useActualRequests) {
    final myId = userProfile.userId;
    final dmsResult = ref
        .watch(directMessageThreadIdsProvider(userProfile))
        .map((id) =>
            ref.watch(directMessageThreadServiceProvider(userProfile, id)))
        .where((t) => !t.allSeen)
        .map((t) {
      final fromAccount = t.participants.firstWhere((p) => p.id != myId);
      final latestMessage =
          t.messages.reduce((s, m) => s.createdAt > m.createdAt ? s : m);
      return UserNotification(
          id: (fromAccount.hashCode ^ t.parentUri.hashCode ^ t.title.hashCode)
              .toString(),
          type: NotificationType.direct_message,
          fromId: fromAccount.id,
          fromName: fromAccount.name,
          fromUrl: fromAccount.profileUrl,
          timestamp: latestMessage.createdAt,
          iid: t.parentUri,
          dismissed: false,
          content: '${fromAccount.name} sent you a direct message',
          link: '');
    });

    final followRequestResult = !useActualRequests
        ? <UserNotification>[]
        : getIt<ActiveProfileSelector<FollowRequestsManager>>()
            .getForProfile(profile)
            .transform(
                (fm) => fm.requests.map((r) => r.toUserNotification()).toList())
            .getValueOrElse(() => []);

    return [...dmsResult, ...followRequestResult];
  }

  void updateNotification(UserNotification notification) {}

  Future<void> _processNewNotifications(
      Iterable<UserNotification> notifications) async {
    final groupNotifications = ref.watch(notificationGroupingSettingProvider);
    final dmsMap = <String, UserNotification>{};
    final crMap = <String, UserNotification>{};
    final unreadMap = <String, UserNotification>{};
    final readMap = <String, UserNotification>{};

    final st = Stopwatch()..start();

    for (int i = 0; i < dms.length; i++) {
      dmsMap[dms[i].id] = dms[i];
    }

    if (st.elapsedMilliseconds > maxProcessingMillis) {
      await Future.delayed(processingSleep, () => st.reset());
    }

    for (int i = 0; i < connectionRequests.length; i++) {
      crMap[connectionRequests[i].id] = connectionRequests[i];
    }

    if (st.elapsedMilliseconds > maxProcessingMillis) {
      await Future.delayed(processingSleep, () => st.reset());
    }

    for (int i = 0; i < unread.length; i++) {
      unreadMap[unread[i].id] = unread[i];
    }

    if (st.elapsedMilliseconds > maxProcessingMillis) {
      await Future.delayed(processingSleep, () => st.reset());
    }

    for (int i = 0; i < read.length; i++) {
      readMap[read[i].id] = read[i];
    }

    dms.clear();
    connectionRequests.clear();
    unread.clear();
    read.clear();

    for (final n in notifications) {
      if (st.elapsedMilliseconds > maxProcessingMillis) {
        await Future.delayed(processingSleep, () => st.reset());
      }
      dmsMap.remove(n.id);
      crMap.remove(n.id);
      unreadMap.remove(n.id);
      readMap.remove(n.id);
      if (n.dismissed) {
        readMap[n.id] = n;
        continue;
      }

      switch (n.type) {
        case NotificationType.direct_message:
          dmsMap[n.id] = n;
          break;
        case NotificationType.follow_request:
          crMap[n.id] = n;
          break;
        default:
          unreadMap[n.id] = n;
      }
    }

    dms
      ..addAll(dmsMap.values)
      ..sort();
    connectionRequests
      ..addAll(crMap.values)
      ..sort();
    unread
      ..addAll(unreadMap.values)
      ..sort(
          (n1, n2) => _compareByTypeStatusAndDate(n1, n2, groupNotifications));
    read
      ..addAll(readMap.values)
      ..sort(
          (n1, n2) => _compareByTypeStatusAndDate(n1, n2, groupNotifications));
  }

  FutureResult<List<UserNotification>, ExecError> _loadOlderUnreadNotifications(
      bool withListenerNotification) async {
    final (lowestId, _) = calcLowHigh(unread);
    final pm = _buildPageManager(
      profile,
      false,
      initialPages: [
        PagedResponse(
          <String>[],
          next: PagingData(
            maxId: lowestId,
            limit: _itemsPerQuery,
          ),
        )
      ],
    );

    final result = await pm
        .nextFromEnd()
        .andThenAsync(
          (page) async =>
              await _postFetchOperations(page.data, withListenerNotification),
        )
        .withError(
            (error) => _logger.info('Error getting more updates: $error'));

    return result.execErrorCast();
  }

  FutureResult<List<UserNotification>, ExecError>
      _loadOlderReadAndUnreadNotifications(
          bool withListenerNotification) async {
    final (lowestId, _) =
        read.isNotEmpty ? calcLowHigh(read) : calcLowHigh(unread);
    final pm = _buildPageManager(
      profile,
      true,
      initialPages: read.isEmpty && unread.isEmpty
          ? []
          : [
              PagedResponse(
                <String>[],
                next: PagingData(
                  maxId: lowestId,
                  limit: _itemsPerQuery,
                ),
              )
            ],
    );
    final result = await (read.isEmpty && unread.isEmpty
            ? pm.initialize(_itemsPerQuery)
            : pm.nextFromEnd())
        .andThenAsync(
          (page) async =>
              await _postFetchOperations(page.data, withListenerNotification),
        )
        .withError(
            (error) => _logger.info('Error getting more updates: $error'));
    return result.execErrorCast();
  }
}

(int lowest, int highest) calcLowHigh(List<UserNotification> notifications) {
  int highestNotificationId = -1;
  int lowestNotificationId = 0x7FFFFFFFFFFFFFFF;
  final ids = notifications
      .where((n) =>
          n.type != NotificationType.direct_message &&
          n.type != NotificationType.follow_request)
      .map((n) => int.parse(n.id));

  for (var id in ids) {
    if (id > highestNotificationId) {
      highestNotificationId = id;
    }

    if (id < lowestNotificationId) {
      lowestNotificationId = id;
    }
  }

  return (lowestNotificationId, highestNotificationId);
}

PagesManager<List<UserNotification>, String> _buildPageManager(
        Profile profile, bool includeAll,
        {List<PagedResponse> initialPages = const []}) =>
    PagesManager<List<UserNotification>, String>(
      initialPages: initialPages,
      idMapper: (nn) => nn.map((n) => n.id).toList(),
      onRequest: (pd) async =>
          await NotificationsClient(profile).getNotifications(pd, includeAll),
    );

int _compareByTypeStatusAndDate(
    UserNotification n1, UserNotification n2, bool groupNotifications) {
  final n1Weight = _notificationTypeToWeight(n1.type);
  final n2Weight = _notificationTypeToWeight(n2.type);
  if (!groupNotifications || n1Weight == n2Weight) {
    return n1.compareTo(n2);
  }

  return (n2Weight - n1Weight).sign.toInt();
}

num _notificationTypeToWeight(NotificationType type) {
  return switch (type) {
    NotificationType.follow_request => 1000,
    NotificationType.follow => 100,
    NotificationType.direct_message => 50,
    NotificationType.mention => 10,
    NotificationType.status => 4,
    NotificationType.reshare => 3,
    NotificationType.reblog => 3,
    NotificationType.favourite => 2,
    NotificationType.unknown => 1,
  };
}