import type { Message, Stream } from "Stream"
import type { Error } from "__IOError__"

import List from "List"
import { Data, Error, NoMessage, Stream, Talkback } from "Stream"
import IOError from "__IOError__"



#iftarget js

schedule :: ({} -> {}) -> {}
schedule = (cb) => #- {
  {Node} process.nextTick(cb) {/Node}
  {Browser} setTimeout(cb, 0) {/Browser}
} -#

#elseif llvm

setTimeout :: ({} -> {}) -> Integer -> {}
setTimeout = extern "__setTimeout__"

schedule :: ({} -> {}) -> {}
schedule = (cb) => {
  setTimeout(cb, 0)
}

#endif



#iftarget llvm


runTalkback :: Message e a -> Integer -> {}
runTalkback = (msg, t) => where(msg) {
  Talkback(cb) =>
    cb(t)

  _ =>
    {}
}


type Handle = Handle


openFileForReadFFI :: String -> ({} -> {}) -> (Integer -> {}) -> Handle
openFileForReadFFI = extern "madlib__file__openFileForBufferedRead"


readChunkFFI :: Handle -> (ByteArray -> {}) -> (Integer -> {}) -> ({} -> {}) -> {}
readChunkFFI = extern "madlib__file__readChunkFromFile"


cancelBufferedRead :: Handle -> {}
cancelBufferedRead = extern "madlib__file__cancelBufferedRead"


readBytes :: String -> Stream Error ByteArray
export readBytes = (path) => Stream(
  (start, sink) => {
    done = false
    handle = Handle
    reading = false

    readChunk = () => {
      if (!reading) do {
        reading := true
        readChunkFFI(
          handle,
          (chunk) => {
            reading := false
            pipe(
              Data,
              sink(1),
            )(chunk)
          },
          pipe(
            IOError.fromLibuvError,
            Error,
            sink(2),
          ),
          () => {
            reading := false
            done := true
            sink(2, NoMessage)
          },
        )
      }
    }

    if (start == 0) do {
      handle := openFileForReadFFI(
        path,
        () => {
          sink(
            0,
            Talkback(
              (t) => {
                if (t == 1 && !done) do {
                  if (!reading) do {
                    readChunk()
                  }
                }
                if (t == 2 && !done) {
                  cancelBufferedRead(handle)
                }
              },
            ),
          )
        },
        pipe(
          IOError.fromLibuvError,
          Error,
          sink(2),
        ),
      )
    }
  },
)


cancelBufferedWrite :: Handle -> {}
cancelBufferedWrite = extern "madlib__file__cancelBufferedWrite"


openFileForWriteFFI :: String -> ({} -> {}) -> (Integer -> {}) -> Handle
openFileForWriteFFI = extern "madlib__file__openBufferedFileForWrite"


writeChunkFFI :: Handle -> ByteArray -> ({} -> {}) -> (Integer -> {}) -> {}
writeChunkFFI = extern "madlib__file__writeChunkToFile"


writeBytes :: String -> Stream e ByteArray -> Stream Error {}
export writeBytes = (path, source) => Stream(
  (start, sink) => {
    open = false
    writing = false
    sourceDone = false
    done = false
    handle = Handle
    talkback = NoMessage
    chunksToWrite = []

    writeChunks = () => {
      if (!writing && open) {
        where(chunksToWrite) {
          [chunk, ...more] =>
            do {
              writing := true
              chunksToWrite := more
              writeChunkFFI(
                handle,
                chunk,
                (_) => {
                  writing := false
                  if (List.isEmpty(chunksToWrite) && sourceDone && !done) do {
                    done := true
                    cancelBufferedWrite(handle)
                    sink(2, NoMessage)
                  } else if (!List.isEmpty(chunksToWrite)) {
                    schedule(writeChunks)
                  } else {
                    runTalkback(talkback, 1)
                  }
                },
                pipe(
                  IOError.fromLibuvError,
                  Error,
                  sink(2),
                ),
              )
            }

          [] =>
            {}
        }
      }
    }

    if (start == 0) do {
      where(source) {
        Stream(src) =>
          src(
            0,
            (t, d) => {
              if (t == 0) do {
                talkback := d
                handle := openFileForWriteFFI(
                  path,
                  () => {
                    open := true
                    writeChunks()

                    where(d) {
                      Talkback(_t) =>
                        sink(0, Talkback(_t))

                      _ =>
                        {}
                    }
                  },
                  pipe(
                    IOError.fromLibuvError,
                    Error,
                    sink(2),
                  ),
                )
              }

              if (t == 1) do {
                where(d) {
                  Data(chunk) =>
                    do {
                      chunksToWrite := [...chunksToWrite, chunk]
                      writeChunks()
                    }

                  _ =>
                    {}
                }
              }

              if (t == 2) {
                where(d) {
                  Error(e) =>
                    // Source failed, we need to cancel the write
                    sink(2, Error(IOError.UnknownError))

                  _ =>
                    do {
                      sourceDone := true
                      if (List.isEmpty(chunksToWrite)) do {
                        done := true
                        cancelBufferedWrite(handle)
                        sink(2, NoMessage)
                      }
                    }
                }
              }
            },
          )
      }
    }
  },
)

#elseif js

#- import prelude_stream_file_fs from "fs" -#


_readBytes :: (e -> Message e a)
  -> (a -> Message e a)
  -> Message e a
  -> String
  -> Stream Error ByteArray
_readBytes = (errorCtor, dataCtor, noMessage, path) => Stream(
  (start, sink) => {
    done = false
    #- let cancel -#
    if (start == 0) do {
      #-
        const readableStream = prelude_stream_file_fs.createReadStream(path, {'bufferSize': 256 * 1024})

        cancel = () => readableStream.destroy()

        readableStream.pause()

        readableStream.on("error", (e) => {
          sink(2)(errorCtor({ __constructor: "UnknownError", __a: 0 }))
        })

        readableStream.on("data", (chunk) => {
          readableStream.pause()
          sink(1)(dataCtor(Uint8Array.from(chunk)))
        })

        readableStream.on("end", () => {
          done = true
          sink(2)(noMessage)
        })
      -#

      sink(
        0,
        Talkback(
          (t) => {
            if (t == 1) do {
              #- readableStream.resume() -#
            }
            if (t == 2 && !done) {
              #- cancel && cancel() -#
            }
          },
        ),
      )
    }
  },
)

readBytes :: String -> Stream Error ByteArray
export readBytes = (path) => _readBytes(Error, Data, NoMessage, path)


_writeBytes :: (({} -> {}) -> {})
  -> (e -> Message e a)
  -> Message e a
  -> String
  -> Stream e ByteArray
  -> Stream Error {}
_writeBytes = (_schedule, errorCtor, noMessage, path, source) => Stream(
  (start, sink) => {
    open = false
    writing = false
    sourceDone = false
    done = false
    #- let writableStream -#
    talkback = NoMessage
    chunksToWrite = []

    #-
    const writeChunks = () => {
      if (!writing && open && chunksToWrite !== null) {
        writing = true
        const chunk = chunksToWrite.v
        chunksToWrite = chunksToWrite.n

        writableStream.write(chunk, 'utf8', () => {
          if (chunksToWrite === null && sourceDone && !done) {
            done = true
            writableStream.destroy()
            sink(2)(noMessage)
          } else {
            talkback._0(1)
            writing = false
            if (chunksToWrite !== null) {
              _schedule(writeChunks)
            }
          }
        })
      }
    }
    -#

    if (start == 0) do {
      where(source) {
        Stream(src) =>
          src(
            0,
            (t, d) => {
              if (t == 0) do {
                talkback := d
                #-
                writableStream = prelude_stream_file_fs.createWriteStream(path)
                writableStream.on('error', () => {
                  sink(2)(errorCtor({ __constructor: "UnknownError", __a: 0 }))
                })
                writableStream.on('ready', () => {
                  open = true
                  sink(0)(d)
                })
                -#
              }

              if (t == 1) do {
                where(d) {
                  Data(chunk) =>
                    do {
                      chunksToWrite := [...chunksToWrite, chunk]
                      #- writeChunks() -#
                    }

                  _ =>
                    {}
                }
              }

              if (t == 2) {
                where(d) {
                  Error(e) =>
                    // Source failed, we need to cancel the write
                    sink(2, Error(IOError.UnknownError))

                  _ =>
                    do {
                      sourceDone := true
                      if (List.isEmpty(chunksToWrite)) do {
                        done := true
                        #- writableStream.destroy() -#
                        sink(2, NoMessage)
                      }
                    }
                }
              }
            },
          )
      }
    }
  },
)

writeBytes :: String -> Stream e ByteArray -> Stream Error {}
export writeBytes = _writeBytes(schedule, Error, NoMessage)

#endif