diff --git a/packages/twenty-front/src/modules/apollo/optimistic-effect/utils/buildSortedConnectionEdges.ts b/packages/twenty-front/src/modules/apollo/optimistic-effect/utils/buildSortedConnectionEdges.ts
new file mode 100644
index 0000000000..5dfa5c1108
--- /dev/null
+++ b/packages/twenty-front/src/modules/apollo/optimistic-effect/utils/buildSortedConnectionEdges.ts
@@ -0,0 +1,53 @@
+import { type FieldFunctionOptions } from '@apollo/client/cache';
+
+import { sortCachedObjectEdges } from '@/apollo/optimistic-effect/utils/sortCachedObjectEdges';
+import { type RecordGqlRefEdge } from '@/object-record/cache/types/RecordGqlRefEdge';
+import { type RecordGqlNode } from '@/object-record/graphql/types/RecordGqlNode';
+import { type RecordGqlOperationOrderBy } from 'twenty-shared/types';
+
+type NewEntry = {
+ edge: RecordGqlRefEdge;
+ record: RecordGqlNode;
+};
+
+type BuildSortedConnectionEdgesArgs = {
+ currentEdges: readonly RecordGqlRefEdge[];
+ newEntries: readonly NewEntry[];
+ orderBy: RecordGqlOperationOrderBy | undefined;
+ readField: FieldFunctionOptions['readField'];
+};
+
+export const buildSortedConnectionEdges = ({
+ currentEdges,
+ newEntries,
+ orderBy,
+ readField,
+}: BuildSortedConnectionEdgesArgs): RecordGqlRefEdge[] => {
+ const firstEdges: RecordGqlRefEdge[] = [];
+ const lastEdges: RecordGqlRefEdge[] = [];
+ const sortableEdges: RecordGqlRefEdge[] = [];
+
+ for (const { edge, record } of newEntries) {
+ if (record.position === 'first') {
+ firstEdges.push(edge);
+ } else if (record.position === 'last') {
+ lastEdges.push(edge);
+ } else {
+ sortableEdges.push(edge);
+ }
+ }
+
+ let middleEdges: RecordGqlRefEdge[];
+
+ if (Array.isArray(orderBy) && orderBy.length > 0) {
+ middleEdges = sortCachedObjectEdges({
+ edges: [...currentEdges, ...sortableEdges],
+ orderBy,
+ readCacheField: readField,
+ });
+ } else {
+ middleEdges = [...sortableEdges, ...currentEdges];
+ }
+
+ return [...firstEdges, ...middleEdges, ...lastEdges];
+};
diff --git a/packages/twenty-front/src/modules/apollo/optimistic-effect/utils/triggerCreateRecordsOptimisticEffect.ts b/packages/twenty-front/src/modules/apollo/optimistic-effect/utils/triggerCreateRecordsOptimisticEffect.ts
index 6cf1d30721..20fd4fa995 100644
--- a/packages/twenty-front/src/modules/apollo/optimistic-effect/utils/triggerCreateRecordsOptimisticEffect.ts
+++ b/packages/twenty-front/src/modules/apollo/optimistic-effect/utils/triggerCreateRecordsOptimisticEffect.ts
@@ -1,6 +1,7 @@
import { type ApolloCache, type StoreObject } from '@apollo/client';
import { isNonEmptyString } from '@sniptt/guards';
+import { buildSortedConnectionEdges } from '@/apollo/optimistic-effect/utils/buildSortedConnectionEdges';
import { triggerUpdateRelationsOptimisticEffect } from '@/apollo/optimistic-effect/utils/triggerUpdateRelationsOptimisticEffect';
import { type EnrichedObjectMetadataItem } from '@/object-metadata/types/EnrichedObjectMetadataItem';
import { type RecordGqlRefEdge } from '@/object-record/cache/types/RecordGqlRefEdge';
@@ -119,120 +120,78 @@ export const triggerCreateRecordsOptimisticEffect = ({
hasPreviousPage?: boolean;
}>('pageInfo', rootQueryCachedObjectRecordConnection);
- const nextRootQueryCachedRecordEdges = rootQueryCachedRecordEdges
- ? [...rootQueryCachedRecordEdges]
- : [];
+ const newEntries = recordsToCreate.flatMap<{
+ edge: RecordGqlRefEdge;
+ record: RecordGqlNode;
+ }>((recordToCreate) => {
+ if (!isNonEmptyString(recordToCreate.id)) {
+ return [];
+ }
- const nextQueryCachedPageInfo = isDefined(rootQueryCachedPageInfo)
- ? { ...rootQueryCachedPageInfo }
- : {};
+ if (
+ isDefined(rootQueryFilter) &&
+ shouldMatchRootQueryFilter === true &&
+ !isRecordMatchingFilter({
+ record: recordToCreate,
+ filter: rootQueryFilter,
+ objectMetadataItem,
+ })
+ ) {
+ return [];
+ }
- const hasAddedRecords = recordsToCreate
- .map((recordToCreate) => {
- if (isNonEmptyString(recordToCreate.id)) {
- if (
- isDefined(rootQueryFilter) &&
- shouldMatchRootQueryFilter === true
- ) {
- const recordToCreateMatchesThisRootQueryFilter =
- isRecordMatchingFilter({
- record: recordToCreate,
- filter: rootQueryFilter,
- objectMetadataItem,
- });
+ const node = toReference(recordToCreate);
- if (!recordToCreateMatchesThisRootQueryFilter) {
- return false;
- }
- }
+ if (!isDefined(node)) {
+ return [];
+ }
- const recordToCreateReference = toReference(recordToCreate);
+ const recordAlreadyInCache = rootQueryCachedRecordEdges?.some(
+ (cachedEdge) =>
+ cache.identify(node) === cache.identify(cachedEdge.node),
+ );
- if (!recordToCreateReference) {
- throw new Error(
- `Failed to create reference for record with id: ${recordToCreate.id}`,
- );
- }
+ if (recordAlreadyInCache === true) {
+ return [];
+ }
- const recordAlreadyInCache = rootQueryCachedRecordEdges?.some(
- (cachedEdge) => {
- return (
- cache.identify(recordToCreateReference) ===
- cache.identify(cachedEdge.node)
- );
- },
- );
+ return [
+ {
+ edge: {
+ __typename: getEdgeTypename(objectMetadataItem.nameSingular),
+ node,
+ cursor: encodeCursor(recordToCreate),
+ },
+ record: recordToCreate,
+ },
+ ];
+ });
- if (isDefined(recordToCreateReference) && !recordAlreadyInCache) {
- const cursor = encodeCursor(recordToCreate);
-
- const edge = {
- __typename: getEdgeTypename(objectMetadataItem.nameSingular),
- node: recordToCreateReference,
- cursor,
- };
-
- if (
- !isDefined(recordToCreate.position) ||
- recordToCreate.position === 'first'
- ) {
- nextRootQueryCachedRecordEdges.unshift(edge);
- nextQueryCachedPageInfo.startCursor = cursor;
- } else if (recordToCreate.position === 'last') {
- nextRootQueryCachedRecordEdges.push(edge);
- nextQueryCachedPageInfo.endCursor = cursor;
- } else if (typeof recordToCreate.position === 'number') {
- let index = Math.round(
- nextRootQueryCachedRecordEdges.length *
- recordToCreate.position,
- );
-
- if (recordToCreate.position < 0) {
- index = Math.max(
- 0,
- nextRootQueryCachedRecordEdges.length +
- Math.round(recordToCreate.position),
- );
- } else if (recordToCreate.position > 1) {
- index = nextRootQueryCachedRecordEdges.length;
- }
-
- index = Math.max(
- 0,
- Math.min(index, nextRootQueryCachedRecordEdges.length),
- );
-
- nextRootQueryCachedRecordEdges.splice(index, 0, edge);
-
- if (index === 0) {
- nextQueryCachedPageInfo.startCursor = cursor;
- } else if (
- index ===
- nextRootQueryCachedRecordEdges.length - 1
- ) {
- nextQueryCachedPageInfo.endCursor = cursor;
- }
- }
-
- return true;
- }
- }
-
- return false;
- })
- .some((hasAddedRecord) => hasAddedRecord);
-
- if (!hasAddedRecords) {
+ if (newEntries.length === 0) {
return rootQueryCachedObjectRecordConnection;
}
+ const sortedEdges = buildSortedConnectionEdges({
+ currentEdges: rootQueryCachedRecordEdges ?? [],
+ newEntries,
+ orderBy: rootQueryVariables?.orderBy,
+ readField,
+ });
+
return {
...rootQueryCachedObjectRecordConnection,
- edges: nextRootQueryCachedRecordEdges,
+ edges: sortedEdges,
totalCount: isDefined(rootQueryCachedRecordTotalCount)
- ? rootQueryCachedRecordTotalCount + 1
+ ? rootQueryCachedRecordTotalCount + newEntries.length
: undefined,
- pageInfo: nextQueryCachedPageInfo,
+ pageInfo: {
+ ...(rootQueryCachedPageInfo ?? {}),
+ startCursor:
+ sortedEdges[0]?.cursor ?? rootQueryCachedPageInfo?.startCursor,
+ endCursor:
+ sortedEdges[sortedEdges.length - 1]?.cursor ??
+ rootQueryCachedPageInfo?.endCursor,
+ },
};
},
},
diff --git a/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx b/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx
index 1231811c76..53c38e46b8 100644
--- a/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx
+++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx
@@ -41,13 +41,35 @@ export const SSEQuerySubscribeEffect = () => {
const requiredQueryListeners = useAtomStateValue(requiredQueryListenersState);
const activeQueryListeners = useAtomStateValue(activeQueryListenersState);
- const updateQueryListeners = useCallback(async () => {
+ const handleError = useCallback(
+ (error: unknown) => {
+ if (CombinedGraphQLErrors.is(error)) {
+ const extensions = getGraphqlErrorExtensionsFromError(error);
+
+ if (
+ isGracefullyHandledEventStreamError({
+ subCode: extensions?.subCode,
+ code: extensions?.code,
+ })
+ ) {
+ store.set(activeQueryListenersState.atom, []);
+ store.set(shouldDestroyEventStreamState.atom, true);
+
+ return;
+ }
+
+ throw new Error(`Unhandled error for event stream: ${error.message}`);
+ }
+ },
+ [store],
+ );
+
+ const syncAdditions = useCallback(async () => {
if (!isDefined(sseEventStreamId)) {
return;
}
const requiredQueryListeners = store.get(requiredQueryListenersState.atom);
-
const activeQueryListeners = store.get(activeQueryListenersState.atom);
const queryListenersToAdd = requiredQueryListeners.filter(
@@ -57,12 +79,9 @@ export const SSEQuerySubscribeEffect = () => {
),
);
- const queryListenersToRemove = activeQueryListeners.filter(
- (listener) =>
- !requiredQueryListeners.some(
- (requiredListener) => requiredListener.queryId === listener.queryId,
- ),
- );
+ if (queryListenersToAdd.length === 0) {
+ return;
+ }
try {
for (const queryListenerToAdd of queryListenersToAdd) {
@@ -83,7 +102,46 @@ export const SSEQuerySubscribeEffect = () => {
return;
}
}
+ } catch (error) {
+ handleError(error);
+ return;
+ }
+
+ const currentActive = store.get(activeQueryListenersState.atom);
+
+ store.set(activeQueryListenersState.atom, [
+ ...currentActive,
+ ...queryListenersToAdd,
+ ]);
+ }, [addQueryToEventStream, handleError, sseEventStreamId, store]);
+
+ const syncRemovals = useCallback(async () => {
+ if (!isDefined(sseEventStreamId)) {
+ return;
+ }
+
+ const freshRequiredQueryListeners = store.get(
+ requiredQueryListenersState.atom,
+ );
+ const activeQueryListeners = store.get(activeQueryListenersState.atom);
+
+ const queryListenersToRemove = activeQueryListeners.filter(
+ (listener) =>
+ !freshRequiredQueryListeners.some(
+ (requiredListener) => requiredListener.queryId === listener.queryId,
+ ),
+ );
+
+ if (queryListenersToRemove.length === 0) {
+ return;
+ }
+
+ const removedQueryIds = new Set(
+ queryListenersToRemove.map((listener) => listener.queryId),
+ );
+
+ try {
for (const queryListenerToRemove of queryListenersToRemove) {
const result = await removeQueryFromEventStream({
variables: {
@@ -102,60 +160,72 @@ export const SSEQuerySubscribeEffect = () => {
}
}
} catch (error) {
- if (CombinedGraphQLErrors.is(error)) {
- const extensions = getGraphqlErrorExtensionsFromError(error);
+ handleError(error);
- if (
- isGracefullyHandledEventStreamError({
- subCode: extensions?.subCode,
- code: extensions?.code,
- })
- ) {
- store.set(activeQueryListenersState.atom, []);
- store.set(shouldDestroyEventStreamState.atom, true);
-
- return;
- }
-
- throw new Error(`Unhandled error for event stream: ${error.message}`);
- }
+ return;
}
- store.set(activeQueryListenersState.atom, requiredQueryListeners);
- }, [
- addQueryToEventStream,
- removeQueryFromEventStream,
- sseEventStreamId,
- store,
- ]);
+ const currentActive = store.get(activeQueryListenersState.atom);
- const debouncedUpdateQueryListeners = useDebouncedCallback(
- updateQueryListeners,
- 1000,
- { leading: true },
- );
+ store.set(
+ activeQueryListenersState.atom,
+ currentActive.filter(
+ (listener) => !removedQueryIds.has(listener.queryId),
+ ),
+ );
+ }, [handleError, removeQueryFromEventStream, sseEventStreamId, store]);
+
+ const debouncedSyncAdditions = useDebouncedCallback(syncAdditions, 1000, {
+ leading: true,
+ });
+
+ const debouncedSyncRemovals = useDebouncedCallback(syncRemovals, 200, {
+ leading: false,
+ });
useEffect(() => {
if (!isNonEmptyString(sseEventStreamId) || !sseEventStreamReady) {
return;
}
- const areRequiredQueryListenersDifferentFromActiveQueryListeners =
- compareArraysOfObjectsByProperty(
- requiredQueryListeners,
- activeQueryListeners,
- 'queryId',
- );
+ const areDifferent = compareArraysOfObjectsByProperty(
+ requiredQueryListeners,
+ activeQueryListeners,
+ 'queryId',
+ );
- if (areRequiredQueryListenersDifferentFromActiveQueryListeners) {
- debouncedUpdateQueryListeners();
+ if (!areDifferent) {
+ return;
+ }
+
+ const hasAdditions = requiredQueryListeners.some(
+ (listener) =>
+ !activeQueryListeners.some(
+ (activeListener) => activeListener.queryId === listener.queryId,
+ ),
+ );
+
+ const hasRemovals = activeQueryListeners.some(
+ (listener) =>
+ !requiredQueryListeners.some(
+ (requiredListener) => requiredListener.queryId === listener.queryId,
+ ),
+ );
+
+ if (hasAdditions) {
+ debouncedSyncAdditions();
+ }
+
+ if (hasRemovals) {
+ debouncedSyncRemovals();
}
}, [
sseEventStreamId,
sseEventStreamReady,
requiredQueryListeners,
activeQueryListeners,
- debouncedUpdateQueryListeners,
+ debouncedSyncAdditions,
+ debouncedSyncRemovals,
]);
return null;
diff --git a/packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/form-action/components/WorkflowEditActionFormFiller.tsx b/packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/form-action/components/WorkflowEditActionFormFiller.tsx
index e8ac3b1f40..46a5f28029 100644
--- a/packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/form-action/components/WorkflowEditActionFormFiller.tsx
+++ b/packages/twenty-front/src/modules/workflow/workflow-steps/workflow-actions/form-action/components/WorkflowEditActionFormFiller.tsx
@@ -6,6 +6,7 @@ import { type FieldMetadata } from '@/object-record/record-field/ui/types/FieldM
import { SidePanelFooter } from '@/ui/layout/side-panel/components/SidePanelFooter';
import { useWorkflowRunIdOrThrow } from '@/workflow/hooks/useWorkflowRunIdOrThrow';
import { type WorkflowFormAction } from '@/workflow/types/Workflow';
+import { WorkflowRunSSESubscribeEffect } from '@/workflow/workflow-diagram/components/WorkflowRunSSESubscribeEffect';
import { WorkflowStepBody } from '@/workflow/workflow-steps/components/WorkflowStepBody';
import { useUpdateWorkflowRunStep } from '@/workflow/workflow-steps/hooks/useUpdateWorkflowRunStep';
import { WorkflowFormFieldInput } from '@/workflow/workflow-steps/workflow-actions/components/WorkflowFormFieldInput';
@@ -100,6 +101,7 @@ export const WorkflowEditActionFormFiller = ({
return (
<>
+
{formData.map((field) => {
if (field.type === 'RECORD') {