@stream coming soon?

[mmahalwy]

Congrats on 4.0! Glad to see @defer is supported. Any news on @stream?

[phryneas]

Hi @mmahalwy,

we're still in the process of figuring out the order of things we want to tackle next - stream is definitely on the list, but I can't give you an exact timeframe right now.

That said, since we have "pluggable" incremental handlers now, you could already experiment around with it by just copying our defer handler into your code base and extending it :)

https://github.com/apollographql/apollo-client/blob/100ce21266d1679db03ea7973fdfee389e7d3e5e/src/incremental/handlers/defer20220824.ts

[mmahalwy]

@phryneas oh cool, thanks for sharing. Does this GPT written code seem right? Happy to eventually create a PR to merge:

import type {
  DocumentNode,
  FormattedExecutionResult,
  GraphQLFormattedError,
} from "graphql";

import type { ApolloLink } from "@apollo/client/link";
import type { DeepPartial, HKT } from "@apollo/client/utilities";
import {
  DeepMerger,
  hasDirectives,
  isNonEmptyArray,
} from "@apollo/client/utilities/internal";

import type { Incremental } from "../types.js";

export declare namespace Stream20220824Handler {
  interface Stream20220824Result extends HKT {
    arg1: unknown; // TData
    arg2: unknown; // TExtensions
    return: Stream20220824Handler.Chunk<Record<string, unknown>>;
  }
  export interface TypeOverrides {
    AdditionalApolloLinkResultTypes: Stream20220824Result;
  }

  export type InitialResult<TData = Record<string, unknown>> = {
    data?: TData | null | undefined;
    errors?: ReadonlyArray<GraphQLFormattedError>;
    extensions?: Record<string, unknown>;
    hasNext: boolean;
    /** Optional optimization hint some servers include (safe to ignore) */
    // initialCount?: number;
  };

  export type IncrementalStreamPayload<TData = Record<string, unknown>> = {
    /** Canonical list streaming payload: append/insert these items at `path` */
    items?: Array<unknown> | null | undefined;
    /** Fallback shape used by some servers: merge like @defer using `data` + `path` */
    data?: TData | null | undefined;
    errors?: ReadonlyArray<GraphQLFormattedError>;
    extensions?: Record<string, unknown>;
    /** Path to the streamed field (the list field for `items`, or parent for `data`) */
    path?: Incremental.Path;
    /** Some servers include a label for routing */
    label?: string;
    /** Optional starting index; if omitted, append to the current array length */
    initialIndex?: number;
  };

  export type SubsequentResult<TData = Record<string, unknown>> = {
    data?: TData | null | undefined;
    errors?: ReadonlyArray<GraphQLFormattedError>;
    extensions?: Record<string, unknown>;
    hasNext: boolean;
    incremental?: Array<IncrementalStreamPayload<TData>>;
  };

  export type Chunk<TData extends Record<string, unknown>> =
    | InitialResult<TData>
    | SubsequentResult<TData>;
}

class StreamRequest<TData extends Record<string, unknown>>
  implements
    Incremental.IncrementalRequest<Stream20220824Handler.Chunk<TData>, TData>
{
  public hasNext = true;

  private errors: Array<GraphQLFormattedError> = [];
  private extensions: Record<string, any> = {};
  private data: any = {};

  private mergeIn(
    normalized: FormattedExecutionResult<TData>,
    merger: DeepMerger<any[]>
  ) {
    if (normalized.data !== undefined) {
      this.data = merger.merge(this.data, normalized.data);
    }
    if (normalized.errors) {
      this.errors.push(...normalized.errors);
    }
    Object.assign(this.extensions, normalized.extensions);
  }

  private ensurePath(parent: any, path: ReadonlyArray<string | number>) {
    let node = parent;
    for (let i = 0; i < path.length; i++) {
      const key = path[i] as any;
      const nextKey = i + 1 < path.length ? path[i + 1] : undefined;
      const shouldBeArray = typeof nextKey === "number";
      const shouldBeObject = !shouldBeArray;

      if (!(key in node) || node[key] == null) {
        node[key] = shouldBeArray ? [] : {};
      } else if (Array.isArray(node[key]) !== shouldBeArray) {
        // Coerce the container if the shape disagrees (defensive)
        node[key] = shouldBeArray ? [] : {};
      }
      node = node[key];
    }
    return node;
  }

  private applyItemsAtPath(
    root: any,
    path: ReadonlyArray<string | number>,
    items: unknown[] | null | undefined,
    initialIndex?: number
  ) {
    if (!items || !path || path.length === 0) return;

    // The path should resolve to the list field itself (e.g., ['user', 'friends'])
    const list = this.ensurePath(root, path);
    if (!Array.isArray(list)) {
      // If the server points one level deeper (rare), try to coerce
      // but we still bail if we can’t get an array
      return;
    }

    // Insert/append respecting `initialIndex` if provided
    const start =
      typeof initialIndex === "number" && initialIndex >= 0
        ? initialIndex
        : list.length;

    for (let i = 0; i < items.length; i++) {
      list[start + i] = items[i];
    }
  }

  handle(
    // In `no-cache` we may get `undefined`; keep last accumulated value
    cacheData: TData | DeepPartial<TData> | null | undefined = this.data,
    chunk: Stream20220824Handler.Chunk<TData>
  ): FormattedExecutionResult<TData> {
    this.hasNext = chunk.hasNext;
    this.data = cacheData;

    this.mergeIn(chunk, new DeepMerger());

    if (hasIncrementalStreamChunks(chunk)) {
      const merger = new DeepMerger();

      for (const incremental of chunk.incremental) {
        const { items, path, errors, extensions, initialIndex } = incremental;

        if (items && path) {
          // Primary stream behavior: place items into the list at `path`
          this.applyItemsAtPath(this.data, path, items, initialIndex);
          // Also hoover up any errors/extensions that accompany this piece
          if (errors?.length) this.errors.push(...errors);
          if (extensions) Object.assign(this.extensions, extensions);
          continue;
        }

        // Fallback: handle `data` + `path` similar to @defer for non-items payloads
        let { data, errors: dErrors, extensions: dExt } = incremental;
        if (data && path) {
          for (let i = path.length - 1; i >= 0; --i) {
            const key = path[i];
            const isNumericKey = !isNaN(+key as any);
            const parent: Record<string | number, any> = isNumericKey ? [] : {};
            parent[key as any] = data;
            data = parent as typeof data;
          }
        }

        this.mergeIn(
          {
            errors: dErrors,
            extensions: dExt,
            data: data ? (data as TData) : undefined,
          },
          merger
        );
      }
    }

    const result: FormattedExecutionResult<TData> = { data: this.data };

    if (isNonEmptyArray(this.errors)) {
      result.errors = this.errors;
    }
    if (Object.keys(this.extensions).length > 0) {
      result.extensions = this.extensions;
    }

    return result;
  }
}

/**
 * This handler implements the `@stream` directive as specified in the
 * same 20220824-era multipart protocol draft as your `@defer` handler.
 * It supports both canonical `items` payloads and a `data`+`path` fallback.
 */
export class Stream20220824Handler
  implements Incremental.Handler<Stream20220824Handler.Chunk<any>>
{
  isIncrementalResult(
    result: Record<string, any>
  ): result is
    | Stream20220824Handler.SubsequentResult
    | Stream20220824Handler.InitialResult {
    return "hasNext" in result;
  }

  extractErrors(result: ApolloLink.Result<any>) {
    const acc: GraphQLFormattedError[] = [];
    const push = ({
      errors,
    }: {
      errors?: ReadonlyArray<GraphQLFormattedError>;
    }) => {
      if (errors) acc.push(...errors);
    };

    if (this.isIncrementalResult(result)) {
      push(result);
      if (hasIncrementalStreamChunks(result)) {
        result.incremental.forEach(push);
      }
    }
    if (acc.length) return acc;
  }

  prepareRequest(request: ApolloLink.Request): ApolloLink.Request {
    if (hasDirectives(["stream"], request.query)) {
      const context = request.context ?? {};
      const http = (context.http ??= {});
      http.accept = [
        "multipart/mixed;streamSpec=20220824",
        ...(http.accept || []),
      ];
    }
    return request;
  }

  startRequest<TData extends Record<string, unknown>>(_: {
    query: DocumentNode;
  }) {
    return new StreamRequest<TData>();
  }
}

// only exported for use in tests
export function hasIncrementalStreamChunks(
  result: Record<string, any>
): result is Required<Stream20220824Handler.SubsequentResult> {
  return isNonEmptyArray(result.incremental);
}