defmodule Mongo.GridFs.UploadStream do
  @moduledoc """
  This is the upload stream for save streams into the grid fs.

  First you need to create a bucket. The bucket contains the configuration for the grid fs.

  ## Example:
  streaming the file `./test/data/test.jpg` into the grid fs using the upload-stream

      bucket = Bucket.new( pid )
      upload_stream = Upload.open_upload_stream(bucket, "test.jpg", j: true)

      src_filename = "./test/data/test.jpg"
      File.stream!(src_filename, [], 512) |> Stream.into(upload_stream) |> Stream.run()

  """

  import Record, only: [defrecordp: 2]

  alias Mongo.GridFs.Bucket
  alias Mongo.GridFs.UploadStream

  @type t :: %__MODULE__{
          bucket: Bucket.t(),
          id: BSON.ObjectId.t(),
          filename: String.t(),
          metadata: {BSON.document() | nil}
        }
  defstruct bucket: nil, id: nil, filename: nil, metadata: nil

  @doc """
  Creates a new upload stream to insert a file into the grid-fs.
  """
  @spec new(Bucket.t(), String.t(), BSON.document() | nil) :: UploadStream.t()
  def new(bucket, filename, metadata \\ nil) do
    %UploadStream{bucket: bucket, filename: filename, id: Mongo.object_id(), metadata: metadata}
  end

  defimpl Collectable, for: UploadStream do
    ##
    # buffer is the current buffer
    # number is the current chunk number
    #
    defrecordp(:state, [:buffer, :number])

    def into(%UploadStream{} = stream) do
      {state(buffer: <<>>, number: 0), make_fun(stream)}
    end

    def make_fun(%UploadStream{bucket: %{chunk_size: chunk_size}} = stream) do
      fn
        # case: buffer is full
        state(buffer: bin) = s, {:cont, x} when byte_size(bin) >= chunk_size ->
          state(buffer: rest, number: next) = write_buffer(stream, s)
          state(buffer: rest <> x, number: next)

        # case: buffer is empty
        state(buffer: bin, number: n), {:cont, x} ->
          state(buffer: bin <> x, number: n)

        # stream stops, write the rest
        state() = s, :done ->
          _ = flush_buffer(stream, s)
          stream

        # steam halts, write the rest
        state() = s, :halt ->
          _ = flush_buffer(stream, s)
          :ok
      end
    end

    ##
    # flushes the buffer and creates the files document
    #
    defp flush_buffer(
           %UploadStream{
             bucket:
               %Bucket{topology_pid: topology_pid, chunk_size: chunk_size, opts: opts} = bucket,
             filename: filename,
             id: file_id,
             metadata: metadata
           },
           state(buffer: buffer, number: chunk_number)
         ) do
      collection = Bucket.chunks_collection_name(bucket)
      length = chunk_number * chunk_size + byte_size(buffer)
      insert_one_chunk_document(topology_pid, collection, file_id, buffer, chunk_number, opts)

      collection = Bucket.files_collection_name(bucket)

      insert_one_file_document(
        topology_pid,
        collection,
        file_id,
        length,
        chunk_size,
        filename,
        metadata,
        opts
      )
    end

    ##
    # checks if the buffer is smaller than the chunk-size
    # in this case we do nothing
    #
    defp write_buffer(
           %UploadStream{bucket: %Bucket{chunk_size: chunk_size}},
           state(buffer: buffer) = s
         )
         when byte_size(buffer) < chunk_size do
      s
    end

    ##
    # otherwise we
    # write the data to the chunk collections and call the function again with the rest of the buffer
    # for the case that the buffer size is still greater than the chunk size
    #
    defp write_buffer(
           %UploadStream{
             bucket:
               %Bucket{topology_pid: topology_pid, chunk_size: chunk_size, opts: opts} = bucket,
             id: file_id
           } = stream,
           state(buffer: buffer, number: chunk_number)
         ) do
      collection = Bucket.chunks_collection_name(bucket)

      fun = fn <<data::bytes-size(chunk_size), rest::binary>> ->
        next =
          insert_one_chunk_document(topology_pid, collection, file_id, data, chunk_number, opts)

        state(buffer: rest, number: next)
      end

      # write the buffer
      new_state = fun.(buffer)

      # try to write the rest of the buffer
      write_buffer(stream, new_state)
    end

    ##
    # inserts one chunk document
    #
    defp insert_one_chunk_document(
           _topology_pid,
           _collection,
           _file_id,
           data,
           chunk_number,
           _opts
         )
         when byte_size(data) == 0 do
      chunk_number
    end

    defp insert_one_chunk_document(topology_pid, collection, file_id, binary, chunk_number, opts) do
      bson_binary = %BSON.Binary{binary: binary}

      {:ok, _} =
        Mongo.insert_one(
          topology_pid,
          collection,
          %{files_id: file_id, n: chunk_number, data: bson_binary},
          opts
        )

      chunk_number + 1
    end

    ##
    # inserts one file document
    #
    defp insert_one_file_document(
           topology_pid,
           collection,
           file_id,
           length,
           chunk_size,
           filename,
           metadata,
           opts
         ) do
      doc = %{
        _id: file_id,
        length: length,
        filename: filename,
        chunkSize: chunk_size,
        uploadDate: now(),
        metadata: metadata
      }

      {:ok, _} = Mongo.insert_one(topology_pid, collection, doc, opts)
    end

    defp now(), do: DateTime.from_unix!(:os.system_time(), :native)
  end
end
