Atomik Pipeline JS

atomik_pipeline.js

const setSym = Symbol( 'operations' );

module.exports = class AtomikPipeline {

  constructor( operationsSet ) {
    this[setSym] = operationsSet;
  }

  async run( _param ) {
    const rollback = [];
    let param = _param;
    for ( const item of this[setSym] ) {
      if ( item.rollback ) {
        rollback.push( item.rollback );
      }
      try {
        param = await item.operation( param );
      } catch ( err ) {
        await rollback.reverse()
          .reduce( ( chain, op ) => chain.then( op ), Promise.resolve( param ) );
        throw err;
      }
    }
    return param;
  }
};

// Usage
const operations = new Set()
  .add( { operation: function1, rollback: null } )
  .add( { operation: function2, rollback: rollbackFn1 } )
  .add( { operation: function3, rollback: rollbackFn2 } )
  .add( { operation: function4, rollback: rollbackFn3 } );
const result = await new AtomikPipeline( operations ).run( args );