stream-data.ts

import { FlatfileEvent, FlatfileListener } from "@flatfile/listener";
import { Flatfile } from "@flatfile/api";

/**
 * Reaction plugin for Flatfile.
 *
 * @param options
 * @param callback
 */
export function streamData<T extends StreamActionOptions>(
  { format, includeInvalid }: T,
  callback: (data: Array<SwitchResponse<T>>, e: FlatfileEvent) => void
) {
  return (listener: FlatfileListener) => {
    listener.on("commit:created", async (e) => {
      const data: Flatfile.RecordsResponseData = await e.data;
      const records = data.records?.filter((r) => includeInvalid || r.valid);
      if (format === "map") {
        const mapped = records?.map<MappedRecord>((record) => {
          const mr = Object.keys(record.values).reduce((acc, key: string) => {
            acc[key] = record.values[key].value;
            return acc;
          }, {} as Record<string, any>);
          return { __recordId: record.id, ...mr };
        });
        if (mapped) {
          await callback(mapped as SwitchResponse<T>[], e);
        }
      }
      if ((format === "full" || !format) && records) {
        await callback(records as unknown as SwitchResponse<T>[], e);
      }
    });
  };
}

type SwitchResponse<T extends StreamActionOptions> = T["format"] extends "map"
  ? MappedRecord
  : T["format"] extends "full"
  ? Flatfile.RecordsWithLinks
  : T["format"] extends undefined
  ? Record<string, any>
  : never;

type MappedRecord = {
  __recordId: string;
  [key: string]: any;
};

type StreamActionOptions = {
  format: "map" | "full";
  includeInvalid?: boolean;
};