// Le problème : un writable plus lent qu'un readable fait exploser la mémoire sans backpressure
import { createReadStream, createWriteStream } from "node:fs";

const readStream = createReadStream("./huge-file.bin");
const writeStream = createWriteStream("./destination.bin");

// Mauvais : ignore le retour de write(), peut charger tout en mémoire si writeStream est lent
readStream.on("data", (chunk) => {
  writeStream.write(chunk); // pas de gestion du backpressure
});

// Bon : respecter le retour de write() — false signifie "le buffer interne est plein"
readStream.on("data", (chunk) => {
  const canContinue = writeStream.write(chunk);

  if (!canContinue) {
    readStream.pause(); // stoppe la lecture tant que le writable n'a pas absorbé son buffer
  }
});

writeStream.on("drain", () => {
  readStream.resume(); // le writable a vidé son buffer, on peut reprendre
});

// En pratique, TOUJOURS préférer .pipe() ou pipeline() qui gèrent ça automatiquement :
readStream.pipe(writeStream);

// Transform stream avec contrôle du highWaterMark (taille du buffer interne)
import { Transform } from "node:stream";

class JsonLinesParser extends Transform {
  constructor(options) {
    super({ ...options, objectMode: true, highWaterMark: 16 }); // buffer 16 objets max
    this.buffer = "";
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split("\n");
    this.buffer = lines.pop();

    for (const line of lines) {
      if (line.trim()) {
        try {
          this.push(JSON.parse(line));
        } catch (err) {
          return callback(err); // erreur propagée dans le pipeline
        }
      }
    }
    callback();
  }

  _flush(callback) {
    if (this.buffer.trim()) {
      try {
        this.push(JSON.parse(this.buffer));
      } catch (err) {
        return callback(err);
      }
    }
    callback();
  }
}

// Traiter un flux JSONL volumineux avec contrôle de concurrence limité
import { pipeline } from "node:stream/promises";
import { Writable } from "node:stream";

async function processInBatches(records, batchSize, processFn) {
  for (let i = 0; i < records.length; i += batchSize) {
    await Promise.all(records.slice(i, i + batchSize).map(processFn));
  }
}

const batchWriter = new Writable({
  objectMode: true,
  highWaterMark: 100,
  async write(record, encoding, callback) {
    try {
      await db.records.insert(record); // insertion asynchrone par record
      callback();
    } catch (err) {
      callback(err);
    }
  },
});

await pipeline(
  createReadStream("./data.jsonl"),
  new JsonLinesParser(),
  batchWriter
);

// Stream Web API (fetch) vs stream Node — interopérabilité via Readable.fromWeb
import { Readable } from "node:stream";
import { createWriteStream } from "node:fs";

const response = await fetch("https://example.com/large-dataset.csv");
const nodeStream = Readable.fromWeb(response.body); // conversion ReadableStream Web -> Node

await pipeline(nodeStream, createWriteStream("./dataset.csv"));