map-reduce.ts

import pMap from "p-map";
import { invokeLambdaAsync, invokeLambdaSync } from "./lambda-utils-aws";

export const mapReduce = async <TItem, TMap, TReduce>(options: {
  items: TItem[];
  map: {
    functionName: string;
    getPayloadForItem: (item: TItem) => TMap;
    concurrency: number;
  };
  reduce: {
    functionName: string;
    payload: TReduce;
  };
}): Promise<void> => {
  const { items, map, reduce } = options;
  // The map part
  await pMap(
    items,
    async (item) => {
      await invokeLambdaSync<TMap>(
        map.functionName,
        map.getPayloadForItem(item)
      );
    },
    { concurrency: map.concurrency }
  );
  // the reduce part
  await invokeLambdaAsync<TReduce>(reduce.functionName, reduce.payload);
};