inline child records that are ready

except for the initial result, which is left alone!

Author: yaacovCR

src/execution/IncrementalPublisher.ts

@@ -169,10 +169,18 @@ export interface FormattedCompletedResult {
   errors?: ReadonlyArray<GraphQLError>;
 }
 
+interface IncrementalStreamTarget {
+  errors?: Array<GraphQLError>;
+  items: Array<unknown>;
+}
+
 interface IncrementalAggregate {
   newPendingSources: Set<DeferredFragmentRecord | StreamRecord>;
   incrementalResults: Array<IncrementalResult>;
   completedResults: Array<CompletedResult>;
+  deferParents: Map<DeferredFragmentRecord, SubsequentDataRecord>;
+  initialStreams: Map<StreamRecord, SubsequentDataRecord>;
+  streamTargets: Map<StreamRecord, IncrementalStreamTarget>;
 }
 
 /**
@@ -234,14 +242,24 @@ export class IncrementalPublisher {
 
         const parentDeferUsage = deferUsage.ancestors[0];
 
-        const parent =
-          parentDeferUsage === undefined
-            ? (incrementalDataRecord as InitialResultRecord | StreamItemsRecord)
-            : this._deferredFragmentRecordFromDeferUsage(
-                parentDeferUsage,
-                newDeferMap,
-              );
-        parent.children.add(deferredFragmentRecord);
+        if (parentDeferUsage === undefined) {
+          const parent = incrementalDataRecord as
+            | InitialResultRecord
+            | StreamItemsRecord;
+          parent.children.add(deferredFragmentRecord);
+          if (isStreamItemsRecord(incrementalDataRecord)) {
+            incrementalDataRecord.childDefers.add(deferredFragmentRecord);
+          }
+        } else {
+          const parent = this._deferredFragmentRecordFromDeferUsage(
+            parentDeferUsage,
+            newDeferMap,
+          );
+          parent.children.add(deferredFragmentRecord);
+          if (!isInitialResultRecord(incrementalDataRecord)) {
+            incrementalDataRecord.childDefers.add(deferredFragmentRecord);
+          }
+        }
 
         newDeferMap.set(deferUsage, deferredFragmentRecord);
       }
@@ -307,9 +325,15 @@ export class IncrementalPublisher {
     });
 
     if (isDeferredGroupedFieldSetRecord(incrementalDataRecord)) {
+      incrementalDataRecord.childStreams.add(streamRecord);
       for (const parent of incrementalDataRecord.deferredFragmentRecords) {
         parent.children.add(streamItemsRecord);
       }
+    } else if (isStreamItemsRecord(incrementalDataRecord)) {
+      if (streamRecord !== incrementalDataRecord.streamRecord) {
+        incrementalDataRecord.childStreams.add(streamRecord);
+      }
+      incrementalDataRecord.children.add(streamItemsRecord);
     } else {
       incrementalDataRecord.children.add(streamItemsRecord);
     }
@@ -596,15 +620,23 @@ export class IncrementalPublisher {
       newPendingSources: new Set<DeferredFragmentRecord | StreamRecord>(),
       incrementalResults: [],
       completedResults: [],
+      deferParents: new Map(),
+      initialStreams: new Map(),
+      streamTargets: new Map(),
     };
   }
 
   private _incrementalReducer(
     aggregate: IncrementalAggregate,
     completedRecords: ReadonlySet<SubsequentResultRecord>,
   ): IncrementalAggregate {
-    const { newPendingSources, incrementalResults, completedResults } =
-      aggregate;
+    const {
+      newPendingSources,
+      incrementalResults,
+      completedResults,
+      deferParents,
+      initialStreams,
+    } = aggregate;
     for (const subsequentResultRecord of completedRecords) {
       for (const child of subsequentResultRecord.children) {
         if (child.filtered) {
@@ -636,14 +668,56 @@ export class IncrementalPublisher {
           if (subsequentResultRecord.streamRecord.errors.length > 0) {
             continue;
           }
-          const incrementalResult: IncrementalStreamResult = {
-            items: subsequentResultRecord.items,
-            path: subsequentResultRecord.streamRecord.path,
-          };
-          if (subsequentResultRecord.errors.length > 0) {
-            incrementalResult.errors = subsequentResultRecord.errors;
+          this._updateTargets(subsequentResultRecord, aggregate);
+          const streamRecord = subsequentResultRecord.streamRecord;
+          const initialStream = initialStreams.get(streamRecord);
+          if (initialStream === undefined) {
+            initialStreams.set(streamRecord, subsequentResultRecord);
+            const streamResult: IncrementalStreamResult = {
+              items: subsequentResultRecord.items,
+              path: streamRecord.path,
+            };
+            if (subsequentResultRecord.errors.length > 0) {
+              streamResult.errors = subsequentResultRecord.errors;
+            }
+            incrementalResults.push(streamResult);
+          } else if (isStreamItemsRecord(initialStream)) {
+            if (initialStream.streamRecord === streamRecord) {
+              if (subsequentResultRecord.items.length > 0) {
+                initialStream.items.push(...subsequentResultRecord.items);
+              }
+              this._updateTargetErrors(
+                initialStream,
+                subsequentResultRecord.errors,
+              );
+            } else {
+              const target = this._findTargetFromStreamPath(
+                initialStream.items,
+                initialStream.path,
+                streamRecord.path,
+              ) as Array<unknown>;
+              if (subsequentResultRecord.items.length > 0) {
+                target.push(...subsequentResultRecord.items);
+              }
+              this._updateTargetErrors(
+                initialStream,
+                subsequentResultRecord.errors,
+              );
+            }
+          } else {
+            const target = this._findTarget(
+              initialStream.data as ObjMap<unknown>,
+              initialStream.path,
+              streamRecord.path,
+            ) as Array<unknown>;
+            if (subsequentResultRecord.items.length > 0) {
+              target.push(...subsequentResultRecord.items);
+            }
+            this._updateTargetErrors(
+              initialStream,
+              subsequentResultRecord.errors,
+            );
           }
-          incrementalResults.push(incrementalResult);
         }
       } else {
         newPendingSources.delete(subsequentResultRecord);
@@ -653,18 +727,50 @@ export class IncrementalPublisher {
         if (subsequentResultRecord.errors.length > 0) {
           continue;
         }
+        const parent = deferParents.get(subsequentResultRecord);
         for (const deferredGroupedFieldSetRecord of subsequentResultRecord.deferredGroupedFieldSetRecords) {
           if (!deferredGroupedFieldSetRecord.sent) {
+            this._updateTargets(deferredGroupedFieldSetRecord, aggregate);
             deferredGroupedFieldSetRecord.sent = true;
-            const incrementalResult: IncrementalDeferResult = {
-              // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-              data: deferredGroupedFieldSetRecord.data!,
-              path: deferredGroupedFieldSetRecord.path,
-            };
-            if (deferredGroupedFieldSetRecord.errors.length > 0) {
-              incrementalResult.errors = deferredGroupedFieldSetRecord.errors;
+            if (parent === undefined) {
+              const incrementalResult: IncrementalDeferResult = {
+                // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+                data: deferredGroupedFieldSetRecord.data!,
+                path: deferredGroupedFieldSetRecord.path,
+              };
+              if (deferredGroupedFieldSetRecord.errors.length > 0) {
+                incrementalResult.errors = deferredGroupedFieldSetRecord.errors;
+              }
+              incrementalResults.push(incrementalResult);
+            } else {
+              const deferredFragmentTarget = isStreamItemsRecord(parent)
+                ? this._findTargetFromStreamPath(
+                    parent.items,
+                    parent.path,
+                    deferredGroupedFieldSetRecord.path,
+                  )
+                : this._findTarget(
+                    parent.data as ObjMap<unknown>,
+                    parent.path,
+                    deferredGroupedFieldSetRecord.path,
+                  );
+
+              const deferredGroupedFieldSetTarget = this._findTarget(
+                deferredFragmentTarget,
+                subsequentResultRecord.path,
+                deferredGroupedFieldSetRecord.path,
+              );
+              const data =
+                deferredGroupedFieldSetRecord.data as ObjMap<unknown>;
+              for (const key of Object.keys(data)) {
+                (deferredGroupedFieldSetTarget as ObjMap<unknown>)[key] =
+                  data[key];
+              }
+              this._updateTargetErrors(
+                parent,
+                deferredGroupedFieldSetRecord.errors,
+              );
             }
-            incrementalResults.push(incrementalResult);
           }
         }
       }
@@ -673,6 +779,72 @@ export class IncrementalPublisher {
     return aggregate;
   }
 
+  private _updateTargets(
+    subsequentDataRecord: SubsequentDataRecord,
+    aggregate: IncrementalAggregate,
+  ): void {
+    const { childDefers, childStreams } = subsequentDataRecord;
+    const { deferParents, initialStreams } = aggregate;
+    for (const childDefer of childDefers) {
+      deferParents.set(childDefer, subsequentDataRecord);
+    }
+    for (const childStream of childStreams) {
+      initialStreams.set(childStream, subsequentDataRecord);
+    }
+  }
+
+  private _findTarget(
+    data: ObjMap<unknown> | Array<unknown>,
+    dataPath: ReadonlyArray<string | number>,
+    targetPath: ReadonlyArray<string | number>,
+  ): ObjMap<unknown> | Array<unknown> {
+    let i = 0;
+    while (i < dataPath.length) {
+      i++;
+    }
+    let dataOrItems = data;
+    while (i < targetPath.length) {
+      const key = targetPath[i++];
+      const value = (dataOrItems as ObjMap<unknown>)[key as string];
+      dataOrItems = value as ObjMap<unknown>;
+    }
+    return dataOrItems;
+  }
+
+  private _findTargetFromStreamPath(
+    data: ObjMap<unknown> | Array<unknown>,
+    dataPath: ReadonlyArray<string | number>,
+    targetPath: ReadonlyArray<string | number>,
+  ): ObjMap<unknown> | Array<unknown> {
+    const pathToStream = [...dataPath];
+    const start = pathToStream.pop() as number;
+    let i = 0;
+    while (i < pathToStream.length) {
+      i++;
+    }
+    const adjustedIndex = (targetPath[i++] as number) - start;
+    let dataOrItems = (data as Array<unknown>)[adjustedIndex];
+    while (i < targetPath.length) {
+      const key = targetPath[i++];
+      const value = (dataOrItems as ObjMap<unknown>)[key as string];
+      dataOrItems = value as ObjMap<unknown>;
+    }
+    return dataOrItems as ObjMap<unknown> | Array<unknown>;
+  }
+
+  private _updateTargetErrors(
+    subsequentDataRecord: SubsequentDataRecord,
+    errors: ReadonlyArray<GraphQLError>,
+  ): void {
+    for (const error of errors) {
+      if (subsequentDataRecord.errors === undefined) {
+        subsequentDataRecord.errors = [error];
+      } else {
+        subsequentDataRecord.errors.push(error);
+      }
+    }
+  }
+
   private _incrementalFinalizer(
     aggregate: IncrementalAggregate,
   ): SubsequentIncrementalExecutionResult {
@@ -832,6 +1004,8 @@ export class DeferredGroupedFieldSetRecord {
   deferPriority: number;
   streamPriority: number;
   deferredFragmentRecords: ReadonlyArray<DeferredFragmentRecord>;
+  childDefers: Set<DeferredFragmentRecord>;
+  childStreams: Set<StreamRecord>;
   groupedFieldSet: GroupedFieldSet;
   shouldInitiateDefer: boolean;
   errors: Array<GraphQLError>;
@@ -852,6 +1026,8 @@ export class DeferredGroupedFieldSetRecord {
     this.deferPriority = opts.deferPriority;
     this.streamPriority = opts.streamPriority;
     this.deferredFragmentRecords = opts.deferredFragmentRecords;
+    this.childDefers = new Set();
+    this.childStreams = new Set();
     this.groupedFieldSet = opts.groupedFieldSet;
     this.shouldInitiateDefer = opts.shouldInitiateDefer;
     this.errors = [];
@@ -918,6 +1094,8 @@ export class StreamItemsRecord {
   streamPriority: number;
   items: Array<unknown>;
   children: Set<SubsequentResultRecord>;
+  childDefers: Set<DeferredFragmentRecord>;
+  childStreams: Set<StreamRecord>;
   isFinalRecord?: boolean;
   isCompletedAsyncIterator?: boolean;
   isCompleted: boolean;
@@ -937,6 +1115,8 @@ export class StreamItemsRecord {
     this.deferPriority = opts.deferPriority;
     this.streamPriority = opts.streamPriority;
     this.children = new Set();
+    this.childDefers = new Set();
+    this.childStreams = new Set();
     this.errors = [];
     this.isCompleted = false;
     this.filtered = false;
@@ -954,13 +1134,18 @@ export class StreamItemsRecord {
   }
 }
 
-export type IncrementalDataRecord =
-  | InitialResultRecord
-  | DeferredGroupedFieldSetRecord
-  | StreamItemsRecord;
+export type IncrementalDataRecord = InitialResultRecord | SubsequentDataRecord;
+
+type SubsequentDataRecord = DeferredGroupedFieldSetRecord | StreamItemsRecord;
 
 export type SubsequentResultRecord = DeferredFragmentRecord | StreamItemsRecord;
 
+function isInitialResultRecord(
+  incrementalDataRecord: unknown,
+): incrementalDataRecord is InitialResultRecord {
+  return incrementalDataRecord instanceof InitialResultRecord;
+}
+
 function isDeferredGroupedFieldSetRecord(
   incrementalDataRecord: unknown,
 ): incrementalDataRecord is DeferredGroupedFieldSetRecord {

src/execution/__tests__/defer-test.ts

@@ -2055,12 +2055,12 @@ describe('Execute: defer directive', () => {
       {
         incremental: [
           {
-            data: { name: 'slow', friends: [{}, {}, {}] },
+            data: {
+              name: 'slow',
+              friends: [{ name: 'Han' }, { name: 'Leia' }, { name: 'C-3PO' }],
+            },
             path: ['hero'],
           },
-          { data: { name: 'Han' }, path: ['hero', 'friends', 0] },
-          { data: { name: 'Leia' }, path: ['hero', 'friends', 1] },
-          { data: { name: 'C-3PO' }, path: ['hero', 'friends', 2] },
         ],
         completed: [
           { path: ['hero'] },