refactor: wip

Author: BioPhoton

packages/utils/src/lib/file-sink.ts

@@ -1,6 +1,6 @@
 import fs from 'node:fs';
 import path from 'node:path';
-import type { Decoder, Encoder } from './sink-source.type';
+import type { Decoder, Encoder, RecoverResult } from './sink-source.type';
 
 export type AppendOptions = {
   filePath: string;
@@ -162,3 +162,60 @@ export const StringCodec = {
   encode: (v: string) => v,
   decode: (v: string) => v,
 };
+
+export abstract class RecoverableEventSink<
+  Raw extends Record<string, unknown>,
+  Domain,
+> {
+  protected readonly sink: JsonlFile<Raw>;
+  private finalized = false;
+
+  constructor(sink: JsonlFile<Raw>) {
+    this.sink = sink;
+  }
+
+  open() {
+    this.sink.open();
+  }
+
+  write(event: Domain) {
+    this.sink.write(this.encode(event));
+  }
+
+  close() {
+    this.finalize();
+  }
+
+  recover(): RecoverResult<Domain> {
+    const { records, errors, partialTail } = this.sink.recover();
+    const out: Domain[] = [];
+    const errs = [...errors];
+
+    records.forEach((r, i) => {
+      try {
+        out.push(this.decode(r));
+      } catch (error) {
+        errs.push({
+          lineNo: i + 1,
+          line: JSON.stringify(r),
+          error: error as Error,
+        });
+      }
+    });
+
+    return { records: out, errors: errs, partialTail };
+  }
+
+  finalize() {
+    if (this.finalized) {
+      return;
+    }
+    this.finalized = true;
+    this.sink.close();
+    this.onFinalize();
+  }
+
+  protected abstract encode(domain: Domain): Raw;
+  protected abstract decode(raw: Raw): Domain;
+  protected abstract onFinalize(): void;
+}

packages/utils/src/lib/sink-source.type.ts

@@ -1,5 +1,3 @@
-import type { JsonlFile } from './file-sink.js';
-
 export type Encoder<T> = (value: T) => string;
 export type Decoder<T> = (line: string) => T;
 
@@ -44,63 +42,6 @@ export type RecoverResult<T = unknown> = {
   partialTail: string | null;
 };
 
-export abstract class RecoverableEventSink<
-  Raw extends Record<string, unknown>,
-  Domain,
-> {
-  protected readonly sink: JsonlFile<Raw>;
-  private finalized = false;
-
-  constructor(sink: JsonlFile<Raw>) {
-    this.sink = sink;
-  }
-
-  open() {
-    this.sink.open();
-  }
-
-  write(event: Domain) {
-    this.sink.write(this.encode(event));
-  }
-
-  close() {
-    this.finalize();
-  }
-
-  recover(): RecoverResult<Domain> {
-    const { records, errors, partialTail } = this.sink.recover();
-    const out: Domain[] = [];
-    const errs = [...errors];
-
-    records.forEach((r, i) => {
-      try {
-        out.push(this.decode(r));
-      } catch (error) {
-        errs.push({
-          lineNo: i + 1,
-          line: JSON.stringify(r),
-          error: error as Error,
-        });
-      }
-    });
-
-    return { records: out, errors: errs, partialTail };
-  }
-
-  finalize() {
-    if (this.finalized) {
-      return;
-    }
-    this.finalized = true;
-    this.sink.close();
-    this.onFinalize();
-  }
-
-  protected abstract encode(domain: Domain): Raw;
-  protected abstract decode(raw: Raw): Domain;
-  protected abstract onFinalize(): void;
-}
-
 export type RecoverOptions = {
   keepInvalid?: boolean;
 };