propagate error between steams

src/record/import/repositories/parsers/parseCsv/index.ts

@@ -1,8 +1,5 @@
-import type { CsvRow } from "../../../../../kintone/types";
-import type { LocalRecord } from "../../../types/record";
 import type { RecordSchema } from "../../../types/schema";
 
-// import csvParse from "csv-parse/lib/sync";
 import csvParse from "csv-parse";
 
 import { convertRecord, recordReader } from "./record";
@@ -17,13 +14,17 @@ export async function* csvReader<T extends NodeJS.ReadableStream>(
   schema: RecordSchema
 ): ReturnType<LocalRecordRepository["reader"]> {
   try {
+    const sourceStream = source();
     const csvStream = source().pipe(
       csvParse({
         columns: true,
         skip_empty_lines: true,
         delimiter: SEPARATOR,
       })
     );
+    sourceStream.on("error", (e) => {
+      csvStream.destroy(e);
+    });
 
     for await (const recordRows of recordReader(csvStream)) {
       yield convertRecord(recordRows, schema);
@@ -45,14 +46,16 @@ export const countRecordsFromCsv = async <T extends NodeJS.ReadableStream>(
         delimiter: SEPARATOR,
       })
     );
+    source.on("error", (e) => {
+      csvStream.destroy(e);
+    });
 
     let count = 0;
     for await (const recordRows of recordReader(csvStream)) {
       count++;
     }
     return count;
   } catch (e) {
-    console.error(e);
     throw new ParserError(e);
   }
 };

src/record/import/utils/file.ts

@@ -1,24 +1,34 @@
 import fs from "fs";
 import path from "path";
 import iconv from "iconv-lite";
+import { Transform } from "stream";
 
 export type SupportedImportEncoding = "utf8" | "sjis";
 
 export const openFsStreamWithEncode: (
   filePath: string,
   encoding?: SupportedImportEncoding
-) => { stream: NodeJS.ReadWriteStream; format: string } = (
+) => { stream: NodeJS.ReadableStream; format: string } = (
   filePath,
   encoding = "utf8"
 ) => {
   const format = extractFileFormat(filePath);
   if (format === "json" && encoding !== "utf8") {
     throw new Error("source file is JSON and JSON MUST be encoded with UTF-8");
   }
-  const stream = fs
-    .createReadStream(filePath)
-    .pipe(iconv.decodeStream(encoding));
-  return { stream, format };
+  const stream = fs.createReadStream(filePath);
+
+  const decodedStream = stream.pipe(
+    Transform.from(iconv.decodeStream(encoding))
+  );
+  stream.on("error", (e) => {
+    decodedStream.destroy(e);
+  });
+
+  decodedStream.on("error", () => {
+    /* noop. but this is necessary :) */
+  });
+  return { stream: decodedStream, format };
 };
 
 export const readFile: (

src/utils/log.ts

@@ -2,6 +2,7 @@ import { AddRecordsError } from "../record/import/usecases/add/error";
 import { UpsertRecordsError } from "../record/import/usecases/upsert/error";
 import { ParserError } from "../record/import/repositories/parsers/error";
 import chalk from "chalk";
+import { RepositoryError } from "../record/import/repositories/error";
 
 const currentISOString = () => new Date().toISOString();
 
@@ -49,6 +50,8 @@ const parseErrorMessage = (error: unknown): string => {
       return error.toString();
     } else if (error instanceof UpsertRecordsError) {
       return error.toString();
+    } else if (error instanceof RepositoryError) {
+      return error.toString();
     } else if (error instanceof ParserError) {
       return error.toString();
     }