Replace a re-sent livestream revision instead of rendering it twice

.changelog/5969-fix-livestream-resent-revision.md

@@ -0,0 +1,3 @@
+### Fixed
+
+- Fixed livestream revisions re-sent with the same ID shown twice, in PR [#5969](https://github.com/microsoft/BotFramework-WebChat/pull/5969), by [@OEvgeny](https://github.com/OEvgeny)

__tests__/html2/livestream/resentRevision.entity.html

@@ -0,0 +1,10 @@
+<!doctype html>
+<html>
+  <head>
+    <title>resentRevision (entity): redirects to ?streamingFormat=entity</title>
+    <script>
+      location = './resentRevision?streamingFormat=entity';
+    </script>
+  </head>
+  <body></body>
+</html>

__tests__/html2/livestream/resentRevision.html

@@ -0,0 +1,142 @@
+<!doctype html>
+<html lang="en-US">
+  <head>
+    <link href="/assets/index.css" rel="stylesheet" type="text/css" />
+    <link href="./css/pauseAnimation.css" rel="stylesheet" type="text/css" />
+    <script crossorigin="anonymous" src="https://unpkg.com/react@16.8.6/umd/react.production.min.js"></script>
+    <script crossorigin="anonymous" src="https://unpkg.com/react-dom@16.8.6/umd/react-dom.production.min.js"></script>
+    <script crossorigin="anonymous" src="/test-harness.js"></script>
+    <script crossorigin="anonymous" src="/test-page-object.js"></script>
+    <script crossorigin="anonymous" src="/__dist__/webchat-es5.js"></script>
+  </head>
+  <body>
+    <main id="webchat"></main>
+    <script type="importmap">
+      {
+        "imports": {
+          "@testduet/wait-for": "https://esm.sh/@testduet/wait-for"
+        }
+      }
+    </script>
+    <script type="module">
+      import { waitFor } from '@testduet/wait-for';
+
+      const streamingFormat = new URL(window.location.href).searchParams.get('streamingFormat') ?? 'channelData'; // can be entity or channelData
+
+      function addLivestreamingMetadata(activity, livestreamingMetadata) {
+        return streamingFormat === 'entity'
+          ? {
+              ...activity,
+              entities: [...(activity.entities ?? []), { ...livestreamingMetadata, type: 'streaminfo' }]
+            }
+          : {
+              ...activity,
+              channelData: { ...activity.channelData, ...livestreamingMetadata }
+            };
+      }
+
+      run(async function () {
+        const {
+          React: { createElement },
+          ReactDOM: { render },
+          WebChat: {
+            Components: { BasicWebChat, Composer },
+            decorator: { WebChatDecorator },
+            hooks: { useActivityKeys, useGetActivitiesByKey }
+          }
+        } = window; // Imports in UMD fashion.
+
+        const { directLine, store } = testHelpers.createDirectLineEmulator();
+        let currentActivityKeysWithId = [];
+
+        const Monitor = () => {
+          const [activityKeys] = useActivityKeys();
+          const getActivitiesByKey = useGetActivitiesByKey();
+
+          currentActivityKeysWithId = Object.freeze(
+            activityKeys.map(key => [key, getActivitiesByKey(key).map(({ id }) => id)])
+          );
+
+          return false;
+        };
+
+        render(
+          createElement(
+            WebChatDecorator,
+            {},
+            createElement(
+              Composer,
+              {
+                directLine,
+                store,
+                styleOptions: { hideSendBox: true }
+              },
+              createElement(BasicWebChat),
+              createElement(Monitor)
+            )
+          ),
+          document.getElementById('webchat')
+        );
+
+        await pageConditions.uiConnected();
+
+        // GIVEN: Bot is streaming a message with 2 revisions.
+        const firstRevision = addLivestreamingMetadata(
+          {
+            from: { id: 'u-00001', name: 'Bot', role: 'bot' },
+            id: 't-00001',
+            text: 'A quick',
+            timestamp: 1,
+            type: 'typing'
+          },
+          { streamSequence: 1, streamType: 'streaming' }
+        );
+
+        const secondRevision = addLivestreamingMetadata(
+          {
+            from: { id: 'u-00001', name: 'Bot', role: 'bot' },
+            id: 't-00002',
+            text: 'A quick brown fox',
+            timestamp: 2,
+            type: 'typing'
+          },
+          { streamId: 't-00001', streamSequence: 2, streamType: 'streaming' }
+        );
+
+        await directLine.emulateIncomingActivity(firstRevision);
+        await directLine.emulateIncomingActivity(secondRevision);
+
+        // WHEN: The second revision is received again (same ID), e.g. after a reconnect.
+        await directLine.emulateIncomingActivity(secondRevision);
+
+        // WHEN: Bot continues with the third revision.
+        // The store processes activities in order, so the re-sent revision is processed once this one appears.
+        await directLine.emulateIncomingActivity(
+          addLivestreamingMetadata(
+            {
+              from: { id: 'u-00001', name: 'Bot', role: 'bot' },
+              id: 't-00003',
+              text: 'A quick brown fox jumped over',
+              timestamp: 3,
+              type: 'typing'
+            },
+            { streamId: 't-00001', streamSequence: 3, streamType: 'streaming' }
+          )
+        );
+
+        // THEN: Should keep each revision once in the store.
+        expect(store.getState().activities.map(({ id }) => id)).toEqual(['t-00001', 't-00002', 't-00003']);
+
+        // THEN: Should have 1 activity key with each revision once.
+        await waitFor(() =>
+          expect(currentActivityKeysWithId).toEqual([[expect.any(String), ['t-00001', 't-00002', 't-00003']]])
+        );
+
+        // THEN: Should display 1 message with the third revision.
+        await pageConditions.numActivitiesShown(1);
+        expect(pageElements.activityContents()[0]).toHaveProperty('textContent', 'A quick brown fox jumped over');
+        await host.snapshot('local');
+      });
+    </script>
+  </body>
+</html>

packages/core/src/reducers/activities/sort/upsert.howToWithLivestream.spec.ts

@@ -513,3 +513,138 @@ scenario('re-sending an older revision of a finalized livestream in a grouping w
       expect(state.sortedActivities).toEqual([activityToExpectation(activity3, 1_000)]);
     });
 });
+
+scenario('re-sending a revision of an ongoing livestream in a grouping', bdd => {
+  const activity1 = buildActivity(
+    {
+      channelData: { streamSequence: 1, streamType: 'streaming' },
+      id: 'a-00001',
+      text: 'A quick',
+      timestamp: new Date(1_000).toISOString(),
+      type: 'typing'
+    },
+    { isPartOf: [{ '@id': '_:how-to:00001', '@type': 'HowTo' }], position: 1 }
+  );
+
+  const activity2 = buildActivity(
+    {
+      channelData: { streamId: 'a-00001', streamSequence: 2, streamType: 'streaming' },
+      id: 'a-00002',
+      text: 'A quick brown fox',
+      timestamp: new Date(2_000).toISOString(),
+      type: 'typing'
+    },
+    { isPartOf: [{ '@id': '_:how-to:00001', '@type': 'HowTo' }], position: 1 }
+  );
+
+  // Same ID with a changed text, e.g. delivered again after a reconnect.
+  const activity2Changed = buildActivity(
+    {
+      channelData: { streamId: 'a-00001', streamSequence: 2, streamType: 'streaming' },
+      id: 'a-00002',
+      text: 'a-00002:CHANGED',
+      timestamp: new Date(2_000).toISOString(),
+      type: 'typing'
+    },
+    { isPartOf: [{ '@id': '_:how-to:00001', '@type': 'HowTo' }], position: 1 }
+  );
+
+  bdd
+    .given('an ongoing livestream session with 2 revisions in a part grouping', () =>
+      upsert({ Date }, upsert({ Date }, INITIAL_STATE, activity1), activity2)
+    )
+    .when('the second revision is re-sent with a changed text', state => upsert({ Date }, state, activity2Changed))
+    .then('`activityMap` should have the changed revision', (_, state) => {
+      expect(state.activityMap.get('_:a-00002' as LocalId)).toEqual({
+        activity: activityToExpectation(activity2Changed),
+        activityLocalId: '_:a-00002' as LocalId,
+        logicalTimestamp: 2_000,
+        type: 'activity'
+      } satisfies ActivityMapEntry);
+    })
+    .and('should keep 2 revisions in the livestream session', (_, state) => {
+      expect(state.livestreamSessionMap.get('a-00001' as LivestreamSessionId)).toEqual({
+        activities: [
+          expect.objectContaining({ activityLocalId: '_:a-00001' as LocalId }),
+          expect.objectContaining({ activityLocalId: '_:a-00002' as LocalId })
+        ],
+        finalized: false,
+        logicalTimestamp: 1_000
+      } satisfies LivestreamSessionMapEntry);
+    })
+    .and('`sortedActivities` should have the changed revision once', (_, state) => {
+      expect(state.sortedActivities).toEqual([
+        activityToExpectation(activity1, 1_000),
+        activityToExpectation(activity2Changed, 2_000)
+      ]);
+    });
+});
+
+scenario('re-sending a middle revision of an ongoing livestream in a grouping', bdd => {
+  const activity1 = buildActivity(
+    {
+      channelData: { streamSequence: 1, streamType: 'streaming' },
+      id: 'a-00001',
+      text: 'A quick',
+      timestamp: new Date(1_000).toISOString(),
+      type: 'typing'
+    },
+    { isPartOf: [{ '@id': '_:how-to:00001', '@type': 'HowTo' }], position: 1 }
+  );
+
+  const activity2 = buildActivity(
+    {
+      channelData: { streamId: 'a-00001', streamSequence: 2, streamType: 'streaming' },
+      id: 'a-00002',
+      text: 'A quick brown fox',
+      timestamp: new Date(2_000).toISOString(),
+      type: 'typing'
+    },
+    { isPartOf: [{ '@id': '_:how-to:00001', '@type': 'HowTo' }], position: 1 }
+  );
+
+  const activity3 = buildActivity(
+    {
+      channelData: { streamId: 'a-00001', streamSequence: 3, streamType: 'streaming' },
+      id: 'a-00003',
+      text: 'A quick brown fox jumped',
+      timestamp: new Date(3_000).toISOString(),
+      type: 'typing'
+    },
+    { isPartOf: [{ '@id': '_:how-to:00001', '@type': 'HowTo' }], position: 1 }
+  );
+
+  // Same ID as the middle revision with a changed text, e.g. delivered again after a reconnect.
+  const activity2Changed = buildActivity(
+    {
+      channelData: { streamId: 'a-00001', streamSequence: 2, streamType: 'streaming' },
+      id: 'a-00002',
+      text: 'a-00002:CHANGED',
+      timestamp: new Date(2_000).toISOString(),
+      type: 'typing'
+    },
+    { isPartOf: [{ '@id': '_:how-to:00001', '@type': 'HowTo' }], position: 1 }
+  );
+
+  bdd
+    .given('an ongoing livestream session with 3 revisions in a part grouping', () =>
+      upsert({ Date }, upsert({ Date }, upsert({ Date }, INITIAL_STATE, activity1), activity2), activity3)
+    )
+    .when('the second revision is re-sent with a changed text', state => upsert({ Date }, state, activity2Changed))
+    .then('`activityMap` should have the changed revision', (_, state) => {
+      expect(state.activityMap.get('_:a-00002' as LocalId)).toEqual({
+        activity: activityToExpectation(activity2Changed),
+        activityLocalId: '_:a-00002' as LocalId,
+        logicalTimestamp: 2_000,
+        type: 'activity'
+      } satisfies ActivityMapEntry);
+    })
+    .and('`sortedActivities` should have the changed revision once, before the third revision', (_, state) => {
+      expect(state.sortedActivities).toEqual([
+        // Positions are not asserted: the slow path may resequence them, only the order matters.
+        activityToExpectation(activity1),
+        activityToExpectation(activity2Changed),
+        activityToExpectation(activity3)
+      ]);
+    });
+});