Data.ts

import { Parser } from "json2csv";
const csv = require("csvtojson");
const fs = require("fs").promises;
import { Config as MssqlConfig, execute } from "lesstedious";

export type Dsl<A> = () => Promise<Array<A>>;

export const union = <A>(x: Dsl<A>, y: Dsl<A>): Dsl<A> => async () => {
  const [x2, y2] = await Promise.all([x(), y()]);
  return [...x2, ...y2];
};

export const map = <A, B>(dsl: Dsl<A>, mapper: (a: A) => B): Dsl<B> => () =>
  dsl().then((x) => x.map(mapper));

export const filter = <A>(dsl: Dsl<A>, pred: (a: A) => boolean): Dsl<A> => () =>
  dsl().then((x) => x.filter(pred));

type Join<A, B> = {
  a: A;
  b: B;
};

export const join = <A, B>(
  x: Dsl<A>,
  y: Dsl<B>,
  on: (a: A, b: B) => boolean
): Dsl<Join<A, B>> => async () => {
  const [as, bs] = await Promise.all([x(), y()]);
  return as.reduce((p, c) => {
    const matches = bs.filter((b2) => on(c, b2));
    if (matches.length > 0) {
      return [...p, ...matches.map((match) => ({ a: c, b: match }))];
    } else {
      return p;
    }
  }, []);
};

export const leftJoin = <A, B>(
  x: Dsl<A>,
  y: Dsl<B>,
  on: (a: A, b: B) => boolean
): Dsl<Join<A, B | undefined>> => async () => {
  const [as, bs] = await Promise.all([x(), y()]);
  return as.reduce((p, c) => {
    const matches = bs.filter((b2) => on(c, b2));
    if (matches.length > 0) {
      return [...p, ...matches.map((match) => ({ a: c, b: match }))];
    } else {
      return [...p, { a: c, b: undefined }];
    }
  }, []);
};

export const toCsv = <A>(dsl: Dsl<A>, path: string): Dsl<A> => async () => {
  const json2csvParser = new Parser();
  const data = await dsl();
  const csvData = json2csvParser.parse(data);
  await fs.writeFile(path, csvData);
  return data;
};

export const fromCsv = <A>(path: string): Dsl<A> => () => csv().fromFile(path);

export const fromJson = <A>(as: A[]): Dsl<A> => () => Promise.resolve(as);

export const fromSql = <A>(config: MssqlConfig, query: string): Dsl<A> => () =>
  execute(config, { sql: query });

example.ts

import * as D from "./data";

type Ride = {
  id: number;
  f_users_id: number;
  description: string;
};

const getRides: D.Dsl<Ride> = D.fromSql<Ride>(
  config,
  "select id, f_users_id, description from Rides"
);

type Rider = {
  id: number;
  name: string;
};

const getRiders: D.Dsl<Rider> = D.fromCsv("./riders.csv");
const getRidesWithRiders = D.leftJoin(
  getRides,
  getRiders,
  (ride, rider) => ride.f_users_id == rider.id
);
const finalRides = D.map(getRidesWithRiders, (x) => ({
  id: x.a.id,
  description: x.a.description,
  rider: x.b ? x.b.name : undefined,
}));

const pipeline = D.toCsv(finalRides, "./finalRides.csv");

pipeline().then(console.log);