fix: prevent monthly schedule stream accumulation

CONTEXT.md

@@ -141,8 +141,23 @@ _Avoid_: Fallback, degraded, time sensitive
 The user-facing status for schedule delivery through an Alarm.
 _Avoid_: Notification, native alarm
 
+**Schedule**:
+A user commitment with a planned time and preparation context in OnTime.
+_Avoid_: Appointment, event, task
+
+**Monthly Calendar**:
+The calendar surface that shows Schedules grouped by day across a calendar month.
+_Avoid_: Month view, calendar grid
+
+**Calendar Month Range**:
+A contiguous span of calendar months whose Schedules are in scope for a Monthly Calendar.
+_Avoid_: Loaded range, stream range, cached range
+
 ## Relationships
 
+- A **Monthly Calendar** displays **Schedules** grouped by calendar day.
+- A **Calendar Month Range** starts at the first day of its first month and ends before the first day of the month after its last month.
+- A **Monthly Calendar** may extend a **Calendar Month Range** when the user moves to an adjacent month.
 - An **OnTime User** may own zero or more **Schedules**.
 - A **Schedule** has one **Place**.
 - A **Schedule** may use one **Preparation**.

lib/presentation/calendar/bloc/monthly_schedules_bloc.dart

@@ -4,7 +4,6 @@ import 'package:equatable/equatable.dart';
 import 'package:flutter_bloc/flutter_bloc.dart';
 import 'package:injectable/injectable.dart';
 import 'package:on_time_front/core/dio/api_error_message.dart';
-import 'package:on_time_front/core/logging/app_logger.dart';
 import 'package:on_time_front/domain/entities/preparation_entity.dart';
 import 'package:on_time_front/domain/entities/schedule_entity.dart';
 import 'package:on_time_front/domain/use-cases/delete_schedule_use_case.dart';
@@ -39,6 +38,8 @@ class MonthlySchedulesBloc
       _onPreparationsPrefetchRequested,
     );
     on<MonthlySchedulesPreparationsStreamChanged>(_onPreparationsStreamChanged);
+    on<_MonthlySchedulesScheduleStreamChanged>(_onScheduleStreamChanged);
+    on<_MonthlySchedulesScheduleStreamFailed>(_onScheduleStreamFailed);
 
     _preparationSubscription = _streamPreparationsUseCase().listen((
       preparations,
@@ -56,101 +57,82 @@ class MonthlySchedulesBloc
   final GetPreparationByScheduleIdUseCase _getPreparationByScheduleIdUseCase;
   final StreamPreparationsUseCase _streamPreparationsUseCase;
 
+  StreamSubscription<List<ScheduleEntity>>? _scheduleSubscription;
   StreamSubscription<Map<String, PreparationEntity>>? _preparationSubscription;
+  int _scheduleRangeRequestId = 0;
 
   Future<void> _onSubscriptionRequested(
     MonthlySchedulesSubscriptionRequested event,
     Emitter<MonthlySchedulesState> emit,
   ) async {
-    emit(state.copyWith(status: () => MonthlySchedulesStatus.loading));
-
-    try {
-      await _loadSchedulesForMonthUseCase(event.date);
-    } catch (_) {
-      emit(state.copyWith(status: () => MonthlySchedulesStatus.error));
-      return;
-    }
-
-    await emit.forEach(
-      _getSchedulesByDateUseCase(event.startDate, event.endDate),
-      onData: (schedules) {
-        final groupedSchedules = _groupSchedulesByDate(schedules);
-        final nextState = state.copyWith(
-          status: () => MonthlySchedulesStatus.success,
-          schedules: () => groupedSchedules,
-          startDate: () => event.startDate,
-          endDate: () => event.endDate,
-        );
-        _requestVisibleDatePreparationPrefetch(nextState);
-        return nextState;
-      },
-      onError: (error, stackTrace) =>
-          state.copyWith(status: () => MonthlySchedulesStatus.error),
+    await _loadAndWatchCalendarMonthRange(
+      loadDate: event.date,
+      startDate: event.startDate,
+      endDate: event.endDate,
+      emit: emit,
     );
   }
 
   Future<void> _onMonthAdded(
     MonthlySchedulesMonthAdded event,
     Emitter<MonthlySchedulesState> emit,
   ) async {
-    late DateTime startDate;
-    late DateTime endDate;
-    if (!(state.startDate!.isAfter(event.startDate) ||
-        state.endDate!.isBefore(event.endDate))) {
-      // If the month is already loaded, we don't need to load the schedules again.
-      startDate = state.startDate!;
-      endDate = state.endDate!;
-    } else if (event.date.month !=
-            state.startDate!.subtract(Duration(days: 1)).month &&
-        (event.date.month != state.endDate!.month)) {
-      // If the month is not consecutive, we need to load the schedules for the
-      add(MonthlySchedulesSubscriptionRequested(date: event.date));
+    final currentStartDate = state.startDate;
+    final currentEndDate = state.endDate;
+    if (currentStartDate == null || currentEndDate == null) {
+      await _loadAndWatchCalendarMonthRange(
+        loadDate: event.date,
+        startDate: event.startDate,
+        endDate: event.endDate,
+        emit: emit,
+      );
       return;
-    } else {
-      // If the month is not consecutive, we need to load the schedules for the
-      // month and update the state with the new schedules.
+    }
 
-      startDate = event.startDate.isBefore(state.startDate!)
-          ? event.startDate
-          : state.startDate!;
-      endDate = event.endDate.isAfter(state.endDate!)
-          ? event.endDate
-          : state.endDate!;
+    final eventIsInsideCurrentRange =
+        !currentStartDate.isAfter(event.startDate) &&
+        !currentEndDate.isBefore(event.endDate);
+    if (eventIsInsideCurrentRange) {
+      return;
+    }
 
-      emit(
-        state.copyWith(
-          status: () => MonthlySchedulesStatus.loading,
-          schedules: () => state.schedules,
-          startDate: () => startDate,
-          endDate: () => endDate,
-        ),
+    final eventMonth = DateTime(event.date.year, event.date.month, 1);
+    final previousAdjacentMonth = DateTime(
+      currentStartDate.year,
+      currentStartDate.month - 1,
+      1,
+    );
+    final nextAdjacentMonth = DateTime(
+      currentEndDate.year,
+      currentEndDate.month,
+      1,
+    );
+    final eventIsAdjacent =
+        eventMonth == previousAdjacentMonth || eventMonth == nextAdjacentMonth;
+
+    if (!eventIsAdjacent) {
+      await _loadAndWatchCalendarMonthRange(
+        loadDate: event.date,
+        startDate: event.startDate,
+        endDate: event.endDate,
+        emit: emit,
       );
-
-      try {
-        await _loadSchedulesForMonthUseCase(event.date);
-      } catch (_) {
-        emit(state.copyWith(status: () => MonthlySchedulesStatus.error));
-        return;
-      }
+      return;
     }
 
-    AppLogger.debug('startDate: $startDate, endDate: $endDate');
-    await emit.forEach(
-      _getSchedulesByDateUseCase(startDate, endDate),
-      onData: (schedules) {
-        final groupedSchedules = _groupSchedulesByDate(schedules);
-        final nextState = state.copyWith(
-          status: () => MonthlySchedulesStatus.success,
-          schedules: () => groupedSchedules,
-          startDate: () => startDate,
-          endDate: () => endDate,
-        );
-        _requestVisibleDatePreparationPrefetch(nextState);
-        return nextState;
-      },
-      onError: (error, stackTrace) {
-        return state.copyWith(status: () => MonthlySchedulesStatus.error);
-      },
+    final startDate = event.startDate.isBefore(currentStartDate)
+        ? event.startDate
+        : currentStartDate;
+    final endDate = event.endDate.isAfter(currentEndDate)
+        ? event.endDate
+        : currentEndDate;
+
+    await _loadAndWatchCalendarMonthRange(
+      loadDate: event.date,
+      startDate: startDate,
+      endDate: endDate,
+      emit: emit,
+      exposeLoadingRange: true,
     );
   }
 
@@ -328,6 +310,97 @@ class MonthlySchedulesBloc
     emit(state.copyWith(preparationDurationByScheduleId: () => nextDurations));
   }
 
+  void _onScheduleStreamChanged(
+    _MonthlySchedulesScheduleStreamChanged event,
+    Emitter<MonthlySchedulesState> emit,
+  ) {
+    if (event.requestId != _scheduleRangeRequestId) {
+      return;
+    }
+
+    final groupedSchedules = _groupSchedulesByDate(event.schedules);
+    final nextState = state.copyWith(
+      status: () => MonthlySchedulesStatus.success,
+      schedules: () => groupedSchedules,
+      startDate: () => event.startDate,
+      endDate: () => event.endDate,
+    );
+    emit(nextState);
+    _requestVisibleDatePreparationPrefetch(nextState);
+  }
+
+  void _onScheduleStreamFailed(
+    _MonthlySchedulesScheduleStreamFailed event,
+    Emitter<MonthlySchedulesState> emit,
+  ) {
+    if (event.requestId != _scheduleRangeRequestId) {
+      return;
+    }
+
+    emit(state.copyWith(status: () => MonthlySchedulesStatus.error));
+  }
+
+  Future<void> _loadAndWatchCalendarMonthRange({
+    required DateTime loadDate,
+    required DateTime startDate,
+    required DateTime endDate,
+    required Emitter<MonthlySchedulesState> emit,
+    bool exposeLoadingRange = false,
+  }) async {
+    final requestId = ++_scheduleRangeRequestId;
+    await _cancelScheduleSubscription();
+
+    var loadingState = state.copyWith(
+      status: () => MonthlySchedulesStatus.loading,
+    );
+    if (exposeLoadingRange) {
+      loadingState = loadingState.copyWith(
+        startDate: () => startDate,
+        endDate: () => endDate,
+      );
+    }
+    emit(loadingState);
+
+    try {
+      await _loadSchedulesForMonthUseCase(loadDate);
+    } catch (_) {
+      if (requestId == _scheduleRangeRequestId) {
+        emit(state.copyWith(status: () => MonthlySchedulesStatus.error));
+      }
+      return;
+    }
+
+    if (requestId != _scheduleRangeRequestId) {
+      return;
+    }
+
+    _scheduleSubscription = _getSchedulesByDateUseCase(startDate, endDate)
+        .listen(
+          (schedules) {
+            if (!isClosed && requestId == _scheduleRangeRequestId) {
+              add(
+                _MonthlySchedulesScheduleStreamChanged(
+                  requestId: requestId,
+                  startDate: startDate,
+                  endDate: endDate,
+                  schedules: schedules,
+                ),
+              );
+            }
+          },
+          onError: (Object error, StackTrace stackTrace) {
+            if (!isClosed && requestId == _scheduleRangeRequestId) {
+              add(_MonthlySchedulesScheduleStreamFailed(requestId: requestId));
+            }
+          },
+        );
+  }
+
+  Future<void> _cancelScheduleSubscription() async {
+    await _scheduleSubscription?.cancel();
+    _scheduleSubscription = null;
+  }
+
   void _requestVisibleDatePreparationPrefetch(
     MonthlySchedulesState sourceState,
   ) {
@@ -393,6 +466,7 @@ class MonthlySchedulesBloc
 
   @override
   Future<void> close() async {
+    await _cancelScheduleSubscription();
     await _preparationSubscription?.cancel();
     return super.close();
   }

lib/presentation/calendar/bloc/monthly_schedules_event.dart

@@ -75,10 +75,36 @@ final class MonthlySchedulesPreparationsStreamChanged
     extends MonthlySchedulesEvent {
   final Map<String, PreparationEntity> preparations;
 
-  const MonthlySchedulesPreparationsStreamChanged({
-    required this.preparations,
-  });
+  const MonthlySchedulesPreparationsStreamChanged({required this.preparations});
 
   @override
   List<Object> get props => [preparations];
 }
+
+final class _MonthlySchedulesScheduleStreamChanged
+    extends MonthlySchedulesEvent {
+  const _MonthlySchedulesScheduleStreamChanged({
+    required this.requestId,
+    required this.startDate,
+    required this.endDate,
+    required this.schedules,
+  });
+
+  final int requestId;
+  final DateTime startDate;
+  final DateTime endDate;
+  final List<ScheduleEntity> schedules;
+
+  @override
+  List<Object> get props => [requestId, startDate, endDate, schedules];
+}
+
+final class _MonthlySchedulesScheduleStreamFailed
+    extends MonthlySchedulesEvent {
+  const _MonthlySchedulesScheduleStreamFailed({required this.requestId});
+
+  final int requestId;
+
+  @override
+  List<Object> get props => [requestId];
+}