perf: Completely eliminate Pipe class usage for massive performance gains

Copilot · DevilTea · DevilTea · commit 29eec3882076 · 2025-11-16T23:30:01.000+08:00
Co-authored-by: DevilTea &lt;16652879+DevilTea@users.noreply.github.com&gt;
diff --git a/packages/internal/src/core/core.ts b/packages/internal/src/core/core.ts
@@ -11,7 +11,7 @@ import type {
 	StepPluginImpl,
 	TStepPluginDef,
 } from './types'
-import { Pipe, runtimeExecutionStepDefMarker } from '../shared'
+import { runtimeExecutionStepDefMarker } from '../shared'
 
 /* @__NO_SIDE_EFFECTS__ */
 export function implStepPlugin<StepPluginDef extends TStepPluginDef>(stepImpl: StepPluginImpl<StepPluginDef>): StepPluginImpl<StepPluginDef> {
@@ -101,12 +101,23 @@ export function createPipeExecutor(
 	runtimeSteps: ((lastResult: ExecutionResult) => MaybePromise<ExecutionResult>)[],
 ): (value: unknown) => MaybePromise<ExecutionResult> {
 	return (value: unknown) => {
-		// Use optimized Pipe class
-		let pipe = new Pipe().add(v => ({ value: v } as ExecutionResult))
-		for (const s of runtimeSteps) {
-			pipe = pipe.add(s)
+		// Optimized: Direct execution without Pipe overhead
+		const len = runtimeSteps.length
+		let result: any = { value } as ExecutionResult
+
+		for (let i = 0; i < len; i++) {
+			// Check if current result is a promise
+			if (result instanceof Promise) {
+				// Once we hit async, chain all remaining steps
+				for (let j = i; j < len; j++) {
+					result = result.then(runtimeSteps[j])
+				}
+				return result
+			}
+			// Execute step synchronously
+			result = runtimeSteps[i](result)
 		}
-		return pipe.exec(value)
+		return result
 	}
 }
 /* @__NO_SIDE_EFFECTS__ */
diff --git a/packages/internal/src/steps/intersection/intersection.ts b/packages/internal/src/steps/intersection/intersection.ts
@@ -1,7 +1,6 @@
 import type { DefineExpectedValchecker, DefineStepMethod, DefineStepMethodMeta, ExecutionIssue, ExecutionResult, InferAsync, InferIssue, InferOutput, Next, TStepPluginDef, Use, Valchecker } from '../../core'
 import type { UnionToIntersection } from '../../shared'
 import { implStepPlugin } from '../../core'
-import { Pipe } from '../../shared'
 
 declare namespace Internal {
 	export type Branches = [Use<Valchecker>, ...Use<Valchecker>[]]
@@ -75,34 +74,59 @@ export const intersection = implStepPlugin<PluginDef>({
 		params: [branches],
 	}) => {
 		addSuccessStep((value) => {
-			const pipe = new Pipe<void>()
+			// Optimized: Direct processing without Pipe overhead
 			const issues: ExecutionIssue[] = []
+			const len = branches.length
 			let isInvalid = false
 
 			const processBranchResult = (result: ExecutionResult) => {
 				if (isFailure(result)) {
-					issues.push(...result.issues)
+					// Optimize: Avoid spread by using direct loop
+					for (const issue of result.issues) {
+						issues.push(issue)
+					}
 					isInvalid = true
 				}
 			}
 
-			for (const branch of branches) {
-				pipe.add(() => {
-					if (isInvalid)
-						return
+			// Process branches synchronously until we hit async or failure
+			for (let i = 0; i < len; i++) {
+				if (isInvalid)
+					break
 
-					const branchResult = branch['~execute'](value)
-					return branchResult instanceof Promise
-						? branchResult.then(processBranchResult)
-						: processBranchResult(branchResult)
-				})
+				const branchResult = branches[i]['~execute'](value)
+
+				if (branchResult instanceof Promise) {
+					// Hit async, chain remaining branches
+					let chain = branchResult.then((r) => {
+						processBranchResult(r)
+						return isInvalid
+					})
+
+					for (let j = i + 1; j < len; j++) {
+						const jBranch = branches[j]
+						chain = chain.then((invalid) => {
+							if (invalid)
+								return true
+							const jResult = jBranch['~execute'](value)
+							return jResult instanceof Promise
+								? jResult.then((r) => {
+										processBranchResult(r)
+										return isInvalid
+									})
+								: (processBranchResult(jResult), isInvalid)
+						})
+					}
+
+					return chain.then(() =>
+						(isInvalid || issues.length > 0) ? failure(issues) : success(value),
+					)
+				}
+
+				processBranchResult(branchResult)
 			}
 
-			const processResult = () => (isInvalid || issues.length > 0) ? failure(issues) : success(value)
-			const result = pipe.exec()
-			return result instanceof Promise
-				? result.then(processResult)
-				: processResult()
+			return (isInvalid || issues.length > 0) ? failure(issues) : success(value)
 		})
 	},
 })
diff --git a/packages/internal/src/steps/looseObject/looseObject.ts b/packages/internal/src/steps/looseObject/looseObject.ts
@@ -1,7 +1,6 @@
 import type { DefineExpectedValchecker, DefineStepMethod, DefineStepMethodMeta, ExecutionIssue, ExecutionResult, InferAsync, InferIssue, InferOutput, MessageHandler, Next, TStepPluginDef, Use, Valchecker } from '../../core'
 import type { IsEqual, IsExactlyAnyOrUnknown, Simplify, ValueOf } from '../../shared'
 import { implStepPlugin } from '../../core'
-import { Pipe } from '../../shared'
 
 declare namespace Internal {
 	export type Struct = Record<PropertyKey, Use<Valchecker> | [optional: Use<Valchecker>]>
@@ -114,8 +113,8 @@ export const looseObject = implStepPlugin<PluginDef>({
 				})
 			}
 
-			const knownKeys = new Set(Reflect.ownKeys(struct))
-			const pipe = new Pipe<void>()
+			// Optimized: Direct processing without Pipe overhead
+			const knownKeys = Array.from(Reflect.ownKeys(struct))
 			const issues: ExecutionIssue<any, any>[] = []
 			const output: Record<PropertyKey, any> = Object.defineProperties(
 				{},
@@ -124,31 +123,54 @@ export const looseObject = implStepPlugin<PluginDef>({
 
 			const processPropResult = (result: ExecutionResult, key: string | symbol) => {
 				if (isFailure(result)) {
-					issues.push(...result.issues.map(issue => prependIssuePath(issue, [key])))
-					return
+					// Optimize: Avoid spread + map by using direct loop
+					for (const issue of result.issues) {
+						issues.push(prependIssuePath(issue, [key]))
+					}
+				}
+				else {
+					output[key] = result.value
 				}
-				output[key] = result.value
 			}
 
-			for (const key of knownKeys) {
+			// Process properties synchronously until we hit async
+			for (let i = 0; i < knownKeys.length; i++) {
+				const key = knownKeys[i]!
 				const isOptional = Array.isArray(struct[key]!)
 				const propSchema = Array.isArray(struct[key]!) ? struct[key]![0]! : struct[key]!
 				const propValue = (value as any)[key]
-				pipe.add(() => {
-					const propResult = (isOptional && propValue === void 0)
-						? success(propValue)
-						: propSchema['~execute'](propValue)
-					return propResult instanceof Promise
-						? propResult.then(r => processPropResult(r, key))
-						: processPropResult(propResult, key)
-				})
+
+				const propResult = (isOptional && propValue === void 0)
+					? success(propValue)
+					: propSchema['~execute'](propValue)
+
+				if (propResult instanceof Promise) {
+					// Hit async, chain remaining properties
+					let chain = propResult.then(r => processPropResult(r, key))
+
+					for (let j = i + 1; j < knownKeys.length; j++) {
+						const jKey = knownKeys[j]!
+						const jIsOptional = Array.isArray(struct[jKey]!)
+						const jPropSchema = Array.isArray(struct[jKey]!) ? struct[jKey]![0]! : struct[jKey]!
+						const jPropValue = (value as any)[jKey]
+
+						chain = chain.then(() => {
+							const jPropResult = (jIsOptional && jPropValue === void 0)
+								? success(jPropValue)
+								: jPropSchema['~execute'](jPropValue)
+							return jPropResult instanceof Promise
+								? jPropResult.then(r => processPropResult(r, jKey))
+								: (processPropResult(jPropResult, jKey), undefined)
+						})
+					}
+
+					return chain.then(() => issues.length > 0 ? failure(issues) : success(output))
+				}
+
+				processPropResult(propResult, key)
 			}
 
-			const processResult = () => issues.length > 0 ? failure(issues) : success(output)
-			const result = pipe.exec()
-			return result instanceof Promise
-				? result.then(processResult)
-				: processResult()
+			return issues.length > 0 ? failure(issues) : success(output)
 		})
 	},
 })
