Use new SSE setup for workflow runs (#17309)

- Add a specific subscriber for workflow run
- Workflow runs use apollo cache. Adding updateRecordFromCache on record
updates

packages/twenty-front/src/modules/object-record/record-table/record-table-row/components/RecordTableRow.tsx

@@ -15,6 +15,8 @@ import { ListenRecordUpdatesEffect } from '@/sse-db-event/components/ListenRecor
 import { getDefaultRecordFieldsToListen } from '@/sse-db-event/utils/getDefaultRecordFieldsToListen';
 import { useRecoilComponentFamilyValue } from '@/ui/utilities/state/component-state/hooks/useRecoilComponentFamilyValue';
 import { useRecoilComponentValue } from '@/ui/utilities/state/component-state/hooks/useRecoilComponentValue';
+import { useIsFeatureEnabled } from '@/workspace/hooks/useIsFeatureEnabled';
+import { FeatureFlagKey } from '~/generated/graphql';
 
 type RecordTableRowProps = {
   recordId: string;
@@ -40,6 +42,9 @@ export const RecordTableRow = ({
   const isRowFocusActive = useRecoilComponentValue(
     isRecordTableRowFocusActiveComponentState,
   );
+  const isSseDbEventsEnabled = useIsFeatureEnabled(
+    FeatureFlagKey.IS_SSE_DB_EVENTS_ENABLED,
+  );
 
   return isFirstRowOfGroup ? (
     <RecordTableDraggableTrFirstRowOfGroup
@@ -58,11 +63,13 @@ export const RecordTableRow = ({
       <RecordTableFieldsCells />
       <RecordTablePlusButtonCellPlaceholder />
       <RecordTableLastEmptyCell />
-      <ListenRecordUpdatesEffect
-        objectNameSingular={objectNameSingular}
-        recordId={recordId}
-        listenedFields={listenedFields}
-      />
+      {!isSseDbEventsEnabled && (
+        <ListenRecordUpdatesEffect
+          objectNameSingular={objectNameSingular}
+          recordId={recordId}
+          listenedFields={listenedFields}
+        />
+      )}
     </RecordTableDraggableTrFirstRowOfGroup>
   ) : (
     <RecordTableDraggableTr

packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerOptimisticEffectFromSseUpdateEvents.ts

@@ -5,6 +5,7 @@ import { type ObjectMetadataItem } from '@/object-metadata/types/ObjectMetadataI
 import { getObjectTypename } from '@/object-record/cache/utils/getObjectTypename';
 import { getRecordFromCache } from '@/object-record/cache/utils/getRecordFromCache';
 import { getRecordNodeFromRecord } from '@/object-record/cache/utils/getRecordNodeFromRecord';
+import { updateRecordFromCache } from '@/object-record/cache/utils/updateRecordFromCache';
 import { generateDepthRecordGqlFieldsFromObject } from '@/object-record/graphql/record-gql-fields/utils/generateDepthRecordGqlFieldsFromObject';
 import { useObjectPermissions } from '@/object-record/hooks/useObjectPermissions';
 import { useRefetchAggregateQueriesForObjectMetadataItem } from '@/object-record/hooks/useRefetchAggregateQueriesForObjectMetadataItem';
@@ -68,12 +69,29 @@ export const useTriggerOptimisticEffectFromSseUpdateEvents = () => {
           computeReferences: false,
         });
 
+        if (
+          !isDefined(cachedRecord) ||
+          !isDefined(cachedRecordWithConnection)
+        ) {
+          continue;
+        }
+
         const computedOptimisticRecord = {
+          ...cachedRecord,
           ...updatedRecord,
           id: updatedRecord.id,
           __typename: getObjectTypename(objectMetadataItem.nameSingular),
         };
 
+        updateRecordFromCache({
+          objectMetadataItems,
+          objectMetadataItem,
+          cache: apolloCoreClient.cache,
+          record: computedOptimisticRecord,
+          recordGqlFields,
+          objectPermissionsByObjectMetadataId,
+        });
+
         const computedOptimisticRecordWithConnection = getRecordNodeFromRecord({
           record: computedOptimisticRecord,
           objectMetadataItem,
@@ -81,10 +99,7 @@ export const useTriggerOptimisticEffectFromSseUpdateEvents = () => {
           recordGqlFields,
         });
 
-        if (
-          !isDefined(cachedRecordWithConnection) ||
-          !isDefined(computedOptimisticRecordWithConnection)
-        ) {
+        if (!isDefined(computedOptimisticRecordWithConnection)) {
           continue;
         }
 

packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowRunCard.tsx

@@ -1,15 +1,19 @@
 import { SKELETON_LOADER_HEIGHT_SIZES } from '@/activities/components/SkeletonLoader';
+import { CoreObjectNameSingular } from '@/object-metadata/types/CoreObjectNameSingular';
 import { ListenRecordUpdatesEffect } from '@/sse-db-event/components/ListenRecordUpdatesEffect';
 import { useTargetRecord } from '@/ui/layout/contexts/useTargetRecord';
 import { getWorkflowVisualizerComponentInstanceId } from '@/workflow/utils/getWorkflowVisualizerComponentInstanceId';
+import { WorkflowRunSSESubscribeEffect } from '@/workflow/workflow-diagram/components/WorkflowRunSSESubscribeEffect';
 import { WorkflowRunVisualizer } from '@/workflow/workflow-diagram/components/WorkflowRunVisualizer';
 import { WorkflowRunVisualizerEffect } from '@/workflow/workflow-diagram/components/WorkflowRunVisualizerEffect';
 import { WorkflowRunVisualizerComponentInstanceContext } from '@/workflow/workflow-diagram/states/contexts/WorkflowRunVisualizerComponentInstanceContext';
 import { WorkflowVisualizerComponentInstanceContext } from '@/workflow/workflow-diagram/states/contexts/WorkflowVisualizerComponentInstanceContext';
+import { useIsFeatureEnabled } from '@/workspace/hooks/useIsFeatureEnabled';
 import { useTheme } from '@emotion/react';
 import styled from '@emotion/styled';
 import { Suspense, useId } from 'react';
 import Skeleton, { SkeletonTheme } from 'react-loading-skeleton';
+import { FeatureFlagKey } from '~/generated/graphql';
 
 const StyledLoadingSkeletonContainer = styled.div`
   display: flex;
@@ -41,6 +45,9 @@ const LoadingSkeleton = () => {
 export const WorkflowRunCard = () => {
   const targetRecord = useTargetRecord();
   const componentId = useId();
+  const isSseDbEventsEnabled = useIsFeatureEnabled(
+    FeatureFlagKey.IS_SSE_DB_EVENTS_ENABLED,
+  );
 
   return (
     <WorkflowVisualizerComponentInstanceContext.Provider
@@ -56,11 +63,15 @@ export const WorkflowRunCard = () => {
         }}
       >
         <WorkflowRunVisualizerEffect workflowRunId={targetRecord.id} />
-        <ListenRecordUpdatesEffect
-          objectNameSingular={targetRecord.targetObjectNameSingular}
-          recordId={targetRecord.id}
-          listenedFields={['status', 'state']}
-        />
+        {isSseDbEventsEnabled ? (
+          <WorkflowRunSSESubscribeEffect workflowRunId={targetRecord.id} />
+        ) : (
+          <ListenRecordUpdatesEffect
+            objectNameSingular={CoreObjectNameSingular.WorkflowRun}
+            recordId={targetRecord.id}
+            listenedFields={['status', 'state']}
+          />
+        )}
         <Suspense fallback={<LoadingSkeleton />}>
           <WorkflowRunVisualizer workflowRunId={targetRecord.id} />
         </Suspense>

packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowRunSSESubscribeEffect.tsx

@@ -0,0 +1,24 @@
+import { CoreObjectNameSingular } from '@/object-metadata/types/CoreObjectNameSingular';
+import { useListenToObjectRecordEventsForQuery } from '@/sse-db-event/hooks/useListenToObjectRecordEventsForQuery';
+
+export const WorkflowRunSSESubscribeEffect = ({
+  workflowRunId,
+}: {
+  workflowRunId: string;
+}) => {
+  const queryId = `workflow-run-${workflowRunId}`;
+
+  useListenToObjectRecordEventsForQuery({
+    queryId,
+    operationSignature: {
+      objectNameSingular: CoreObjectNameSingular.WorkflowRun,
+      variables: {
+        filter: {
+          id: { eq: workflowRunId },
+        },
+      },
+    },
+  });
+
+  return null;
+};