feat: improve memory usage of record import (#218)

src/record/import/index.ts

@@ -1,14 +1,14 @@
 import type { RestAPIClientOptions } from "../../kintone/client";
 import { buildRestAPIClient } from "../../kintone/client";
 import type { SupportedImportEncoding } from "../../utils/file";
-import { readFile } from "../../utils/file";
-import { parseRecords } from "./parsers";
+import { extractFileFormat, openFsStreamWithEncode } from "../../utils/file";
 import { addRecords } from "./usecases/add";
 import { upsertRecords } from "./usecases/upsert";
 import { createSchema } from "./schema";
 import { noop as defaultTransformer } from "./schema/transformers/noop";
 import { userSelected } from "./schema/transformers/userSelected";
 import { logger } from "../../utils/log";
+import { LocalRecordRepositoryFromStream } from "./repositories/localRecordRepositoryFromStream";
 
 export type Options = {
   app: string;
@@ -42,24 +42,33 @@ export const run: (
         ? userSelected(fields, fieldsJson, updateKey)
         : defaultTransformer()
     );
-    const { content, format } = await readFile(filePath, encoding);
-    const records = await parseRecords({
-      source: content,
+    const format = extractFileFormat(filePath);
+    const localRecordRepository = new LocalRecordRepositoryFromStream(
+      () => openFsStreamWithEncode(filePath, encoding),
       format,
-      schema,
-    });
-    if (records.length === 0) {
+      schema
+    );
+
+    if ((await localRecordRepository.length()) === 0) {
       logger.warn("The input file does not have any records");
       return;
     }
+
     const skipMissingFields = !fields;
     if (updateKey) {
-      await upsertRecords(apiClient, app, records, schema, updateKey, {
-        attachmentsDir,
-        skipMissingFields,
-      });
+      await upsertRecords(
+        apiClient,
+        app,
+        localRecordRepository,
+        schema,
+        updateKey,
+        {
+          attachmentsDir,
+          skipMissingFields,
+        }
+      );
     } else {
-      await addRecords(apiClient, app, records, schema, {
+      await addRecords(apiClient, app, localRecordRepository, schema, {
         attachmentsDir,
         skipMissingFields,
       });

src/record/import/repositories/error.ts

@@ -0,0 +1,30 @@
+import { ParserError } from "./parsers/error";
+
+export class RepositoryError extends Error {
+  private readonly cause: unknown;
+
+  constructor(cause: unknown) {
+    const message =
+      "An error occurred while loading records from the data source";
+    super(message);
+
+    this.name = "RepositoryError";
+    this.message = message;
+    this.cause = cause;
+
+    // https://github.com/Microsoft/TypeScript/wiki/Breaking-Changes#extending-built-ins-like-error-array-and-map-may-no-longer-work
+    // Set the prototype explicitly.
+    Object.setPrototypeOf(this, RepositoryError.prototype);
+  }
+
+  toString(): string {
+    let errorMessage = "";
+    errorMessage += this.message + "\n";
+    if (this.cause instanceof ParserError) {
+      errorMessage += this.cause.toString();
+    } else {
+      errorMessage += this.cause + "\n";
+    }
+    return errorMessage;
+  }
+}

src/record/import/repositories/localRecordRepositoryFromStream.ts

@@ -0,0 +1,31 @@
+import type { LocalRecordRepository } from "../usecases/interface";
+import type { RecordSchema } from "../types/schema";
+import { RepositoryError } from "./error";
+import { countRecordsFromCsv, csvReader } from "./parsers/parseCsv";
+import type { LocalRecord } from "../types/record";
+
+export class LocalRecordRepositoryFromStream implements LocalRecordRepository {
+  readonly format: string;
+  readonly length: () => Promise<number>;
+
+  readonly reader: () => AsyncGenerator<LocalRecord, void, undefined>;
+
+  constructor(
+    source: () => NodeJS.ReadableStream,
+    format: string,
+    schema: RecordSchema
+  ) {
+    this.format = format;
+    this.length = () => countRecordsFromCsv(source());
+
+    switch (format) {
+      case "csv":
+        this.reader = () => csvReader(source, schema);
+        break;
+      default:
+        throw new RepositoryError(
+          `Unexpected file type: ${format} is unacceptable.`
+        );
+    }
+  }
+}

src/record/import/repositories/localRecordRepositoryMock.ts

@@ -0,0 +1,22 @@
+import type { LocalRecordRepository } from "../usecases/interface";
+import type { LocalRecord } from "../types/record";
+
+export class LocalRecordRepositoryMock implements LocalRecordRepository {
+  readonly format: string;
+  readonly length: () => Promise<number>;
+
+  readonly reader: () => AsyncGenerator<LocalRecord, void, undefined>;
+
+  constructor(source: LocalRecord[], format: string) {
+    this.format = format;
+    this.length = async () => source.length;
+    this.reader = () => asyncGeneratorFromStream(source);
+  }
+}
+
+// eslint-disable-next-line func-style
+async function* asyncGeneratorFromStream(
+  source: LocalRecord[]
+): AsyncGenerator<LocalRecord, void, undefined> {
+  yield* source;
+}

src/record/import/parsers/parseCsv/__tests__/fieldValue.test.ts → src/record/import/repositories/parsers/parseCsv/__tests__/fieldValue.test.ts

@@ -1,5 +1,5 @@
-import type * as Fields from "../../../types/field";
-import type { FieldSchema } from "../../../types/schema";
+import type * as Fields from "../../../../types/field";
+import type { FieldSchema } from "../../../../types/schema";
 
 import { convertFieldValue } from "../fieldValue";
 

src/record/import/repositories/parsers/parseCsv/__tests__/fixtures/emptyCsv/expected.ts

@@ -0,0 +1,3 @@
+import type { LocalRecord } from "../../../../../../types/record";
+
+export const expected: LocalRecord[] = [];

src/record/import/parsers/parseCsv/__tests__/fixtures/withNoRecord/schema.ts → src/record/import/repositories/parsers/parseCsv/__tests__/fixtures/emptyCsv/schema.ts

@@ -1,4 +1,4 @@
-import type { RecordSchema } from "../../../../../types/schema";
+import type { RecordSchema } from "../../../../../../types/schema";
 
 export const schema: RecordSchema = {
   fields: [

src/record/import/parsers/parseCsv/__tests__/fixtures/withLf/expected.ts → src/record/import/repositories/parsers/parseCsv/__tests__/fixtures/withCr/expected.ts

@@ -1,6 +1,6 @@
-import type { KintoneRecord } from "../../../../../types/record";
+import type { LocalRecord } from "../../../../../../types/record";
 
-export const expected: KintoneRecord[] = [
+export const expected: LocalRecord[] = [
   {
     data: {
       singleLineText: {
@@ -13,7 +13,9 @@ export const expected: KintoneRecord[] = [
         value: ['"sample3"', "sample4,sample5"],
       },
     },
-    metadata: { format: { type: "csv", firstRowIndex: 1, lastRowIndex: 1 } },
+    metadata: {
+      format: { type: "csv", firstRowIndex: 1, lastRowIndex: 1 },
+    },
   },
   {
     data: {
@@ -27,6 +29,8 @@ export const expected: KintoneRecord[] = [
         value: ['"sample4"', "sample5,sample6"],
       },
     },
-    metadata: { format: { type: "csv", firstRowIndex: 2, lastRowIndex: 2 } },
+    metadata: {
+      format: { type: "csv", firstRowIndex: 2, lastRowIndex: 2 },
+    },
   },
 ];

src/record/import/parsers/parseCsv/__tests__/fixtures/withCr/schema.ts → src/record/import/repositories/parsers/parseCsv/__tests__/fixtures/withCr/schema.ts

@@ -1,4 +1,4 @@
-import type { RecordSchema } from "../../../../../types/schema";
+import type { RecordSchema } from "../../../../../../types/schema";
 
 export const schema: RecordSchema = {
   fields: [

src/record/import/parsers/parseCsv/__tests__/fixtures/withCrLf/expected.ts → src/record/import/repositories/parsers/parseCsv/__tests__/fixtures/withCrLf/expected.ts

@@ -1,6 +1,6 @@
-import type { KintoneRecord } from "../../../../../types/record";
+import type { LocalRecord } from "../../../../../../types/record";
 
-export const expected: KintoneRecord[] = [
+export const expected: LocalRecord[] = [
   {
     data: {
       singleLineText: {
@@ -13,7 +13,9 @@ export const expected: KintoneRecord[] = [
         value: ['"sample3"', "sample4,sample5"],
       },
     },
-    metadata: { format: { type: "csv", firstRowIndex: 1, lastRowIndex: 1 } },
+    metadata: {
+      format: { type: "csv", firstRowIndex: 1, lastRowIndex: 1 },
+    },
   },
   {
     data: {
@@ -27,6 +29,8 @@ export const expected: KintoneRecord[] = [
         value: ['"sample4"', "sample5,sample6"],
       },
     },
-    metadata: { format: { type: "csv", firstRowIndex: 2, lastRowIndex: 2 } },
+    metadata: {
+      format: { type: "csv", firstRowIndex: 2, lastRowIndex: 2 },
+    },
   },
 ];

src/record/import/parsers/parseCsv/__tests__/fixtures/withLf/schema.ts → src/record/import/repositories/parsers/parseCsv/__tests__/fixtures/withCrLf/schema.ts

@@ -1,4 +1,4 @@
-import type { RecordSchema } from "../../../../../types/schema";
+import type { RecordSchema } from "../../../../../../types/schema";
 
 export const schema: RecordSchema = {
   fields: [