https://site.fr/_static/8c8af5c529149f32/scripts/domain-info-stream.js

1import { applySidebarProductEvent, createEmptyDomainInfo, resolveSidebarDomainName } from './sidebar-products.js';
2import { CURRENT_USER_ENDPOINT, isCurrentUserGuestError } from './current-user-api.js';
3
4const globalScope = typeof window !== 'undefined' ? window : globalThis;
5
6const RECONNECT_DELAY_IN_MS = 5000;
7const SIDEBAR_STREAM_MAX_RECONNECT_ATTEMPTS = 3;
8const SIDEBAR_STREAM_LIFETIME_IN_MS = 10 * 60 * 1000;
9const LIFETIME_EPSILON_IN_MS = 2 * 60 * 1000;
10
11const noop = () => {};
12
13const cloneValue = (value) => {
14  if (typeof structuredClone === 'function') {
15    try {
16      return structuredClone(value);
17    } catch {}
18  }
19
20  return JSON.parse(JSON.stringify(value));
21};
22
23const createInitialDomainInfoStreamState = () => ({
24  status: 'idle',
25  retryCount: 0,
26  maxRetryCount: SIDEBAR_STREAM_MAX_RECONNECT_ATTEMPTS,
27  needsManualRefresh: false,
28  isSessionExpired: false,
29  stopReason: null,
30  lastDisconnectReason: null,
31});
32
33const state = {
34  streamId: 0,
35  userId: null,
36  data: createEmptyDomainInfo(),
37  isLoading: false,
38  isLoaded: false,
39  stream: createInitialDomainInfoStreamState(),
40  subscribers: new Set(),
41  connection: null,
42};
43
44const getApiFetcher = () => {
45  const apiFetcher = globalScope.fetcher;
46  if (!apiFetcher || typeof apiFetcher.getJson !== 'function') {
47    return null;
48  }
49  return apiFetcher;
50};
51
52const isCurrentStream = (streamId) => state.streamId === streamId;
53
54const countDomainInfoItems = (data) =>
55  Object.values(data || {}).reduce((total, items) => total + (Array.isArray(items) ? items.length : 0), 0);
56
57const getUserId = (user) => {
58  const userId = Number(user?.user_id);
59  if (!Number.isFinite(userId) || userId <= 0) {
60    return null;
61  }
62  return userId;
63};
64
65const getStateSnapshot = () => ({
66  userId: state.userId,
67  data: cloneValue(state.data),
68  isLoading: state.isLoading,
69  isLoaded: state.isLoaded,
70  stream: { ...state.stream },
71});
72
73const notifySubscribers = () => {
74  const snapshot = getStateSnapshot();
75  state.subscribers.forEach((listener) => {
76    try {
77      listener(snapshot);
78    } catch {}
79  });
80};
81
82const setStreamState = (patch) => {
83  state.stream = {
84    ...state.stream,
85    ...(patch || {}),
86  };
87  notifySubscribers();
88};
89
90const setDomainInfoData = (nextData, { isLoading = state.isLoading, isLoaded = state.isLoaded } = {}) => {
91  state.data = nextData || createEmptyDomainInfo();
92  state.isLoading = Boolean(isLoading);
93  state.isLoaded = Boolean(isLoaded);
94  notifySubscribers();
95};
96
97const mutateDomainInfoData = (mutator, options = {}) => {
98  if (typeof mutator !== 'function') {
99    return getStateSnapshot();
100  }
101
102  const draft = cloneValue(state.data);
103  const nextDataCandidate = mutator(draft);
104  const nextData =
105    nextDataCandidate && typeof nextDataCandidate === 'object' && !Array.isArray(nextDataCandidate)
106      ? nextDataCandidate
107      : draft;
108
109  setDomainInfoData(nextData, {
110    isLoading: Object.prototype.hasOwnProperty.call(options, 'isLoading') ? options.isLoading : state.isLoading,
111    isLoaded: Object.prototype.hasOwnProperty.call(options, 'isLoaded') ? options.isLoaded : state.isLoaded,
112  });
113
114  return getStateSnapshot();
115};
116
117const getEventData = (messageBlock) => {
118  const data = [];
119
120  String(messageBlock || '')
121    .split('\n')
122    .forEach((line) => {
123      if (!line || line.startsWith(':')) {
124        return;
125      }
126
127      const separatorIndex = line.indexOf(':');
128      const field = separatorIndex === -1 ? line : line.slice(0, separatorIndex);
129      const value = separatorIndex === -1 ? '' : line.slice(separatorIndex + 1).trimStart();
130
131      if (field === 'data') {
132        data.push(value);
133      }
134    });
135
136  if (!data.length) {
137    return null;
138  }
139
140  return data.join('\n');
141};
142
143const normalizeChunk = (chunk) =>
144  String(chunk || '')
145    .replace(/\r\n/g, '\n')
146    .replace(/\r/g, '\n');
147
148const parseEventPayload = (eventData) => {
149  try {
150    const parsedData = JSON.parse(eventData);
151
152    if (typeof parsedData === 'string') {
153      return parseEventPayload(parsedData);
154    }
155
156    if (typeof parsedData?.data === 'string') {
157      try {
158        return JSON.parse(parsedData.data);
159      } catch {
160        return parsedData;
161      }
162    }
163
164    return parsedData;
165  } catch {
166    return null;
167  }
168};
169
170const checkUserSession = async () => {
171  const apiFetcher = getApiFetcher();
172  if (!apiFetcher) {
173    return 'unknown';
174  }
175
176  try {
177    await apiFetcher.getJson(CURRENT_USER_ENDPOINT, {
178      cacheMode: 'no-store',
179      credentials: 'include',
180    });
181    return 'active';
182  } catch (error) {
183    if (isCurrentUserGuestError(error)) {
184      return 'expired';
185    }
186
187    return 'unknown';
188  }
189};
190
191const createDomainInfoStreamConnection = (userId, streamId) => {
192  const apiFetcher = getApiFetcher();
193  if (!apiFetcher || !apiFetcher.baseUrl) {
194    return {
195      close: noop,
196    };
197  }
198
199  const sidebarStreamUrl = new URL(`/v2/sidebar/new_stream/${userId}`, apiFetcher.baseUrl).toString();
200  const startedAt = Date.now();
201
202  let reloading = false;

203  let isChecked = false;
204  let items = 0;
205  let isCompleted = false;
206  let reconnectAttempts = 0;
207  let sidebarProducts = createEmptyDomainInfo();
208  let messageQueue = Promise.resolve();
209  let processingProducts = 0;
210  let completedData = null;
211  let abortController = null;
212  let reconnectTimeoutId = null;
213  let lifetimeTimeoutId = null;
214  let isClosed = false;
215  let isTerminalStateReached = false;
216
217  const clearReconnectTimeout = () => {
218    if (reconnectTimeoutId !== null) {
219      clearTimeout(reconnectTimeoutId);
220      reconnectTimeoutId = null;
221    }
222  };
223
224  const clearLifetimeTimeout = () => {
225    if (lifetimeTimeoutId !== null) {
226      clearTimeout(lifetimeTimeoutId);
227      lifetimeTimeoutId = null;
228    }
229  };
230
231  const closeStreamConnection = () => {
232    isClosed = true;
233    clearReconnectTimeout();
234    clearLifetimeTimeout();
235    abortController?.abort();
236    abortController = null;
237  };
238
239  const isLifetimeExceeded = () => Date.now() - startedAt + LIFETIME_EPSILON_IN_MS >= SIDEBAR_STREAM_LIFETIME_IN_MS;
240
241  const stopStream = (
242    stopReason,
243    { disconnectReason = null, isSessionExpired = false, needsManualRefresh = false } = {},
244  ) => {
245    if (isTerminalStateReached || !isCurrentStream(streamId)) {
246      return;
247    }
248
249    isTerminalStateReached = true;
250    closeStreamConnection();
251
252    state.isLoading = false;
253    setStreamState({
254      status: 'stopped',
255      retryCount: reconnectAttempts,
256      needsManualRefresh,
257      isSessionExpired,
258      stopReason,
259      lastDisconnectReason: disconnectReason,
260    });
261  };
262
263  const tryFinishStream = () => {
264    if (!completedData || processingProducts > 0 || !isCurrentStream(streamId)) {
265      return;
266    }
267
268    completedData = null;
269    setDomainInfoData(sidebarProducts, {
270      isLoading: false,
271      isLoaded: true,
272    });
273  };
274
275  const processMessage = async (eventData) => {
276    if (!isCurrentStream(streamId)) {
277      return;
278    }
279
280    const payload = parseEventPayload(eventData);
281    if (!payload) {
282      return;
283    }
284
285    if (!isChecked) {
286      reloading = countDomainInfoItems(state.data) > 0;
287      sidebarProducts = createEmptyDomainInfo();
288      isChecked = true;
289    }
290
291    if (payload.status === 'completed') {
292      isCompleted = true;
293      completedData = payload;
294      tryFinishStream();
295      return;
296    }
297
298    processingProducts += 1;
299    try {
300      if (payload.data?.full_domain_name) {
301        payload.data.full_domain_name = await resolveSidebarDomainName(payload.data.full_domain_name, getApiFetcher());
302      }
303
304      if (!isCurrentStream(streamId)) {
305        return;
306      }
307
308      sidebarProducts = applySidebarProductEvent(sidebarProducts, payload);
309
310      // Once the initial snapshot is complete, later SSE messages are live deltas.
311      // Publishing one must not put every sidebar section back into its initial loading state.
312      if (isCompleted) {
313        setDomainInfoData(sidebarProducts, {
314          isLoading: false,
315          isLoaded: true,
316        });
317      } else if (!reloading) {
318        if (items <= 100 || items % 100 === 0) {
319          setDomainInfoData(sidebarProducts, {
320            isLoading: true,
321            isLoaded: false,
322          });
323        }
324        items += 1;
325      }
326    } finally {
327      processingProducts -= 1;
328      tryFinishStream();
329    }
330  };
331
332  const handleMessage = (eventData) => {
333    messageQueue = messageQueue.then(() => processMessage(eventData)).catch(noop);
334  };
335
336  const processMessageBlock = (messageBlock) => {
337    const eventData = getEventData(messageBlock);
338    if (eventData) {
339      handleMessage(eventData);
340    }
341  };
342
343  const flushBuffer = (buffer) => {
344    const messageBlocks = buffer.split('\n\n');
345    const nextBuffer = messageBlocks.pop() || '';
346    messageBlocks.forEach(processMessageBlock);
347    return nextBuffer;
348  };
349
350  const finalizeBuffer = (buffer, decoder) => {
351    const rest = decoder.decode();
352    const finalBuffer = rest ? `${buffer}${normalizeChunk(rest)}` : buffer;
353    const lastEventData = getEventData(finalBuffer.trim());
354    if (lastEventData) {
355      handleMessage(lastEventData);
356    }
357  };
358
359  const readStream = async (reader, decoder, buffer = '') => {
360    if (isClosed || isTerminalStateReached || !isCurrentStream(streamId)) {
361      return;
362    }
363
364    const { done, value } = await reader.read();
365    if (done) {
366      finalizeBuffer(buffer, decoder);
367      return;
368    }
369
370    const nextBuffer = flushBuffer(`${buffer}${normalizeChunk(decoder.decode(value, { stream: true }))}`);
371    await readStream(reader, decoder, nextBuffer);
372  };
373
374  const scheduleReconnect = (reconnect) => {
375    if (isClosed || isTerminalStateReached || reconnectTimeoutId !== null || !isCurrentStream(streamId)) {
376      return;
377    }
378
379    reconnectTimeoutId = setTimeout(() => {
380      reconnectTimeoutId = null;
381      reconnect().catch(noop);
382    }, RECONNECT_DELAY_IN_MS);
383  };
384
385  const handleUnexpectedDisconnect = async (disconnectReason, reconnect) => {
386    if (isClosed || isTerminalStateReached || !isCurrentStream(streamId)) {
387      return;
388    }
389
390    if (isLifetimeExceeded()) {
391      stopStream('timeout', {
392        disconnectReason,
393        needsManualRefresh: true,
394      });
395      return;
396    }
397
398    const sessionState = await checkUserSession();
399    if (sessionState === 'expired') {
400      stopStream('session_expired', {
401        disconnectReason,
402        isSessionExpired: true,
403      });
404      return;
405    }
406
407    if (reconnectAttempts >= SIDEBAR_STREAM_MAX_RECONNECT_ATTEMPTS) {
408      stopStream('retry_limit_reached', {
409        disconnectReason,
410        needsManualRefresh: true,
411      });
412      return;
413    }
414
415    reconnectAttempts += 1;
416    setStreamState({
417      status: 'reconnecting',
418      retryCount: reconnectAttempts,
419      needsManualRefresh: false,
420      isSessionExpired: false,
421      stopReason: null,
422      lastDisconnectReason: disconnectReason,
423    });
424    scheduleReconnect(reconnect);
425  };
426
427  const connectToStream = async () => {
428    if (isClosed || isTerminalStateReached || !isCurrentStream(streamId)) {
429      return;
430    }
431
432    const currentAbortController = new AbortController();
433    abortController = currentAbortController;
434
435    try {
436      const response = await fetch(sidebarStreamUrl, {
437        credentials: 'include',
438        headers: {
439          Accept: 'text/event-stream',
440        },
441        signal: currentAbortController.signal,
442      });
443
444      if (!response.ok || !response.body) {
445        throw new Error(`Domain info stream request failed with status ${response.status}`);
446      }
447
448      if (!isCurrentStream(streamId)) {
449        closeStreamConnection();
450        return;
451      }
452
453      reconnectAttempts = 0;
454      setStreamState({
455        status: 'active',
456        retryCount: 0,
457        needsManualRefresh: false,
458        isSessionExpired: false,
459        stopReason: null,
460        lastDisconnectReason: null,
461      });
462
463      const reader = response.body.getReader();
464      const decoder = new TextDecoder();
465
466      await readStream(reader, decoder);
467
468      if (!isClosed && !isTerminalStateReached && isCurrentStream(streamId)) {
469        await handleUnexpectedDisconnect('closed', connectToStream);
470      }
471    } catch {
472      if (!isClosed && !isTerminalStateReached && !currentAbortController.signal.aborted && isCurrentStream(streamId)) {
473        await handleUnexpectedDisconnect('error', connectToStream);
474      }
475    }
476  };
477
478  lifetimeTimeoutId = setTimeout(() => {
479    stopStream('timeout', {
480      needsManualRefresh: true,
481    });
482  }, SIDEBAR_STREAM_LIFETIME_IN_MS);
483
484  setStreamState({
485    ...createInitialDomainInfoStreamState(),
486    status: 'connecting',
487  });
488  connectToStream().catch(noop);
489
490  return {
491    close: closeStreamConnection,
492  };
493};
494
495const stopDomainInfoStream = ({ clearData = false } = {}) => {
496  state.streamId += 1;
497  state.connection?.close?.();
498  state.connection = null;
499  state.isLoading = false;

500  state.stream = createInitialDomainInfoStreamState();
501
502  if (clearData) {
503    state.userId = null;
504    state.data = createEmptyDomainInfo();
505    state.isLoaded = false;
506  }
507
508  notifySubscribers();
509};
510
511const ensureDomainInfoStream = (user, { force = false } = {}) => {
512  const userId = getUserId(user);
513  if (!userId) {
514    stopDomainInfoStream({ clearData: true });
515    return null;
516  }
517
518  const isSameUser = state.userId === userId;
519  const isRunning = ['connecting', 'active', 'reconnecting'].includes(state.stream.status);
520  if (isSameUser && isRunning && !force) {
521    return state.connection;
522  }
523
524  const preserveData = isSameUser && countDomainInfoItems(state.data) > 0;
525  if (state.connection) {
526    stopDomainInfoStream({ clearData: !preserveData });
527  }
528
529  state.userId = userId;
530  state.isLoading = true;
531  state.isLoaded = preserveData;
532  if (!preserveData) {
533    state.data = createEmptyDomainInfo();
534  }
535  notifySubscribers();
536
537  state.streamId += 1;
538  state.connection = createDomainInfoStreamConnection(userId, state.streamId);
539  return state.connection;
540};
541
542const getDomainInfo = () => cloneValue(state.data);
543
544const getDomainInfoState = () => getStateSnapshot();
545
546const subscribeDomainInfo = (listener, { emitCurrent = false } = {}) => {
547  if (typeof listener !== 'function') {
548    return noop;
549  }
550
551  state.subscribers.add(listener);
552  if (emitCurrent) {
553    listener(getStateSnapshot());
554  }
555
556  return () => {
557    state.subscribers.delete(listener);
558  };
559};
560
561const waitForDomainInfo = ({ timeoutMs = 0 } = {}) =>
562  new Promise((resolve, reject) => {
563    if (state.isLoaded) {
564      resolve(getDomainInfoState());
565      return;
566    }
567
568    let timeoutId = null;
569    const unsubscribe = subscribeDomainInfo((snapshot) => {
570      if (!snapshot.isLoaded) {
571        return;
572      }
573
574      if (timeoutId !== null) {
575        clearTimeout(timeoutId);
576      }
577      unsubscribe();
578      resolve(snapshot);
579    });
580
581    if (timeoutMs > 0) {
582      timeoutId = setTimeout(() => {
583        unsubscribe();
584        reject(new Error('Timed out waiting for domain info'));
585      }, timeoutMs);
586    }
587  });
588
589export {
590  createEmptyDomainInfo,
591  createInitialDomainInfoStreamState,
592  ensureDomainInfoStream,
593  stopDomainInfoStream,
594  getDomainInfo,
595  getDomainInfoState,
596  mutateDomainInfoData,
597  subscribeDomainInfo,
598  waitForDomainInfo,
599};