TS: add transform and transformAsync type definitions

.github/workflows/tests.yml

@@ -54,6 +54,8 @@ jobs:
     - uses: actions/setup-node@v6
     - run: npm ci
     - run: npm run test-type-definitions
+    - run: npm i typescript@4.7 @types/node@18.0.0
+    - run: npm run test-type-definitions-legacy
 
   lint:
     name: ESLint

lib/types/index.d.ts

@@ -8,6 +8,8 @@ export type NodeWebStream<T extends Data> = NodeWebReadableStream<T>;
 
 type Stream<T extends Data> = WebStream<T> | NodeWebStream<T>;
 type MaybeStream<T extends Data> = T | Stream<T>;
+/** Extract the type parameter `X` given `MaybeStream<X>` */
+type MaybeStreamDataType<T extends MaybeStream<Data>> = T extends Stream<infer X> ? X : T;
 
 export function readToEnd<T extends Data, JoinFn extends (chunks: T[]) => any = (chunks: T[]) => T>(
   input: MaybeStream<T>,
@@ -25,3 +27,61 @@ export function slice<T extends Data, InputType extends MaybeStream<T>>(
   begin: number,
   end?: number | typeof Infinity
 ): InputType; // same as 'typeof input'
+
+type Defined<T> = T extends undefined ? never : T;
+/**
+ * This type takes care of streamed return types for stream transform functions.
+ * This is needed to avoid adding support for `undefined` payloads in the Stream type:
+ * - if the callbacks return `Data | undefined`, we cannot infer output type info. Since this is considered a rare edge case,
+ *  this is translated into the overly-generic Stream<any>.
+ * - if CallbackReturnType is `undefined`, then an empty stream is returned, and it is modeled as Stream<never>.
+ */
+type TransformOutputMaybeToStream<
+  InputType,
+  CallbackReturnType extends Data | undefined | void
+> = InputType extends Data ?
+  CallbackReturnType :
+  Stream<CallbackReturnType extends Data ? CallbackReturnType : Data extends CallbackReturnType ? any /* never | Data */ : never>;
+export function transform<
+  InputType extends MaybeStream<Data>,
+  OutData extends Data,
+  ProcessFn extends undefined | ((chunk: MaybeStreamDataType<InputType>) => OutData | undefined | void) = undefined,
+  FinishFn extends undefined | (() => OutData | undefined | void) = undefined,
+>(
+  input: InputType,
+  process?: ProcessFn,
+  finish?: FinishFn,
+  queuingStrategy?: { highWaterMark: number }
+): TransformOutputMaybeToStream<
+  InputType,
+  ProcessFn extends undefined ?
+    FinishFn extends undefined ?
+      undefined :
+      // we do not cover the case where FinishFn or ProcessFn are (undefined | function)
+      ReturnType<Defined<FinishFn>> :
+    FinishFn extends undefined ?
+      ReturnType<Defined<ProcessFn>> :
+      (ReturnType<Defined<FinishFn>> | ReturnType<Defined<ProcessFn>>)
+>;
+
+export function transformAsync<
+  InputType extends MaybeStream<Data>,
+  OutData extends Data,
+  ProcessFn extends undefined | ((chunk: MaybeStreamDataType<InputType>) => Promise<OutData | undefined | void>) = undefined,
+  FinishFn extends undefined | (() => Promise<OutData | undefined | void>) = undefined,
+>(
+  input: InputType,
+  process?: ProcessFn,
+  finish?: FinishFn,
+  queuingStrategy?: { highWaterMark: number }
+): Promise<TransformOutputMaybeToStream<
+  InputType,
+  Awaited<ProcessFn extends undefined ?
+    FinishFn extends undefined ?
+      undefined :
+      // we do not cover the case where FinishFn or ProcessFn are (undefined | function)
+      ReturnType<Defined<FinishFn>> :
+    FinishFn extends undefined ?
+      ReturnType<Defined<ProcessFn>> :
+      (ReturnType<Defined<FinishFn>> | ReturnType<Defined<ProcessFn>>)
+  >>>;

lib/types/v4.7-v5.4/index.d.ts

@@ -10,6 +10,7 @@ export type NodeWebStream<T extends Data> = NodeWebReadableStream<T>;
 
 type Stream<T extends Data> = WebStream<T> | NodeWebStream<T>;
 type MaybeStream<T extends Data> = T | Stream<T>;
+type MaybeStreamDataType<T extends MaybeStream<Data>> = T extends Stream<infer X> ? X : T;
 
 export function readToEnd<T extends Data, JoinFn extends (chunks: T[]) => any = (chunks: T[]) => T>(
   input: MaybeStream<T>,
@@ -27,3 +28,54 @@ export function slice<T extends Data, InputType extends MaybeStream<T>>(
   begin: number,
   end?: number | typeof Infinity
 ): InputType; // same as 'typeof input'
+
+type Defined<T> = T extends undefined ? never : T;
+type TransformOutputMaybeToStream<
+  InputType,
+  CallbackReturnType extends Data | undefined | void
+> = InputType extends Data ?
+  CallbackReturnType :
+  Stream<CallbackReturnType extends Data ? CallbackReturnType : Data extends CallbackReturnType ? any /* never | Data */ : never>;
+export function transform<
+  InputType extends MaybeStream<Data>,
+  OutData extends Data,
+  ProcessFn extends undefined | ((chunk: MaybeStreamDataType<InputType>) => OutData | undefined | void) = undefined,
+  FinishFn extends undefined | (() => OutData | undefined | void) = undefined,
+>(
+  input: InputType,
+  process?: ProcessFn,
+  finish?: FinishFn,
+  queuingStrategy?: { highWaterMark: number }
+): TransformOutputMaybeToStream<
+  InputType,
+  ProcessFn extends undefined ?
+    FinishFn extends undefined ?
+      undefined :
+      // we do not cover the case where FinishFn or ProcessFn are (undefined | function)
+      ReturnType<Defined<FinishFn>> :
+    FinishFn extends undefined ?
+      ReturnType<Defined<ProcessFn>> :
+      (ReturnType<Defined<FinishFn>> | ReturnType<Defined<ProcessFn>>)
+>;
+
+export function transformAsync<
+  InputType extends MaybeStream<Data>,
+  OutData extends Data,
+  ProcessFn extends undefined | ((chunk: MaybeStreamDataType<InputType>) => Promise<OutData | undefined | void>) = undefined,
+  FinishFn extends undefined | (() => Promise<OutData | undefined | void>) = undefined,
+>(
+  input: InputType,
+  process?: ProcessFn,
+  finish?: FinishFn,
+  queuingStrategy?: { highWaterMark: number }
+): Promise<TransformOutputMaybeToStream<
+  InputType,
+  Awaited<ProcessFn extends undefined ?
+    FinishFn extends undefined ?
+      undefined :
+      // we do not cover the case where FinishFn or ProcessFn are (undefined | function)
+      ReturnType<Defined<FinishFn>> :
+    FinishFn extends undefined ?
+      ReturnType<Defined<ProcessFn>> :
+      (ReturnType<Defined<FinishFn>> | ReturnType<Defined<ProcessFn>>)
+  >>>;

package.json

@@ -79,6 +79,7 @@
   "scripts": {
     "test-type-definitions": "tsc --project test/tsconfig.json && tsx test/typescript.test.ts && npm run test-type-definitions-es5",
     "test-type-definitions-es5": "tsc --project test/tsconfig.es5.json && tsx --tsconfig test/tsconfig.es5.json test/typescript.test.ts",
+    "test-type-definitions-legacy": "tsc --project test/tsconfig.v4.7.json && tsx test/typescript.v4.7.test.ts",
     "test-browser": "karma start karma.conf.cjs",
     "test-node": "NODE_OPTIONS='--import=tsx' mocha ./test/node.test.ts ./test/common.test.ts",
     "lint": "eslint lib test eslint.config.js",

test/common.test.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 // @ts-expect-error Missing type definitions
-import { toStream, toArrayStream, readToEnd, slice, pipe, ArrayStream, transform, transformAsync } from  '@openpgp/web-stream-tools';
+import { toStream, toArrayStream, readToEnd, slice, pipe, ArrayStream, transform, transformAsync, type Stream } from  '@openpgp/web-stream-tools';
 
 describe('Common integration tests', () => {
   it('toStream/readToEnd', async () => {
@@ -42,7 +42,7 @@ describe('Common integration tests', () => {
 
   it('pipe from arraystream to arraystream', async () => {
     const input = 'chunk';
-    const inputStream = toArrayStream(input);
+    const inputStream: Stream<string> = toArrayStream(input);
     const outputStream = new ArrayStream();
     pipe(inputStream, outputStream);
     expect(await readToEnd(outputStream)).to.equal('chunk');
@@ -56,7 +56,7 @@ describe('Common integration tests', () => {
 
   it('transform arraystream', async () => {
     const input = 'chunk';
-    const streamedData = toArrayStream(input);
+    const streamedData: Stream<string> = toArrayStream(input);
     const transformed = transform(streamedData, (str: string) => str.toUpperCase());
     expect(await readToEnd(transformed)).to.equal('CHUNK');
   });
@@ -76,7 +76,7 @@ describe('Common integration tests', () => {
 
   it('transformAsync arraystream', async () => {
     const input = 'chunk';
-    const streamedData = toArrayStream(input);
+    const streamedData: Stream<string> = toArrayStream(input);
     const transformed = await transformAsync(streamedData, async (str: string) => str.toUpperCase());
     expect(await readToEnd(transformed)).to.equal('CHUNK');
   });

test/tsconfig.v4.7.json

@@ -0,0 +1,7 @@
+{
+    "extends": "../tsconfig.json",
+    "compilerOptions": {
+        "noEmit": true, // code tested by tsx
+    },
+    "include": ["./typescript.v4.7.test.ts"]
+}

test/typescript.test.ts

@@ -3,8 +3,7 @@ import assert from 'assert';
 import { Readable as NodeNativeReadableStream } from 'stream';
 import { ReadableStream as NodeWebReadableStream } from 'node:stream/web';
 import { ReadableStream as PonyfilledWebReadableStream } from 'web-streams-polyfill';
-import { type WebStream, type NodeWebStream, type Stream, toStream, type Data } from '@openpgp/web-stream-tools';
-import { readToEnd } from '@openpgp/web-stream-tools';
+import { type WebStream, type NodeWebStream, type Stream, toStream, type Data, transform, transformAsync, type MaybeStream, readToEnd } from '@openpgp/web-stream-tools';
 // @ts-expect-error missing defs
 import { ArrayStream, isArrayStream } from '@openpgp/web-stream-tools';
 
@@ -46,6 +45,45 @@ const newEmptyWebStream = <T extends Data>(): WebStream<T> => (
 
   assert(isArrayStream(new ArrayStream())) ; // ensure Array is actually extended in e.g. es5
 
+  const transformDefaultOutput: undefined = transform('string');
+  assert(transformDefaultOutput === undefined);
+  const transformUndefinedOutput: void = transform('string', () => {});
+  assert(transformUndefinedOutput === undefined);
+  const transformStreamUndefinedOutput: Stream<never> = transform(newEmptyWebStream(), () => {});
+  assert(transformStreamUndefinedOutput instanceof NodeWebReadableStream);
+  const transformOutputStreamString: Stream<string> = transform(newEmptyWebStream<string>(), () => '');
+  assert(transformOutputStreamString instanceof NodeWebReadableStream);
+  const transformProcessOutputString: string = transform('string', () => '');
+  assert(typeof transformProcessOutputString === 'string');
+  const transformConcatOutputString: string = transform('string', () => '', () => '');
+  assert(typeof transformConcatOutputString === 'string');
+  const transformFinishOutputString: string = transform('string', undefined, () => '');
+  assert(typeof transformFinishOutputString === 'string');
+  const transformProcessOutputStreamBytes: Stream<Uint8Array<ArrayBuffer>> = transform(
+    newEmptyWebStream<string>(),
+    () => new Uint8Array()
+  );
+  assert(transformProcessOutputStreamBytes instanceof NodeWebReadableStream);
+  const filterStream = <S extends MaybeStream<Data>>(stream: S, filterFn: (data: any) => boolean) => (
+    transform(stream, (chunk: Data) => filterFn(chunk) ? '' : undefined)
+  );
+  const transformUnionOutput: string | undefined = filterStream('string', () => false);
+  assert(transformUnionOutput === undefined);
+  const transformMismatchingProcessFinishOutput: Stream<string | Uint8Array<ArrayBuffer>> =
+    transform(newEmptyWebStream<string>(), () => new Uint8Array(), () => '');
+  assert(transformMismatchingProcessFinishOutput instanceof NodeWebReadableStream);
+
+  // @ts-expect-error on async callback
+  transform(newEmptyWebStream<string>(), async () => 'string');
+  const transformAsyncOutputStreamStringToBytes: Stream<Uint8Array> = await transformAsync(
+    newEmptyWebStream<string>(),
+    async () => new Uint8Array(),
+    async () => new Uint8Array()
+  );
+  assert(transformAsyncOutputStreamStringToBytes instanceof NodeWebReadableStream);
+  // @ts-expect-error on sync callback
+  transformAsync(newEmptyWebStream<string>(), async () => new Uint8Array(), () => new Uint8Array());
+
   console.log('TypeScript definitions are correct');
 })().catch(e => {
   console.error('TypeScript definitions tests failed by throwing the following error');

test/typescript.v4.7.test.ts

@@ -0,0 +1,81 @@
+/* global process */
+/**
+ * Tests cases are taken from `typescript.test.ts` but do not include the WebStreamPonyfill
+ * since the v3.3.3 ponyfill module resolution (required with the legacy TS version) does not work properly.
+ */
+import assert from 'assert';
+import { Readable as NodeNativeReadableStream } from 'stream';
+import { ReadableStream as NodeWebReadableStream } from 'node:stream/web';
+import { type NodeWebStream,type Stream, toStream, transform, transformAsync, type Data, readToEnd } from '@openpgp/web-stream-tools';
+// @ts-expect-error missing defs
+import { ArrayStream, isArrayStream } from '@openpgp/web-stream-tools';
+
+const newEmptyTypedNodeWebStream = <T extends Data>() => (
+  new NodeWebReadableStream({ start(ctrl) { ctrl.close(); } }) as NodeWebStream<T>
+);
+
+(async () => {
+  const nodeWebStream: NodeWebStream<string> = NodeNativeReadableStream.toWeb(new NodeNativeReadableStream());
+  assert(nodeWebStream instanceof NodeWebReadableStream);
+  // @ts-expect-error detect node stream type mismatch
+  const nodeNativeStream: NodeWebStream<string> = new NodeNativeReadableStream();
+  assert(nodeNativeStream instanceof NodeNativeReadableStream);
+
+  await readToEnd(new Uint8Array([1])) as Uint8Array;
+  await readToEnd(new Uint8Array([1]), _ => _) as Uint8Array[];
+  // @ts-expect-error expect string type
+  await readToEnd(newEmptyTypedNodeWebStream<string>()) as Uint8Array;
+  await readToEnd(newEmptyTypedNodeWebStream<string>(), () => new Uint8Array()) as Uint8Array;
+
+  const anotherNodeWebStream: NodeWebStream<string> = toStream(nodeWebStream);
+  assert(anotherNodeWebStream instanceof NodeWebReadableStream);
+  // The following type assertion may fail or not depending on the TS and @types/node versions;
+  // it doesn't fail with TS v4.7
+  // // @ts-expect-error expect node stream in output
+  // const expectedWebStreamButActualNodeStream: WebStream<string> = toStream(nodeWebStream);
+  // assert(expectedWebStreamButActualNodeStream instanceof NodeWebReadableStream);
+  const newStringStream: Stream<string> = toStream('chunk');
+  assert(newStringStream instanceof NodeWebReadableStream);
+  // @ts-expect-error detect type parameter mismatch
+  const anotherStringStream: Stream<Uint8Array> = toStream('chunk');
+  assert(anotherStringStream instanceof NodeWebReadableStream);
+
+  assert(isArrayStream(new ArrayStream())) ; // ensure Array is actually extended in e.g. es5
+
+  const transformDefaultOutput: undefined = transform('string');
+  assert(transformDefaultOutput === undefined);
+  const transformStreamUndefinedOutput: Stream<never> = transform(newEmptyTypedNodeWebStream(), () => {});
+  assert(transformStreamUndefinedOutput instanceof NodeWebReadableStream);
+  const transformOutputStreamString: Stream<string> = transform(newEmptyTypedNodeWebStream<string>(), () => '');
+  assert(transformOutputStreamString instanceof NodeWebReadableStream);
+  const transformProcessOutputString: string = transform('string', () => '');
+  assert(typeof transformProcessOutputString === 'string');
+  const transformConcatOutputString: string = transform('string', () => '', () => '');
+  assert(typeof transformConcatOutputString === 'string');
+  const transformFinishOutputString: string = transform('string', undefined, () => '');
+  assert(typeof transformFinishOutputString === 'string');
+  const transformProcessOutputStreamBytes: Stream<Uint8Array> = transform(
+    newEmptyTypedNodeWebStream<string>(),
+    () => new Uint8Array()
+  );
+  assert(transformProcessOutputStreamBytes instanceof NodeWebReadableStream);
+  const transformMismatchingProcessFinishOutput: Stream<string | Uint8Array> =
+    transform(newEmptyTypedNodeWebStream<string>(), () => new Uint8Array(), () => '');
+  assert(transformMismatchingProcessFinishOutput instanceof NodeWebReadableStream);
+  // @ts-expect-error on async callback
+  transform(newEmptyTypedNodeWebStream<string>(), async () => 'string');
+  const transformAsyncOutputStreamStringToBytes: Stream<Uint8Array> = await transformAsync(
+    newEmptyTypedNodeWebStream<string>(),
+    async () => new Uint8Array(),
+    async () => new Uint8Array()
+  );
+  assert(transformAsyncOutputStreamStringToBytes instanceof NodeWebReadableStream);
+  // @ts-expect-error on sync callback
+  transformAsync(newEmptyTypedNodeWebStream<string>(), async () => new Uint8Array(), () => new Uint8Array());
+
+  console.log('TypeScript definitions are correct');
+})().catch(e => {
+  console.error('TypeScript definitions tests failed by throwing the following error');
+  console.error(e);
+  process.exit(1);
+});