From f5e861c267291c232c70e8303a9cdc10860c6758 Mon Sep 17 00:00:00 2001
From: Ardis Lu <ardisglu@gmail.com>
Date: Sun, 26 Jan 2025 22:43:02 -0800
Subject: [PATCH] Add public status property to CSVWriter for tracking progress

---
 src/core.js       | 43 +++++++++++++++++++++++++++++++++++++++++--
 test/core.test.js | 46 ++++++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 85 insertions(+), 4 deletions(-)

diff --git a/src/core.js b/src/core.js
index 8809327..00c953b 100644
--- a/src/core.js
+++ b/src/core.js
@@ -1,6 +1,6 @@
 import { createReadStream } from 'node:fs';
 import { mkdir, open } from 'node:fs/promises';
-import { dirname, normalize } from 'node:path';
+import { dirname, basename, normalize } from 'node:path';
 import { Readable } from 'node:stream';
 import { fileURLToPath } from 'node:url';
 /** @import { PathLike } from 'node:fs'; */
@@ -327,6 +327,8 @@ export class CSVTransformer extends TransformStream {
  * @extends WritableStream
  */
 export class CSVWriter extends WritableStream {
+  #status;
+
   /**
    * @param {PathLike} path A `string`, `Buffer`, or `URL` representing a path to a local CSV file destination. If the file already
    * exists, its **data will be overwritten**.
@@ -334,6 +336,13 @@ export class CSVWriter extends WritableStream {
   constructor(path) {
     const fullPath = parsePathLike(path);
     const dir = dirname(fullPath);
+    const status = {
+      name: basename(fullPath),
+      start: performance.now(),
+      elapsed: 0,
+      rows: 0,
+      done: false
+    };
     let handle;
     super({
       async start() {
@@ -343,17 +352,47 @@ export class CSVWriter extends WritableStream {
       async write(chunk) {
         let data;
         try {
-          data = CSVWriter.arrayToCSVString(JSON.parse(chunk));
+          const parsed = JSON.parse(chunk);
+          data = CSVWriter.arrayToCSVString(parsed);
+          status.rows += parsed.length;
         }
         catch {
           data = chunk;
+          status.rows++;
         }
         await handle.write(data);
       },
       async close() {
         await handle.close();
+        status.elapsed = performance.now() - status.start;
+        status.done = true;
       }
     });
+
+    this.#status = status;
+  }
+
+  /**
+   * @typedef CSVWriterStatus The current status of the CSVWriter.
+   * @property {string} name The name of the output CSV file that this writer is writing to.
+   * @property {number} elapsed The number of milliseconds that have elapsed between this writer's creation and the output CSV file
+   * handle closing, or until now if the writer is still writing. Measured using `performance.now()`.
+   * @property {number} rows The estimated number of CSV rows that have been written. If the data passed to the writer is
+   * a string, the string is counted as one CSV row (i.e., raw string data is NOT parsed again).
+   * @property {boolean} done If `true`, the writer is finished writing to the output CSV file and the file handle is closed.
+   */
+
+  /** @type {CSVWriterStatus} */
+  get status() {
+    if (!this.#status.done) {
+      this.#status.elapsed = performance.now() - this.#status.start;
+    }
+    return {
+      name: this.#status.name,
+      elapsed: this.#status.elapsed,
+      rows: this.#status.rows,
+      done: this.#status.done
+    }
   }
 
   /**
diff --git a/test/core.test.js b/test/core.test.js
index b064509..9418c38 100644
--- a/test/core.test.js
+++ b/test/core.test.js
@@ -1,7 +1,8 @@
 import { suite, test } from 'node:test';
-import { deepStrictEqual, throws, rejects } from 'node:assert/strict';
+import { ok, deepStrictEqual, throws, rejects } from 'node:assert/strict';
 import { unlink } from 'node:fs/promises';
-import { normalize, resolve } from 'node:path';
+import { normalize, basename, resolve } from 'node:path';
+import { setTimeout } from 'node:timers/promises';
 import { pathToFileURL } from 'node:url';
 
 import { csvStreamEqualWritable, csvStreamNotEqualWritable, createCSVMockStream, createTempFile } from './utils.js';
@@ -478,4 +479,45 @@ suite('CSVWriter', { concurrency: true }, () => {
       await new CSVReader(temp).pipeTo(csvStreamEqualWritable(csv));
     });
   }
+});
+
+suite('CSVWriter status', { concurrency: true }, () => {
+  test('records file name', { concurrency: true }, async (t) => {
+    const temp = await createTempFile();
+    t.after(async () => await unlink(temp));
+    const writer = new CSVWriter(temp);
+    await createCSVMockStream([['']]).pipeTo(writer);
+    deepStrictEqual(writer.status.name, basename(temp));
+  });
+  test('records elapsed time', { concurrency: true }, async (t) => {
+    const temp = await createTempFile();
+    t.after(async () => await unlink(temp));
+    const writer = new CSVWriter(temp);
+    await createCSVMockStream([['']]).pipeTo(writer);
+    ok(writer.status.elapsed > 0);
+  });
+  test('elapsed time does not change after finishing writing', { concurrency: true }, async (t) => {
+    const temp = await createTempFile();
+    t.after(async () => await unlink(temp));
+    const writer = new CSVWriter(temp);
+    await createCSVMockStream([['']]).pipeTo(writer);
+    const { elapsed: elapsed1 } = writer.status;
+    await setTimeout(5);
+    const { elapsed: elapsed2 } = writer.status;
+    deepStrictEqual(elapsed1, elapsed2);
+  });
+  test('records rows written', { concurrency: true }, async (t) => {
+    const temp = await createTempFile();
+    t.after(async () => await unlink(temp));
+    const writer = new CSVWriter(temp);
+    await createCSVMockStream([[''], [''], ['']]).pipeTo(writer);
+    deepStrictEqual(writer.status.rows, 3);
+  });
+  test('records done status', { concurrency: true }, async (t) => {
+    const temp = await createTempFile();
+    t.after(async () => await unlink(temp));
+    const writer = new CSVWriter(temp);
+    await createCSVMockStream([['']]).pipeTo(writer);
+    ok(writer.status.done);
+  });
 });
\ No newline at end of file
-- 
2.47.3