packages/pass/store/sagas/events/channel.worker.ts

   1 import type { Action } from 'redux';
   2 import type { Task } from 'redux-saga';
   3 import { call, cancel, cancelled, fork, put, select, take, takeLeading } from 'redux-saga/effects';
   4
   5 import { ACTIVE_POLLING_TIMEOUT } from '@proton/pass/lib/events/constants';
   6 import type { EventManagerEvent } from '@proton/pass/lib/events/manager';
   7 import { channelAcknowledge, wakeupSuccess } from '@proton/pass/store/actions';
   8 import { channelRequest } from '@proton/pass/store/actions/requests';
   9 import type { RequestEntry } from '@proton/pass/store/request/types';
  10 import { selectRequest } from '@proton/pass/store/selectors';
  11 import type { RootSagaOptions } from '@proton/pass/store/types';
  12 import type { Maybe } from '@proton/pass/types';
  13 import { logger } from '@proton/pass/utils/logger';
  14 import { epochToMs, msToEpoch } from '@proton/pass/utils/time/epoch';
  15 import { wait } from '@proton/shared/lib/helpers/promise';
  16 import noop from '@proton/utils/noop';
  17
  18 import type { EventChannel } from './types';
  19
  20 /* generic worker over an EventChannel : responsible for polling
  21  * the underlying redux-saga event channel and triggering the
  22  * appropriate callback generators. Closes the channel if the
  23  * parent task is canceled */
  24 export function* channelEventsWorker<T extends {}>(eventChannel: EventChannel<T>, options: RootSagaOptions): Generator {
  25     const { channel, onEvent, onError, manager } = eventChannel;
  26     try {
  27         while (true) {
  28             try {
  29                 manager.setInterval(options.getPollingInterval());
  30                 const event = (yield take(channel)) as EventManagerEvent<T>;
  31                 yield call(onEvent, event, eventChannel, options);
  32                 yield put(channelAcknowledge(channelRequest(eventChannel.channelId)));
  33             } catch (error: unknown) {
  34                 logger.warn(`[Saga::Events] received an event error`, error);
  35                 if (onError) yield call(onError, error, eventChannel, options);
  36             }
  37         }
  38     } finally {
  39         if (yield cancelled()) channel.close();
  40     }
  41 }
  42
  43 /* This worker will call the event manager immediately and
  44  * on every wakeupSuccess action coming  from the pop-up in
  45  * in order to sync as quickly as possible. Take the leading
  46  * wakeup call in order to avoid unnecessary parallel calls */
  47 export function* channelInitWorker<T extends {}>(
  48     { manager, channelId }: EventChannel<T>,
  49     options: RootSagaOptions
  50 ): Generator {
  51     const init = (yield fork(function* () {
  52         const request: Maybe<RequestEntry<'success'>> = yield select(selectRequest(channelRequest(channelId)));
  53         const interval = msToEpoch(options.getPollingInterval());
  54         const delay: number = options.getPollingDelay?.(interval, request?.requestedAt) ?? 0;
  55         yield wait(epochToMs(delay));
  56         yield manager.call().catch(noop);
  57     })) as Task;
  58
  59     yield takeLeading(
  60         (action: Action) => wakeupSuccess.match(action) && action.meta.receiver.endpoint === 'popup',
  61         function* () {
  62             yield cancel(init);
  63             yield manager.call().catch(noop);
  64             /* wait the channel's interval to process
  65              * the next wakeupSuccess in case user is
  66              * repeatedly opening the pop-up */
  67             yield wait(ACTIVE_POLLING_TIMEOUT);
  68         }
  69     );
  70 }