lib/src/util/support/Pipe.ts

/**
 * A closure that maps a given pipe item to a different type.
 */
import {Awaitable, Maybe} from './types'

export type PipeOperator<T, T2> = (subject: T) => T2

/**
 * A closure that maps a given pipe item to an item of the same type.
 */
export type ReflexivePipeOperator<T> = (subject: T) => Maybe<T>

/**
 * A condition or condition-resolving function for pipe methods.
 */
export type PipeCondition<T> = boolean | ((subject: T) => boolean)

/**
 * A class for writing chained/conditional operations in a data-flow manner.
 *
 * This is useful when you need to do a series of operations on an object, perhaps conditionally.
 */
export class Pipeline<TIn, TOut> {
    static id<T>(): Pipeline<T, T> {
        return new Pipeline(x => x)
    }

    constructor(
        protected readonly factory: (TIn: TIn) => TOut,
    ) {}

    /**
     * Apply the given operator to the item in the pipe, and return a new pipe with the result.
     *
     * @example
     * ```typescript
     * Pipe.wrap(2)
     *  .tap(x => x * 4)
     *  .get()  // => 8
     * ```
     *
     * @param op
     */
    tap<T2>(op: PipeOperator<TOut, T2>): Pipeline<TIn, T2> {
        return new Pipeline((val: TIn) => {
            return op(this.factory(val))
        })
    }

    /**
     * Like tap, but operates on a tuple with both the first value and the tapped value.
     * @param op
     */
    first<T2>(op: PipeOperator<[TIn, TOut], T2>): Pipeline<TIn, T2> {
        return new Pipeline((val: TIn) => {
            return op([val, this.factory(val)])
        })
    }

    /**
     * Like tap, but always returns the original pipe type.
     * @param op
     */
    peek<T2>(op: PipeOperator<TOut, T2>): Pipeline<TIn, TOut> {
        return new Pipeline((val: TIn) => {
            const nextVal = this.factory(val)
            op(nextVal)
            return nextVal
        })
    }

    /**
     * If `check` is truthy, apply the given operator to the item in the pipe and return the result.
     * Otherwise, just return the current pipe unchanged.
     *
     * @param check
     * @param op
     */
    when(check: PipeCondition<TOut>, op: ReflexivePipeOperator<TOut>): Pipeline<TIn, TOut> {
        return new Pipeline((val: TIn) => {
            const nextVal = this.factory(val)
            if ( this.checkCondition(check, nextVal) ) {
                const appliedVal = op(nextVal)
                if ( typeof appliedVal === 'undefined' ) {
                    return nextVal
                }

                return appliedVal
            }

            return nextVal
        })
    }

    /**
     * If `check` is falsy, apply the given operator to the item in the pipe and return the result.
     * Otherwise, just return the current pipe unchanged.
     *
     * @param check
     * @param op
     */
    unless(check: PipeCondition<TOut>, op: ReflexivePipeOperator<TOut>): Pipeline<TIn, TOut> {
        return new Pipeline((val: TIn) => {
            const nextVal = this.factory(val)
            if ( !this.checkCondition(check, nextVal) ) {
                const appliedVal = op(nextVal)
                if ( typeof appliedVal === 'undefined' ) {
                    return nextVal
                }

                return appliedVal
            }

            return nextVal
        })
    }

    /**
     * Apply the pipeline to an input.
     */
    apply(input: TIn): TOut {
        return this.factory(input)
    }

    protected checkCondition(check: PipeCondition<TOut>, val: TOut): boolean {
        return (typeof check === 'function' && check(val))
            || (typeof check !== 'function' && check)
    }
}

/**
 * A subject function that yields the value in the AsyncPipe.
 */
export type AsyncPipeResolver<T> = () => Awaitable<T>

/**
 * A closure that maps a given pipe item to a different type.
 */
export type AsyncPipeOperator<T, T2> = (subject: T) => Awaitable<T2>

export type PromisePipeOperator<T, T2> = (subject: T, resolve: (val: T2) => unknown, reject: (err: Error) => unknown) => Awaitable<unknown>

/**
 * A closure that maps a given pipe item to an item of the same type.
 */
export type ReflexiveAsyncPipeOperator<T> = (subject: T) => Awaitable<T|void>

/**
 * A condition or condition-resolving function for pipe methods.
 */
export type AsyncPipeCondition<T> = boolean | ((subject: T) => Awaitable<boolean>)

/**
 * An asynchronous version of the Pipe helper.
 */
export class AsyncPipe<T> {
    /**
     * Get an AsyncPipe with the given value in it.
     * @param subject
     */
    static wrap<subjectType>(subject: subjectType): AsyncPipe<subjectType> {
        return new AsyncPipe<subjectType>(() => subject)
    }

    constructor(
        /** The current value resolver of the pipe. */
        private subject: AsyncPipeResolver<T>,
    ) {}

    /**
     * Apply a transformative operator to the pipe.
     * @param op
     */
    tap<T2>(op: AsyncPipeOperator<T, T2>): AsyncPipe<T2> {
        return new AsyncPipe<T2>(async () => op(await this.subject()))
    }

    /**
     * Apply a transformative operator to the pipe, wrapping it
     * in a Promise and passing the resolve/reject callbacks to the
     * closure.
     * @param op
     */
    promise<T2>(op: PromisePipeOperator<T, T2>): AsyncPipe<T2> {
        return new AsyncPipe<T2>(() => {
            return new Promise<T2>((res, rej) => {
                (async () => this.subject())()
                    .then(subject => {
                        op(subject, res, rej)
                    })
            })
        })
    }

    /**
     * Apply an operator to the pipe, but return the reference
     * to the current pipe. The operator is resolved when the
     * overall pipe is resolved.
     * @param op
     */
    peek<T2>(op: AsyncPipeOperator<T, T2>): AsyncPipe<T> {
        return new AsyncPipe<T>(async () => {
            const subject = await this.subject()
            await op(subject)
            return subject
        })
    }

    /**
     * Apply an operator to the pipe, if the check condition passes.
     * @param check
     * @param op
     */
    when(check: AsyncPipeCondition<T>, op: ReflexiveAsyncPipeOperator<T>): AsyncPipe<T> {
        return new AsyncPipe<T>(async () => {
            let subject

            if ( typeof check === 'function' ) {
                check = await check(subject = await this.subject())
            }

            subject = subject ?? await this.subject()
            if ( check ) {
                return ((await op(subject)) ?? subject) as T
            }

            return subject as T
        })
    }

    /**
     * Apply an operator to the pipe, if the check condition fails.
     * @param check
     * @param op
     */
    unless(check: AsyncPipeCondition<T>, op: ReflexiveAsyncPipeOperator<T>): AsyncPipe<T> {
        if ( typeof check === 'function' ) {
            return this.when(async (subject: T) => !(await check(subject)), op)
        }

        return this.when(!check, op)
    }

    /**
     * Alias of `unless()`.
     * @param check
     * @param op
     */
    whenNot(check: AsyncPipeCondition<T>, op: ReflexiveAsyncPipeOperator<T>): AsyncPipe<T> {
        return this.unless(check, op)
    }

    /**
     * Get the transformed value from the pipe.
     */
    async resolve(): Promise<T> {
        return this.subject()
    }

    /** Get the transformed value from the pipe. Allows awaiting the pipe directly. */
    then(): Promise<T> {
        return this.resolve()
    }
}