/* eslint-disable no-console */
import fs from "node:fs";
import { Sink, SinkConfig } from "./types.js";
import path from "node:path";

/** File sink: append NDJSON lines into a single file. */
export class FileSink implements Sink {
    private buf: string[] = [];
    private flushEvery: number;
    private outPath: string;
    private stream!: fs.WriteStream;

    constructor(cfg: SinkConfig) {
        if (!cfg.outPath) throw new Error("FileSink requires outPath");
        this.outPath = cfg.outPath;
        this.flushEvery = Math.max(1, cfg.flushEvery ?? 100);
    }

    async init(): Promise<void> {
        await fs.promises.mkdir(path.dirname(this.outPath), { recursive: true });
        this.stream = fs.createWriteStream(this.outPath, { flags: "a" });
        await new Promise<void>((resolve) => this.stream.once("open", () => resolve()));
    }

    async write(line: string): Promise<void> {
        this.buf.push(line);
        if (this.buf.length >= this.flushEvery) {
            const chunk = this.buf.join("\n") + "\n";
            this.buf.length = 0;
            if (!this.stream.write(chunk)) {
                await new Promise<void>((resolve, reject) => {
                    this.stream.once("drain", () => resolve());
                    this.stream.once("error", (e) => reject(e));
                });
            }
        }
    }

    async flush(): Promise<void> {
        if (this.buf.length === 0) return;
        const chunk = this.buf.join("\n") + "\n";
        this.buf.length = 0;
        if (!this.stream.write(chunk)) {
            await new Promise<void>((resolve, reject) => {
                this.stream.once("drain", () => resolve());
                this.stream.once("error", (e) => reject(e));
            });
        }
    }

    async close(): Promise<void> {
        await this.flush();
        await new Promise<void>((resolve, reject) => {
            this.stream.end(() => resolve());
            this.stream.once("error", (e) => reject(e));
        });
    }
}