Promise-aware pipe implementation using recursion.

pipe.js

import { setTimeout } from 'timers/promises'

const pipe = async (context, ...callables) =>
  callables.length ? pipe(await callables.shift()(context), ...callables) : context
const flow = (...callables) => async (initial) => pipe(initial, ...callables)

// Use pipe to pass an initial value through the callables
console.log(await pipe(
  12,
  ($) => $ * 2,
))

// Use flow to create an async function that takes an initial value
// that is passed through the callables
const fetchTodo = flow(
  (id) => fetch(`https://jsonplaceholder.typicode.com/todos/${id}`),
  (response) => response.json()
)

console.log(await fetchTodo(1))

const console_log = ($) => (console.log($), $)

const waiter = flow(
  ($) => setTimeout(2000, $),
  ($) => $.split(","),
  console_log,
  ($) => setTimeout(2000, $),
  ($) => $.join(':'),
)

console.log(await waiter('1,2,3'))

const result = await pipe(
  "1, 2, 3",
  (_) => _.split(", "),
  (_) => _.map((i) => parseInt(i)),
  (_) => _.map((i) => i * 2),
  console_log,
  (_) => _.map((i) => i.toString()),
  (_) => (_.push('end'), _),
  (_) => _.join('-')
)

console.log(result)

// Since flows are functions that take a single argument, they can also be used as a callable
// in a pipe. 
const metaFlow = flow(
  waiter,
  ($) => $.split(':')[0],
  fetchTodo,
)

console.log(await metaFlow('1,2,3'))