refactor(functional): move pipe function, rename old pipe to flow

src/core/remote/client.ts

@@ -7,7 +7,7 @@ import {
 import { Maybe } from '@/src/types';
 import { Deferred, defer } from '@/src/utils';
 import { debug } from '@/src/utils/loggers';
-import pipe from '@/src/utils/pipe';
+import { flow } from '@/src/utils/functional';
 import { nanoid } from 'nanoid';
 import { Socket, io } from 'socket.io-client';
 import { z } from 'zod';
@@ -160,13 +160,8 @@ export default class RpcClient {
     this.socket.on(STREAM_RESULT_EVENT, this.onStreamResultEvent);
   }
 
-  protected serialize = (obj: any) => {
-    return pipe(...this.serializers)(obj);
-  };
-
-  protected deserialize = (obj: any) => {
-    return pipe(...this.deserializers)(obj);
-  };
+  protected serialize = flow(...this.serializers);
+  protected deserialize = flow(...this.deserializers);
 
   async connect(uri: string) {
     await this.disconnect();

src/io/import/processors/restoreStateFile.ts

@@ -14,7 +14,8 @@ import {
   fileToDataSource,
 } from '@/src/io/import/dataSource';
 import { MANIFEST, isStateFile } from '@/src/io/state-file';
-import { ensureError, partition, pipe } from '@/src/utils';
+import { ensureError, partition } from '@/src/utils';
+import { pipe } from '@/src/utils/functional';
 import Pipeline, { PipelineContext } from '@/src/core/pipeline';
 import { Awaitable } from '@vueuse/core';
 import doneWithDataSource from '@/src/io/import/processors/doneWithDataSource';

src/utils/functional.ts

@@ -0,0 +1,42 @@
+type FlowFunction<T> = (o: T) => T;
+
+export function flow<T>(...fns: Array<FlowFunction<T>>) {
+  return (input: T) => fns.reduce((result, fn) => fn(result), input);
+}
+
+// Pipe code from
+// https://dev.to/ecyrbe/how-to-use-advanced-typescript-to-define-a-pipe-function-381h
+// Changed second parameter to rest/spread argument.
+type AnyFunc = (...arg: any) => any;
+
+type LastFnReturnType<F extends Array<AnyFunc>, Else = never> = F extends [
+  ...any[],
+  (...arg: any) => infer R
+]
+  ? R
+  : Else;
+
+type PipeArgs<F extends AnyFunc[], Acc extends AnyFunc[] = []> = F extends [
+  (...args: infer A) => infer B
+]
+  ? [...Acc, (...args: A) => B]
+  : F extends [(...args: infer A) => any, ...infer Tail]
+  ? Tail extends [(arg: infer B) => any, ...any[]]
+    ? PipeArgs<Tail, [...Acc, (...args: A) => B]>
+    : Acc
+  : Acc;
+
+// Example:
+// const myNumber = pipe(
+//   "1",
+//   (a: string) => Number(a),
+//   (c: number) => c + 1,
+//   (d: number) => `${d}`,
+//   (e: string) => Number(e)
+// );
+export function pipe<F extends AnyFunc[]>(
+  arg: Parameters<F[0]>[0],
+  ...fns: PipeArgs<F> extends F ? F : PipeArgs<F>
+): LastFnReturnType<F, ReturnType<F[0]>> {
+  return (fns.slice(1) as AnyFunc[]).reduce((acc, fn) => fn(acc), fns[0](arg));
+}

src/utils/index.ts

@@ -326,40 +326,3 @@ export function normalizeForStore<T, K extends keyof T>(objects: T[], key: K) {
 
   return { order, byKey };
 }
-
-// Pipe code from
-// https://dev.to/ecyrbe/how-to-use-advanced-typescript-to-define-a-pipe-function-381h
-type AnyFunc = (...arg: any) => any;
-
-type LastFnReturnType<F extends Array<AnyFunc>, Else = never> = F extends [
-  ...any[],
-  (...arg: any) => infer R
-]
-  ? R
-  : Else;
-
-type PipeArgs<F extends AnyFunc[], Acc extends AnyFunc[] = []> = F extends [
-  (...args: infer A) => infer B
-]
-  ? [...Acc, (...args: A) => B]
-  : F extends [(...args: infer A) => any, ...infer Tail]
-  ? Tail extends [(arg: infer B) => any, ...any[]]
-    ? PipeArgs<Tail, [...Acc, (...args: A) => B]>
-    : Acc
-  : Acc;
-
-// Example:
-// const valid = pipe(
-//   "1",
-//   (a: string) => Number(a),
-//   (c: number) => c + 1,
-//   (d: number) => `${d}`,
-//   (e: string) => Number(e)
-// );
-export function pipe<FirstFn extends AnyFunc, F extends AnyFunc[]>(
-  arg: Parameters<FirstFn>[0],
-  firstFn: FirstFn,
-  ...fns: PipeArgs<F> extends F ? F : PipeArgs<F>
-): LastFnReturnType<F, ReturnType<FirstFn>> {
-  return (fns as AnyFunc[]).reduce((acc, fn) => fn(acc), firstFn(arg));
-}