/**
 * Encode and decode newline-delimited JSON streams in Effect channels.
 *
 * NDJSON represents a stream as one complete JSON value per line. This module
 * keeps that framing explicit for streaming boundaries: byte helpers handle
 * UTF-8 transport chunks, string helpers handle already-decoded text, and
 * schema helpers validate or transform each record at the edge of the pipeline.
 *
 * **Mental model**
 *
 * - Encoders receive non-empty chunks of values and emit one NDJSON chunk ending
 *   with a newline
 * - Decoders accumulate partial lines across input chunks and emit parsed JSON
 *   records as soon as complete lines are available
 * - Schema helpers compose channel encoding and decoding with the plain NDJSON
 *   framing helpers
 *
 * **Common tasks**
 *
 * - Encode values to UTF-8 bytes: {@link encode}
 * - Encode values to strings: {@link encodeString}
 * - Decode UTF-8 bytes or strings: {@link decode}, {@link decodeString}
 * - Add schema validation at the boundary: {@link encodeSchema}, {@link decodeSchema}
 *
 * **Gotchas**
 *
 * - Records must be valid JSON values, not JavaScript expressions
 * - Empty lines are ignored only when `ignoreEmptyLines` is enabled
 * - Encoders append a trailing newline, so downstream consumers should treat it
 *   as record framing rather than as an extra record
 *
 * @since 4.0.0
 */
import * as Arr from "../../Array.ts"
import * as Channel from "../../Channel.ts"
import * as ChannelSchema from "../../ChannelSchema.ts"
import * as Data from "../../Data.ts"
import * as Effect from "../../Effect.ts"
import { dual, identity } from "../../Function.ts"
import type * as Schema from "../../Schema.ts"

const NdjsonErrorTypeId = "~effect/encoding/Ndjson/NdjsonError"

const encoder = new TextEncoder()

/**
 * Error raised when NDJSON encoding or decoding fails.
 *
 * **Details**
 *
 * The `kind` field identifies whether the failure happened while packing or
 * unpacking, and `cause` preserves the original error.
 *
 * @category errors
 * @since 4.0.0
 */
export class NdjsonError extends Data.TaggedError("NdjsonError")<{
  readonly kind: "Pack" | "Unpack"
  readonly cause: unknown
}> {
  /**
   * Marks this value as an NDJSON encoding or decoding error for runtime guards.
   *
   * @since 4.0.0
   */
  readonly [NdjsonErrorTypeId] = NdjsonErrorTypeId

  /**
   * Uses the failed NDJSON operation as the public message.
   *
   * @since 4.0.0
   */
  override get message() {
    return this.kind
  }
}

/**
 * Creates a channel that encodes chunks of values as NDJSON strings.
 *
 * **Details**
 *
 * Each input item is `JSON.stringify`-encoded, separated by newlines, and the
 * output chunk ends with a trailing newline.
 *
 * @category constructors
 * @since 4.0.0
 */
export const encodeString = <IE = never, Done = unknown>(): Channel.Channel<
  Arr.NonEmptyReadonlyArray<string>,
  IE | NdjsonError,
  Done,
  Arr.NonEmptyReadonlyArray<unknown>,
  IE,
  Done
> =>
  Channel.fromTransform((upstream, _scope) =>
    Effect.succeed(Effect.flatMap(upstream, (input) => {
      try {
        return Effect.succeed(Arr.of(input.map((item) => JSON.stringify(item)).join("\n") + "\n"))
      } catch (cause) {
        return Effect.fail(new NdjsonError({ kind: "Pack", cause }))
      }
    }))
  )

/**
 * Creates a channel that encodes chunks of values as UTF-8 NDJSON bytes.
 *
 * @category constructors
 * @since 4.0.0
 */
export const encode = <IE = never, Done = unknown>(): Channel.Channel<
  Arr.NonEmptyReadonlyArray<Uint8Array>,
  IE | NdjsonError,
  Done,
  Arr.NonEmptyReadonlyArray<unknown>,
  IE,
  Done
> => Channel.map(encodeString(), Arr.map((_) => encoder.encode(_)))

/**
 * Creates an NDJSON byte encoder channel for values of a schema.
 *
 * **Details**
 *
 * Values are first encoded with the schema and then written as UTF-8
 * newline-delimited JSON.
 *
 * @category constructors
 * @since 4.0.0
 */
export const encodeSchema = <S extends Schema.Top>(
  schema: S
) =>
<IE = never, Done = unknown>(): Channel.Channel<
  Arr.NonEmptyReadonlyArray<Uint8Array>,
  NdjsonError | Schema.SchemaError | IE,
  Done,
  Arr.NonEmptyReadonlyArray<S["Type"]>,
  IE,
  Done,
  S["EncodingServices"]
> => Channel.pipeTo(ChannelSchema.encode(schema)(), encode())

/**
 * Creates an NDJSON string encoder channel for values of a schema.
 *
 * **Details**
 *
 * Values are first encoded with the schema and then written as newline-delimited
 * JSON strings.
 *
 * @category constructors
 * @since 4.0.0
 */
export const encodeSchemaString = <S extends Schema.Top>(
  schema: S
) =>
<IE = never, Done = unknown>(): Channel.Channel<
  Arr.NonEmptyReadonlyArray<string>,
  NdjsonError | Schema.SchemaError | IE,
  Done,
  Arr.NonEmptyReadonlyArray<S["Type"]>,
  IE,
  Done,
  S["EncodingServices"]
> => Channel.pipeTo(ChannelSchema.encode(schema)(), encodeString())

/**
 * Creates a channel that parses NDJSON string chunks into values.
 *
 * **Details**
 *
 * Lines may span input chunks.
 *
 * **Gotchas**
 *
 * Set `ignoreEmptyLines` to skip blank lines before calling `JSON.parse`;
 * otherwise blank lines are parsed and fail as invalid JSON.
 *
 * @category constructors
 * @since 4.0.0
 */
export const decodeString = <IE = never, Done = unknown>(options?: {
  readonly ignoreEmptyLines?: boolean | undefined
}): Channel.Channel<
  Arr.NonEmptyReadonlyArray<unknown>,
  IE | NdjsonError,
  Done,
  Arr.NonEmptyReadonlyArray<string>,
  IE,
  Done
> => {
  const lines = Channel.splitLines<IE, Done>().pipe(
    options?.ignoreEmptyLines === true ?
      Channel.filterArray((line) => line.length > 0) :
      identity
  )
  return Channel.mapEffect(lines, (chunk) => {
    try {
      return Effect.succeed(Arr.map(chunk, (line) => JSON.parse(line)))
    } catch (cause) {
      return Effect.fail(new NdjsonError({ kind: "Unpack", cause }))
    }
  })
}

/**
 * Creates a channel that decodes UTF-8 byte chunks and parses them as NDJSON.
 *
 * **Details**
 *
 * Lines may span input chunks, and `ignoreEmptyLines` controls whether blank
 * lines are skipped before JSON parsing.
 *
 * @category constructors
 * @since 4.0.0
 */
export const decode = <IE = never, Done = unknown>(options?: {
  readonly ignoreEmptyLines?: boolean | undefined
}): Channel.Channel<
  Arr.NonEmptyReadonlyArray<unknown>,
  IE | NdjsonError,
  Done,
  Arr.NonEmptyReadonlyArray<Uint8Array>,
  IE,
  Done
> => {
  return Channel.pipeTo(Channel.decodeText(), decodeString(options))
}

/**
 * Creates an NDJSON byte decoder channel for values of a schema.
 *
 * **Details**
 *
 * The channel decodes UTF-8 bytes, parses each NDJSON line, and then decodes
 * each parsed value with the schema.
 *
 * @category constructors
 * @since 4.0.0
 */
export const decodeSchema = <S extends Schema.Top>(
  schema: S
) =>
<IE = never, Done = unknown>(options?: {
  readonly ignoreEmptyLines?: boolean | undefined
}): Channel.Channel<
  Arr.NonEmptyReadonlyArray<S["Type"]>,
  Schema.SchemaError | NdjsonError | IE,
  Done,
  Arr.NonEmptyReadonlyArray<Uint8Array>,
  IE,
  Done,
  S["DecodingServices"]
> => Channel.pipeTo(decode(options), ChannelSchema.decodeUnknown(schema)())

/**
 * Creates an NDJSON string decoder channel for values of a schema.
 *
 * **Details**
 *
 * The channel parses each line as JSON and then decodes each parsed value with
 * the schema.
 *
 * @category constructors
 * @since 4.0.0
 */
export const decodeSchemaString = <S extends Schema.Top>(
  schema: S
) =>
<IE = never, Done = unknown>(options?: {
  readonly ignoreEmptyLines?: boolean | undefined
}): Channel.Channel<
  Arr.NonEmptyReadonlyArray<S["Type"]>,
  Schema.SchemaError | NdjsonError | IE,
  Done,
  Arr.NonEmptyReadonlyArray<string>,
  IE,
  Done,
  S["DecodingServices"]
> => Channel.pipeTo(decodeString(options), ChannelSchema.decodeUnknown(schema)())

/**
 * Wraps a bidirectional byte channel with NDJSON encoding and decoding.
 *
 * **Details**
 *
 * Outgoing values are written as UTF-8 NDJSON bytes, and incoming bytes are
 * parsed as NDJSON values.
 *
 * @category combinators
 * @since 4.0.0
 */
export const duplex: {
  /**
   * Wraps a bidirectional byte channel with NDJSON encoding and decoding.
   *
   * **Details**
   *
   * Outgoing values are written as UTF-8 NDJSON bytes, and incoming bytes are
   * parsed as NDJSON values.
   *
   * @category combinators
   * @since 4.0.0
   */
  (
    options?: {
      readonly ignoreEmptyLines?: boolean | undefined
    }
  ): <R, IE, OE, OutDone, InDone>(
    self: Channel.Channel<
      Arr.NonEmptyReadonlyArray<Uint8Array>,
      OE,
      OutDone,
      Arr.NonEmptyReadonlyArray<Uint8Array>,
      IE | NdjsonError,
      InDone,
      R
    >
  ) => Channel.Channel<
    Arr.NonEmptyReadonlyArray<unknown>,
    NdjsonError | OE,
    OutDone,
    Arr.NonEmptyReadonlyArray<unknown>,
    IE,
    InDone,
    R
  >
  /**
   * Wraps a bidirectional byte channel with NDJSON encoding and decoding.
   *
   * **Details**
   *
   * Outgoing values are written as UTF-8 NDJSON bytes, and incoming bytes are
   * parsed as NDJSON values.
   *
   * @category combinators
   * @since 4.0.0
   */
  <R, IE, OE, OutDone, InDone>(
    self: Channel.Channel<
      Arr.NonEmptyReadonlyArray<Uint8Array>,
      OE,
      OutDone,
      Arr.NonEmptyReadonlyArray<Uint8Array>,
      IE | NdjsonError,
      InDone,
      R
    >,
    options?: {
      readonly ignoreEmptyLines?: boolean | undefined
    }
  ): Channel.Channel<
    Arr.NonEmptyReadonlyArray<unknown>,
    NdjsonError | OE,
    OutDone,
    Arr.NonEmptyReadonlyArray<unknown>,
    IE,
    InDone,
    R
  >
} = dual((args) => Channel.isChannel(args[0]), <R, IE, OE, OutDone, InDone>(
  self: Channel.Channel<
    Arr.NonEmptyReadonlyArray<Uint8Array>,
    OE,
    OutDone,
    Arr.NonEmptyReadonlyArray<Uint8Array>,
    IE | NdjsonError,
    InDone,
    R
  >,
  options?: {
    readonly ignoreEmptyLines?: boolean | undefined
  }
): Channel.Channel<
  Arr.NonEmptyReadonlyArray<unknown>,
  NdjsonError | OE,
  OutDone,
  Arr.NonEmptyReadonlyArray<unknown>,
  IE,
  InDone,
  R
> =>
  Channel.pipeTo(
    Channel.pipeTo(encode(), self),
    decode(options)
  ))

/**
 * Wraps a bidirectional string channel with NDJSON encoding and decoding.
 *
 * **Details**
 *
 * Outgoing values are written as NDJSON strings, and incoming strings are parsed
 * as NDJSON values.
 *
 * @category combinators
 * @since 4.0.0
 */
export const duplexString: {
  /**
   * Wraps a bidirectional string channel with NDJSON encoding and decoding.
   *
   * **Details**
   *
   * Outgoing values are written as NDJSON strings, and incoming strings are parsed
   * as NDJSON values.
   *
   * @category combinators
   * @since 4.0.0
   */
  (
    options?: {
      readonly ignoreEmptyLines?: boolean | undefined
    }
  ): <R, IE, OE, OutDone, InDone>(
    self: Channel.Channel<
      Arr.NonEmptyReadonlyArray<string>,
      OE,
      OutDone,
      Arr.NonEmptyReadonlyArray<string>,
      IE | NdjsonError,
      InDone,
      R
    >
  ) => Channel.Channel<
    Arr.NonEmptyReadonlyArray<unknown>,
    NdjsonError | OE,
    OutDone,
    Arr.NonEmptyReadonlyArray<unknown>,
    IE,
    InDone,
    R
  >
  /**
   * Wraps a bidirectional string channel with NDJSON encoding and decoding.
   *
   * **Details**
   *
   * Outgoing values are written as NDJSON strings, and incoming strings are parsed
   * as NDJSON values.
   *
   * @category combinators
   * @since 4.0.0
   */
  <R, IE, OE, OutDone, InDone>(
    self: Channel.Channel<
      Arr.NonEmptyReadonlyArray<string>,
      OE,
      OutDone,
      Arr.NonEmptyReadonlyArray<string>,
      IE | NdjsonError,
      InDone,
      R
    >,
    options?: {
      readonly ignoreEmptyLines?: boolean | undefined
    }
  ): Channel.Channel<
    Arr.NonEmptyReadonlyArray<unknown>,
    NdjsonError | OE,
    OutDone,
    Arr.NonEmptyReadonlyArray<unknown>,
    IE,
    InDone,
    R
  >
} = dual((args) => Channel.isChannel(args[0]), <R, IE, OE, OutDone, InDone>(
  self: Channel.Channel<
    Arr.NonEmptyReadonlyArray<string>,
    OE,
    OutDone,
    Arr.NonEmptyReadonlyArray<string>,
    IE | NdjsonError,
    InDone,
    R
  >,
  options?: {
    readonly ignoreEmptyLines?: boolean | undefined
  }
): Channel.Channel<
  Arr.NonEmptyReadonlyArray<unknown>,
  NdjsonError | OE,
  OutDone,
  Arr.NonEmptyReadonlyArray<unknown>,
  IE,
  InDone,
  R
> =>
  Channel.pipeTo(
    Channel.pipeTo(encodeString(), self),
    decodeString(options)
  ))

/**
 * Wraps a bidirectional byte channel with schema-aware NDJSON encoding and
 * decoding.
 *
 * **Details**
 *
 * Values sent to the wrapped channel are encoded with `inputSchema`; bytes
 * received from it are parsed as NDJSON and decoded with `outputSchema`.
 *
 * @category combinators
 * @since 4.0.0
 */
export const duplexSchema: {
  /**
   * Wraps a bidirectional byte channel with schema-aware NDJSON encoding and
   * decoding.
   *
   * **Details**
   *
   * Values sent to the wrapped channel are encoded with `inputSchema`; bytes
   * received from it are parsed as NDJSON and decoded with `outputSchema`.
   *
   * @category combinators
   * @since 4.0.0
   */
  <In extends Schema.Top, Out extends Schema.Top>(
    options: {
      readonly inputSchema: In
      readonly outputSchema: Out
      readonly ignoreEmptyLines?: boolean | undefined
    }
  ): <OutErr, OutDone, InErr, InDone, R>(
    self: Channel.Channel<
      Arr.NonEmptyReadonlyArray<Uint8Array>,
      OutErr,
      OutDone,
      Arr.NonEmptyReadonlyArray<Uint8Array>,
      NdjsonError | Schema.SchemaError | InErr,
      InDone,
      R
    >
  ) => Channel.Channel<
    Arr.NonEmptyReadonlyArray<Out["Type"]>,
    NdjsonError | Schema.SchemaError | OutErr,
    OutDone,
    Arr.NonEmptyReadonlyArray<In["Type"]>,
    InErr,
    InDone,
    R | In["EncodingServices"] | Out["DecodingServices"]
  >
  /**
   * Wraps a bidirectional byte channel with schema-aware NDJSON encoding and
   * decoding.
   *
   * **Details**
   *
   * Values sent to the wrapped channel are encoded with `inputSchema`; bytes
   * received from it are parsed as NDJSON and decoded with `outputSchema`.
   *
   * @category combinators
   * @since 4.0.0
   */
  <Out extends Schema.Top, In extends Schema.Top, OutErr, OutDone, InErr, InDone, R>(
    self: Channel.Channel<
      Arr.NonEmptyReadonlyArray<Uint8Array>,
      OutErr,
      OutDone,
      Arr.NonEmptyReadonlyArray<Uint8Array>,
      NdjsonError | Schema.SchemaError | InErr,
      InDone,
      R
    >,
    options: {
      readonly inputSchema: In
      readonly outputSchema: Out
      readonly ignoreEmptyLines?: boolean | undefined
    }
  ): Channel.Channel<
    Arr.NonEmptyReadonlyArray<Out["Type"]>,
    NdjsonError | Schema.SchemaError | OutErr,
    OutDone,
    Arr.NonEmptyReadonlyArray<In["Type"]>,
    InErr,
    InDone,
    R | In["EncodingServices"] | Out["DecodingServices"]
  >
} = dual(2, <Out extends Schema.Top, In extends Schema.Top, OutErr, OutDone, InErr, InDone, R>(
  self: Channel.Channel<
    Arr.NonEmptyReadonlyArray<Uint8Array>,
    OutErr,
    OutDone,
    Arr.NonEmptyReadonlyArray<Uint8Array>,
    NdjsonError | Schema.SchemaError | InErr,
    InDone,
    R
  >,
  options: {
    readonly inputSchema: In
    readonly outputSchema: Out
    readonly ignoreEmptyLines?: boolean | undefined
  }
): Channel.Channel<
  Arr.NonEmptyReadonlyArray<Out["Type"]>,
  NdjsonError | Schema.SchemaError | OutErr,
  OutDone,
  Arr.NonEmptyReadonlyArray<In["Type"]>,
  InErr,
  InDone,
  R | In["EncodingServices"] | Out["DecodingServices"]
> => ChannelSchema.duplexUnknown(duplex(self, options), options))

/**
 * Wraps a bidirectional string channel with schema-aware NDJSON encoding and
 * decoding.
 *
 * **Details**
 *
 * Values sent to the wrapped channel are encoded with `inputSchema`; strings
 * received from it are parsed as NDJSON and decoded with `outputSchema`.
 *
 * @category combinators
 * @since 4.0.0
 */
export const duplexSchemaString: {
  /**
   * Wraps a bidirectional string channel with schema-aware NDJSON encoding and
   * decoding.
   *
   * **Details**
   *
   * Values sent to the wrapped channel are encoded with `inputSchema`; strings
   * received from it are parsed as NDJSON and decoded with `outputSchema`.
   *
   * @category combinators
   * @since 4.0.0
   */
  <In extends Schema.Top, Out extends Schema.Top>(
    options: {
      readonly inputSchema: In
      readonly outputSchema: Out
      readonly ignoreEmptyLines?: boolean | undefined
    }
  ): <OutErr, OutDone, InErr, InDone, R>(
    self: Channel.Channel<
      Arr.NonEmptyReadonlyArray<string>,
      OutErr,
      OutDone,
      Arr.NonEmptyReadonlyArray<string>,
      NdjsonError | Schema.SchemaError | InErr,
      InDone,
      R
    >
  ) => Channel.Channel<
    Arr.NonEmptyReadonlyArray<Out["Type"]>,
    NdjsonError | Schema.SchemaError | OutErr,
    OutDone,
    Arr.NonEmptyReadonlyArray<In["Type"]>,
    InErr,
    InDone,
    R | In["EncodingServices"] | Out["DecodingServices"]
  >
  /**
   * Wraps a bidirectional string channel with schema-aware NDJSON encoding and
   * decoding.
   *
   * **Details**
   *
   * Values sent to the wrapped channel are encoded with `inputSchema`; strings
   * received from it are parsed as NDJSON and decoded with `outputSchema`.
   *
   * @category combinators
   * @since 4.0.0
   */
  <Out extends Schema.Top, In extends Schema.Top, OutErr, OutDone, InErr, InDone, R>(
    self: Channel.Channel<
      Arr.NonEmptyReadonlyArray<string>,
      OutErr,
      OutDone,
      Arr.NonEmptyReadonlyArray<string>,
      NdjsonError | Schema.SchemaError | InErr,
      InDone,
      R
    >,
    options: {
      readonly inputSchema: In
      readonly outputSchema: Out
      readonly ignoreEmptyLines?: boolean | undefined
    }
  ): Channel.Channel<
    Arr.NonEmptyReadonlyArray<Out["Type"]>,
    NdjsonError | Schema.SchemaError | OutErr,
    OutDone,
    Arr.NonEmptyReadonlyArray<In["Type"]>,
    InErr,
    InDone,
    R | In["EncodingServices"] | Out["DecodingServices"]
  >
} = dual(2, <Out extends Schema.Top, In extends Schema.Top, OutErr, OutDone, InErr, InDone, R>(
  self: Channel.Channel<
    Arr.NonEmptyReadonlyArray<string>,
    OutErr,
    OutDone,
    Arr.NonEmptyReadonlyArray<string>,
    NdjsonError | Schema.SchemaError | InErr,
    InDone,
    R
  >,
  options: {
    readonly inputSchema: In
    readonly outputSchema: Out
    readonly ignoreEmptyLines?: boolean | undefined
  }
): Channel.Channel<
  Arr.NonEmptyReadonlyArray<Out["Type"]>,
  NdjsonError | Schema.SchemaError | OutErr,
  OutDone,
  Arr.NonEmptyReadonlyArray<In["Type"]>,
  InErr,
  InDone,
  R | In["EncodingServices"] | Out["DecodingServices"]
> => ChannelSchema.duplexUnknown(duplexString(self, options), options))