import * as fsp from "node:fs/promises";
import type { Source, CallableSource } from "../../operations/from.ts";
import type { FileOptions } from "./types.ts";
import { HeadersKeys, type ExchangeHeaders } from "../../exchange.ts";
import { forEachLine, throwFileError } from "../shared/line-reader.ts";

/**
 * FileSourceAdapter implements the Source interface for reading files.
 * Reads the file once and emits its content as a string.
 * When chunked is true, emits one exchange per line.
 */
export class FileSourceAdapter implements Source<string> {
  readonly adapterId = "routecraft.adapter.file";

  constructor(private readonly options: FileOptions) {}

  /**
   * Source implementation: subscribe to file content.
   * Reads the file once (or line-by-line when chunked).
   */
  subscribe: CallableSource<string> = async (
    _context,
    handler,
    abortController,
    onReady,
  ) => {
    // Check if already aborted
    if (abortController.signal.aborted) return;

    const {
      path: filePath,
      encoding = "utf-8",
      chunked = false,
    } = this.options;

    if (typeof filePath !== "string") {
      throw new Error(
        "file adapter: path must be a string for source mode (dynamic paths are only supported for destinations)",
      );
    }

    if (chunked) {
      try {
        await forEachLine(
          filePath,
          encoding,
          abortController.signal,
          async (line, lineNumber) => {
            const headers: ExchangeHeaders = {
              [HeadersKeys.FILE_LINE]: lineNumber,
              [HeadersKeys.FILE_PATH]: filePath,
            } as ExchangeHeaders;
            await handler(line, headers);
          },
        );
      } catch (err) {
        if (abortController.signal.aborted) return;
        throwFileError("file", filePath, err);
      }
    } else {
      const content = await fsp
        .readFile(filePath, { encoding })
        .catch((err) => throwFileError("file", filePath, err));

      // Check if aborted before emitting
      if (abortController.signal.aborted) return;

      // Emit the content
      await handler(content);
    }

    // Signal that source is ready
    if (onReady) onReady();
  };
}
