Middleware down into step execution

grafast/grafast/src/bucket.ts

@@ -1,6 +1,6 @@
 // import type { GraphQLScalarType } from "graphql";
 
-import type { ExecutableStep } from ".";
+import type { ExecutableStep, GrafastExecutionArgs } from ".";
 import type { LayerPlan } from "./engine/LayerPlan";
 import type { MetaByMetaKey } from "./engine/OperationPlan";
 import type {
@@ -13,6 +13,8 @@ import type {
  * @internal
  */
 export interface RequestTools {
+  /** @internal */
+  args: GrafastExecutionArgs;
   /** The `timeSource.now()` at which the request started executing */
   startTime: number;
   /** The `timeSource.now()` at which the request should stop executing (if a timeout was configured) */

grafast/grafast/src/engine/executeBucket.ts

@@ -7,17 +7,22 @@ import { isFlaggedValue, SafeError } from "../error.js";
 import { inspect } from "../inspect.js";
 import type {
   BatchExecutionValue,
+  ExecuteStepEvent,
+  ExecutionDetails,
   ExecutionEntryFlags,
   ExecutionExtra,
   ExecutionValue,
   ForcedValues,
+  GrafastExecutionArgs,
   GrafastInternalResultsOrStream,
   GrafastResultsList,
   GrafastResultStreamList,
   IndexForEach,
   IndexMap,
   PromiseOrDirect,
+  StreamDetails,
   StreamMaybeMoreableArray,
+  StreamStepEvent,
   UnaryExecutionValue,
   UnbatchedExecutionExtra,
 } from "../interfaces.js";
@@ -111,7 +116,8 @@ export function executeBucket(
     }
   }
 
-  const { stopTime, eventEmitter } = requestContext;
+  const { stopTime, eventEmitter, args } = requestContext;
+  const { middlewares } = args;
   const {
     metaByMetaKey,
     size,
@@ -750,27 +756,45 @@ export function executeBucket(
           `${step} is using a legacy form of 'stream' which accepts multiple arguments, please see https://err.red/gev2`,
         );
       }
-      return step.stream({
+      const streamDetails: StreamDetails<readonly any[]> = {
         indexMap: makeIndexMap(count),
         indexForEach: makeIndexForEach(count),
         count,
         values,
         extra,
         streamOptions,
-      });
+      };
+      if (!step.isSyncAndSafe && middlewares) {
+        return middlewares.run(
+          "streamStep",
+          { args, step, streamDetails },
+          streamStepFromEvent,
+        );
+      } else {
+        return step.stream(streamDetails);
+      }
     } else {
       if (step.execute.length > 1) {
         throw new Error(
           `${step} is using a legacy form of 'execute' which accepts multiple arguments, please see https://err.red/gev2`,
         );
       }
-      return step.execute({
+      const executeDetails: ExecutionDetails<readonly any[]> = {
         indexMap: makeIndexMap(count),
         indexForEach: makeIndexForEach(count),
         count,
         values,
         extra,
-      });
+      };
+      if (!step.isSyncAndSafe && middlewares) {
+        return middlewares.run(
+          "executeStep",
+          { args, step, executeDetails },
+          executeStepFromEvent,
+        );
+      } else {
+        return step.execute(executeDetails);
+      }
     }
   }
 
@@ -1406,3 +1430,11 @@ function makeIndexForEach(count: number) {
   }
   return result;
 }
+
+function streamStepFromEvent(event: StreamStepEvent) {
+  return event.step.stream(event.streamDetails);
+}
+
+function executeStepFromEvent(event: ExecuteStepEvent) {
+  return event.step.execute(event.executeDetails);
+}

grafast/grafast/src/index.ts

@@ -47,11 +47,13 @@ import type {
   DataFromStep,
   EstablishOperationPlanEvent,
   ExecuteEvent,
+  ExecuteStepEvent,
   GrafastExecutionArgs,
   GrafastTimeouts,
   ParseAndValidateEvent,
   PrepareArgsEvent,
   ScalarInputPlanResolver,
+  StreamStepEvent,
   ValidateSchemaEvent,
 } from "./interfaces.js";
 import {
@@ -758,6 +760,12 @@ declare global {
       execute(event: ExecuteEvent): ReturnType<typeof execute>;
       subscribe(event: ExecuteEvent): ReturnType<typeof subscribe>;
       establishOperationPlan(event: EstablishOperationPlanEvent): OperationPlan;
+      executeStep(
+        event: ExecuteStepEvent,
+      ): PromiseOrDirect<GrafastResultsList<any>>;
+      streamStep(
+        event: StreamStepEvent,
+      ): PromiseOrDirect<GrafastResultStreamList<unknown>>;
     }
     interface Plugin {
       grafast?: {

grafast/grafast/src/interfaces.ts

@@ -29,7 +29,12 @@ import type { ObjMap } from "graphql/jsutils/ObjMap.js";
 import type { Bucket, RequestTools } from "./bucket.js";
 import type { OperationPlan } from "./engine/OperationPlan.js";
 import type { FlaggedValue, SafeError } from "./error.js";
-import type { ExecutableStep, ListCapableStep, ModifierStep } from "./step.js";
+import type {
+  ExecutableStep,
+  ListCapableStep,
+  ModifierStep,
+  StreamableStep,
+} from "./step.js";
 import type { __InputDynamicScalarStep } from "./steps/__inputDynamicScalar.js";
 import type {
   __InputListStep,
@@ -984,3 +989,13 @@ export interface EstablishOperationPlanEvent {
   planningTimeout: number | undefined;
   args: GrafastExecutionArgs;
 }
+export interface ExecuteStepEvent {
+  args: GrafastExecutionArgs;
+  step: ExecutableStep;
+  executeDetails: ExecutionDetails;
+}
+export interface StreamStepEvent {
+  args: GrafastExecutionArgs;
+  step: StreamableStep<unknown>;
+  streamDetails: StreamDetails;
+}

grafast/grafast/src/prepare.ts

@@ -299,6 +299,7 @@ function outputBucket(
 }
 
 function executePreemptive(
+  args: GrafastExecutionArgs,
   operationPlan: OperationPlan,
   variableValues: any,
   context: any,
@@ -337,6 +338,7 @@ function executePreemptive(
   const stopTime =
     executionTimeout !== null ? startTime + executionTimeout : null;
   const requestContext: RequestTools = {
+    args,
     startTime,
     stopTime,
     // toSerialize: [],
@@ -644,6 +646,7 @@ export function grafastPrepare(
 
   const executionTimeout = options.timeouts?.execution ?? null;
   return executePreemptive(
+    args,
     operationPlan,
     variableValues,
     context,