defmodule DetsPlus do
  @moduledoc """
  DetsPlus persistent tuple storage.

  [DetsPlus](https://github.com/dominicletz/dets_plus) has a similiar API as `dets` but without
  the 2GB file storage limit. Writes are buffered in an
  internal ETS table and synced every `auto_save` period
  to the persistent storage.

  While `sync()` or `auto_save` is in progress the database
  can still read and written.

  There is no commitlog so not synced writes are lost.
  Lookups are possible by key and non-matches are accelerated
  using a bloom filter. The persistent file concept follows
  DJ Bernsteins CDB database format, but uses an Elixir
  encoded header https://cr.yp.to/cdb.html

  Limits are:
  - Total file size: 18_446 Petabyte
  - Maximum entry size: 4 Gigabyte
  - Maximum entry count: :infinity
  """

  # This limits the total database size to 18_446 PB
  @slot_size 8
  @slot_size_bits @slot_size * 8

  # This limits the biggest entry size to 4 GB
  @entry_size_size 4
  @entry_size_size_bits @entry_size_size * 8

  # We're using sha256 - should not have conflicts ever
  @hash_size 32
  @hash_size_bits @hash_size * 8

  use GenServer

  @enforce_keys [:version]
  defstruct [
    :version,
    :header_offset,
    :bloom,
    :bloom_size,
    :filename,
    :fp,
    :name,
    :ets,
    :mode,
    :auto_save,
    :keypos,
    :type,
    :file_entries,
    :slot_counts,
    :table_offsets,
    :sync,
    :sync_waiters,
    :file_size,
    :sync_fallback
  ]

  @version 1

  @doc """
    Opens an existing table or creates a new table. If no
    `file` argument is provided the table name will be used.

    Arguments:

    - `auto_save` - The autosave interval. If the interval is an integer Time, the table is flushed to disk whenever it is not accessed for Time milliseconds. A table that has been flushed requires no reparation when reopened after an uncontrolled emulator halt. If the interval is the atom infinity, autosave is disabled. Defaults to 180000 (3 minutes).
    - `keypos` - The position of the element of each object to be used as key. Defaults to 1. The ability to explicitly state the key position is most convenient when we want to store Erlang records in which the first position of the record is the name of the record type.
  """
  def open_file(name, args \\ []) when is_atom(name) do
    filename = Keyword.get(args, :file, name) |> do_string()

    state =
      if File.exists?(filename) do
        load_state(filename)
        |> init_table_offsets()
      else
        mode = Keyword.get(args, :access, :read_write)
        auto_save = Keyword.get(args, :auto_save, 18_000)
        keypos = Keyword.get(args, :keypos, 1)
        type = Keyword.get(args, :type, :set)

        # unused options from dets:
        # - ram_file
        # - max_no_slots
        # - min_no_slots
        # - repair

        %DetsPlus{
          version: @version,
          bloom: "",
          bloom_size: 0,
          header_offset: 0,
          filename: filename,
          fp: nil,
          name: name,
          mode: mode,
          auto_save: auto_save,
          keypos: keypos,
          type: type,
          file_entries: 0,
          slot_counts: %{},
          file_size: 0,
          sync: nil,
          sync_waiters: [],
          sync_fallback: %{}
        }
      end

    case GenServer.start_link(__MODULE__, state, hibernate_after: 5_000, name: name) do
      {:ok, _pid} -> {:ok, name}
      err -> err
    end
  end

  defp load_state(filename) do
    fp = file_open(filename)

    {:ok, <<header_size::unsigned-size(@entry_size_size_bits)>>} =
      PagedFile.pread(fp, 0, @entry_size_size)

    {:ok, header} = PagedFile.pread(fp, @entry_size_size, header_size)

    %DetsPlus{version: version} = state = :erlang.binary_to_term(header)

    if version > @version do
      raise("incompatible dets+ version #{version}")
    end

    %DetsPlus{
      state
      | fp: fp,
        header_offset: header_size + @entry_size_size,
        file_size: File.stat!(filename).size
    }
  end

  defp store_state(state = %DetsPlus{}, fp) do
    bin =
      :erlang.term_to_binary(
        %DetsPlus{
          state
          | fp: nil,
            sync: nil,
            sync_waiters: [],
            header_offset: 0,
            table_offsets: %{},
            ets: nil
        },
        [:compressed]
      )

    size = byte_size(bin)
    :ok = PagedFile.pwrite(fp, 0, <<size::unsigned-size(@entry_size_size_bits)>>)
    :ok = PagedFile.pwrite(fp, @entry_size_size, bin)
    %DetsPlus{state | header_offset: size + @entry_size_size}
  end

  @impl true
  def init(state = %DetsPlus{auto_save: auto_save}) do
    if is_integer(auto_save) do
      :timer.send_interval(auto_save, :auto_save)
    end

    {:ok, init_ets(state)}
  end

  defp init_ets(state = %DetsPlus{name: name, type: type, keypos: keypos}) do
    %DetsPlus{state | ets: :ets.new(name, [type, {:keypos, keypos}])}
  end

  defp init_table_offsets(
         state = %DetsPlus{header_offset: header_offset, slot_counts: slot_counts}
       ) do
    table_offsets =
      Enum.reduce(1..256, %{0 => header_offset}, fn table_idx, table_offsets ->
        offset =
          Map.get(table_offsets, table_idx - 1) +
            Map.get(slot_counts, table_idx - 1, 0) * @slot_size

        Map.put(table_offsets, table_idx, offset)
      end)

    %DetsPlus{state | table_offsets: table_offsets}
  end

  @doc """
  Syncs pending writes to the persistent file and closes the table.
  """
  @spec close(atom | pid) :: :ok
  def close(pid) do
    call(pid, :sync)
    GenServer.stop(pid)
  end

  @doc """
    Inserts one or more objects into the table. If there already exists an object with a key matching the key of some of the given objects, the old object will be replaced.
  """
  @spec insert(atom | pid, tuple()) :: :ok | {:error, atom()}
  def insert(pid, tuple) do
    call(pid, {:insert, tuple})
  end

  @doc """
    Inserts one or more objects into the table. If there already exists an object with a key matching the key of some of the given objects, the old object will be replaced.
  """
  def insert_new(pid, tuple) do
    call(pid, {:insert_new, tuple})
  end

  @doc """
  Returns a list of all objects with key Key stored in the table.

  Example:

  ```
  2> DetsPlus.open_file(:abc)
  {ok,:abc}
  3> DetsPlus.insert(:abc, {1,2,3})
  ok
  4> DetsPlus.insert(:abc, {1,3,4})
  ok
  5> DetsPlus.lookup(:abc, 1).
  [{1,3,4}]
  ```

  If the table type is set, the function returns either the empty list or a list with one object, as there cannot be more than one object with a given key. If the table type is bag or duplicate_bag, the function returns a list of arbitrary length.

  Notice that the order of objects returned is unspecified. In particular, the order in which objects were inserted is not reflected.
  """
  @spec lookup(atom | pid, any) :: [tuple()] | {:error, atom()}
  def lookup(pid, key) do
    call(pid, {:lookup, key})
  end

  @doc """
  Works like `lookup/2`, but does not return the objects. Returns true if one or more table elements has the key `key`, otherwise false.
  """
  @spec member?(atom | pid, any) :: false | true | {:error, atom}
  def member?(pid, key) do
    case lookup(pid, key) do
      [] -> false
      list when is_list(list) -> true
      error -> error
    end
  end

  @doc """
  Same as `member?/2`
  """
  @spec member(atom | pid, any) :: false | true | {:error, atom}
  def member(pid, key) do
    member?(pid, key)
  end

  @doc """
  Ensures that all updates made to table are written to disk. While the sync is running the
  table can still be used for reads and writes, but writes issed after the `sync/1` call
  will not be part of the persistent file. These new changes will only be included in the
  next sync call.
  """
  @spec sync(atom | pid) :: :ok
  def sync(pid) do
    call(pid, :sync, :infinity)
  end

  @doc """
  Returns information about table Name as a list of tuples:

  - `{file_size, integer() >= 0}}` - The file size, in bytes.
  - `{filename, file:name()}` - The name of the file where objects are stored.
  - `{keypos, keypos()}` - The key position.
  - `{size, integer() >= 0}` - The number of objects estimated in the table.
  - `{type, type()}` - The table type.
  """
  @spec info(atom | pid) :: [] | nil
  def info(pid) do
    call(pid, :info)
  end

  @doc """
  Returns the information associated with `item` for the table. The following items are allowed:

  - `{file_size, integer() >= 0}}` - The file size, in bytes.
  - `{filename, file:name()}` - The name of the file where objects are stored.
  - `{keypos, keypos()}` - The key position.
  - `{size, integer() >= 0}` - The number of objects estimated in the table.
  - `{type, type()}` - The table type.
  """
  @spec info(atom | pid, :file_size | :filename | :keypos | :size | :type) :: any()
  def info(pid, item)
      when item == :file_size or item == :filename or item == :keypos or item == :size or
             item == :type do
    case info(pid) do
      nil -> nil
      list -> Keyword.get(list, item)
    end
  end

  defp call(pid, cmd, timeout \\ :infinity) do
    GenServer.call(pid, cmd, timeout)
  end

  @impl true
  def handle_call(
        {:insert, tuple},
        _from,
        state = %DetsPlus{ets: ets, sync: sync, sync_fallback: fallback}
      ) do
    if sync == nil do
      :ets.insert(ets, tuple)
      {:reply, :ok, check_auto_save(state)}
    else
      fallback =
        List.wrap(tuple)
        |> Enum.reduce(fallback, fn tuple, fallback ->
          Map.put(fallback, do_key(state, tuple), tuple)
        end)

      {:reply, :ok, %DetsPlus{state | sync_fallback: fallback}}
    end
  end

  def handle_call(
        {:insert_new, tuple},
        from,
        state = %DetsPlus{ets: ets, sync_fallback: fallback}
      ) do
    tuples = List.wrap(tuple)

    exists =
      Enum.any?(tuples, fn tuple ->
        key = do_key(state, tuple)

        Map.has_key?(fallback, key) || :ets.lookup(ets, key) != [] ||
          file_lookup(state, key) != []
      end)

    if exists do
      {:reply, false, state}
    else
      {:reply, :ok, state} = handle_call({:insert, tuples}, from, state)
      {:reply, true, state}
    end
  end

  def handle_call(
        {:lookup, key},
        _from,
        state = %DetsPlus{ets: ets, sync_fallback: fallback}
      ) do
    case Map.get(fallback, key) do
      nil ->
        case :ets.lookup(ets, key) do
          [] -> {:reply, file_lookup(state, key), state}
          other -> {:reply, other, state}
        end

      value ->
        {:reply, [value], state}
    end
  end

  def handle_call(
        :info,
        _from,
        state = %DetsPlus{
          filename: filename,
          keypos: keypos,
          type: type,
          ets: ets,
          file_size: file_size,
          file_entries: file_entries
        }
      ) do
    size = :ets.info(ets, :size) + file_entries
    info = [file_size: file_size, filename: filename, keypos: keypos, size: size, type: type]
    {:reply, info, state}
  end

  def handle_call(:sync, from, state = %DetsPlus{sync: nil, ets: ets}) do
    if :ets.info(ets, :size) == 0 do
      {:reply, :ok, state}
    else
      sync = spawn_sync_worker(state)
      {:noreply, %DetsPlus{state | sync: sync, sync_waiters: [from]}}
    end
  end

  def handle_call(:sync, from, state = %DetsPlus{sync: sync, sync_waiters: sync_waiters})
      when is_pid(sync) do
    {:noreply, %DetsPlus{state | sync_waiters: [from | sync_waiters]}}
  end

  @impl true
  def handle_cast(
        {:sync_complete, new_filename, new_state = %DetsPlus{}},
        %DetsPlus{
          fp: fp,
          filename: filename,
          ets: ets,
          sync_waiters: waiters,
          sync_fallback: fallback
        }
      ) do
    if fp != nil do
      :ok = PagedFile.close(fp)
    end

    File.rename!(new_filename, filename)
    fp = file_open(filename)

    :ets.delete_all_objects(ets)
    :ets.insert(ets, Map.values(fallback))

    for w <- waiters do
      :ok = GenServer.reply(w, :ok)
    end

    {:noreply, %DetsPlus{new_state | fp: fp, sync: nil, sync_fallback: %{}, sync_waiters: []}}
  end

  @impl true
  def handle_info(:auto_save, state = %DetsPlus{sync: sync, ets: ets}) do
    sync =
      if sync == nil and :ets.info(ets, :size) > 0 do
        spawn_sync_worker(state)
      else
        sync
      end

    {:noreply, %DetsPlus{state | sync: sync}}
  end

  defp check_auto_save(state) do
    state
  end

  defp spawn_sync_worker(
         state = %DetsPlus{
           ets: ets,
           fp: fp,
           filename: filename,
           file_entries: file_entries
         }
       ) do
    # assumptions here
    # 1. ets data set is small enough to fit into memory
    # 2. fp entries are sorted by hash
    dets = self()

    worker =
      spawn_link(fn ->
        new_dataset = :ets.tab2list(ets)
        send(dets, :continue)

        # Ensuring hash function sort order
        new_dataset =
          parallel_hash(state, new_dataset)
          |> Enum.sort_by(fn {hash, _tuple} -> hash end, :asc)

        new_filename = "#{filename}.buffer"

        # opts =
        #   if String.ends_with?(filename, ".gz") do
        #     [:raw, :read, :read_ahead, :write, :delayed_write, :binary, :compressed]
        #   else
        #     [:raw, :read, :read_ahead, :write, :delayed_write, :binary]
        #   end
        opts = [page_size: 1_000_000, max_pages: 1000]

        # setting the bloom size based of a size estimate
        bloom_size = (file_entries + length(new_dataset)) * 10
        state = bloom_create(state, bloom_size)

        old_file =
          if fp != nil do
            FileReader.new(fp, table_offset(state, 256), module: PagedFile, buffer_size: 512_000)
          else
            nil
          end

        state = pre_scan_file(state, new_dataset, old_file)

        PagedFile.delete(new_filename)
        {:ok, new_file} = PagedFile.open(new_filename, opts)

        state =
          %DetsPlus{state | fp: new_file}
          |> bloom_finalize()
          |> store_state(new_file)
          |> init_table_offsets()

        state = write_to_file(state, table_offset(state, 256), new_dataset, old_file)
        PagedFile.close(new_file)
        GenServer.cast(dets, {:sync_complete, new_filename, state})
      end)

    # Profiler.fprof(worker)

    receive do
      :continue ->
        :erlang.garbage_collect()
        worker
    end
  end

  @min_chunk_size 10_000
  @max_tasks 4
  defp parallel_hash(state, new_dataset, tasks \\ 1) do
    len = length(new_dataset)

    if len > @min_chunk_size and tasks < @max_tasks do
      {a, b} = Enum.split(new_dataset, div(len, 2))
      task = Task.async(fn -> parallel_hash(state, a, tasks * 2) end)
      result = parallel_hash(state, b, tasks * 2)
      Task.await(task, :infinity) ++ result
    else
      Enum.map(new_dataset, fn tuple ->
        key = do_key(state, tuple)
        hash = do_hash(state, key)
        {hash, tuple}
      end)
    end
  end

  defp pre_scan_file(state, new_dataset, old_file) do
    state = %DetsPlus{state | file_entries: 0, slot_counts: %{}}

    state =
      %DetsPlus{slot_counts: slot_counts} =
      iterate(
        state,
        state,
        new_dataset,
        old_file,
        fn state = %DetsPlus{file_entries: file_entries, slot_counts: slot_counts},
           entry_hash,
           _ ->
          table_idx = rem(entry_hash, 256)
          slot_counts = Map.update(slot_counts, table_idx, 1, fn count -> count + 1 end)
          state = bloom_add(state, entry_hash)
          %DetsPlus{state | file_entries: file_entries + 1, slot_counts: slot_counts}
        end
      )

    new_slot_counts =
      Enum.map(slot_counts, fn {key, value} -> {key, trunc(value * 1.5) + 1} end)
      |> Map.new()

    %DetsPlus{state | slot_counts: new_slot_counts}
  end

  defp write_to_file(
         state = %DetsPlus{},
         new_file_offset,
         new_dataset,
         old_file
       ) do
    {state, new_file_offset} =
      iterate(state, {state, new_file_offset}, new_dataset, old_file, fn {state, new_file_offset},
                                                                         entry_hash,
                                                                         entry ->
        new_file_offset = file_put_entry(state, entry_hash, entry, new_file_offset)
        {state, new_file_offset}
      end)

    %DetsPlus{state | file_size: new_file_offset}
  end

  # this function takes a new_dataset and an old file and merges them, it calls
  # on every entry the callback `fun.(acc, entry_hash, binary_entry)` and returns the final acc
  defp iterate(
         state = %DetsPlus{},
         acc,
         new_dataset,
         file_reader,
         fun
       ) do
    # Reading a new entry from the top of the dataset or nil
    {new_entry, new_entry_hash} =
      case new_dataset do
        [{new_entry_hash, new_entry} | _rest] ->
          {new_entry, new_entry_hash}

        [] ->
          {nil, nil}
      end

    # Reading a new entry from the file or falling back to else if there is no next
    # entry in the file anymore
    with false <- file_reader == nil,
         {new_file_reader, <<old_size::unsigned-size(@entry_size_size_bits)>>} <-
           FileReader.read(file_reader, @entry_size_size) do
      {new_file_reader, old_entry} = FileReader.read(new_file_reader, old_size)
      entry = :erlang.binary_to_term(old_entry)
      entry_key = do_key(state, entry)
      entry_hash = do_hash(state, entry_key)
      {entry, old_entry, entry_hash, new_file_reader}

      case new_entry_hash do
        # reached end of new_dataset
        nil ->
          iterate(state, fun.(acc, entry_hash, old_entry), [], new_file_reader, fun)

        # replacing an old entry with a new entry
        ^entry_hash ->
          bin = :erlang.term_to_binary(new_entry)
          iterate(state, fun.(acc, new_entry_hash, bin), tl(new_dataset), new_file_reader, fun)

        # inserting the new entry before the old
        new_entry_hash when new_entry_hash < entry_hash ->
          bin = :erlang.term_to_binary(new_entry)
          iterate(state, fun.(acc, new_entry_hash, bin), tl(new_dataset), file_reader, fun)

        # inserting the old entry before the new
        new_entry_hash when new_entry_hash > entry_hash ->
          iterate(state, fun.(acc, entry_hash, old_entry), new_dataset, new_file_reader, fun)
      end
    else
      atom when atom == true or atom == :eof ->
        # reached end of both lines
        if new_entry == nil do
          acc
        else
          # reached end of file for the old file
          bin = :erlang.term_to_binary(new_entry)
          iterate(state, fun.(acc, new_entry_hash, bin), tl(new_dataset), nil, fun)
        end
    end
  end

  defp bloom_create(state = %DetsPlus{}, bloom_size) do
    {:ok, bloom} = :file.open(:binary.copy(<<0>>, bloom_size), [:ram, :read, :write, :binary])
    %DetsPlus{state | bloom_size: bloom_size, bloom: bloom}
  end

  defp bloom_add(state = %DetsPlus{bloom_size: bloom_size, bloom: bloom}, hash) do
    :ok = :file.pwrite(bloom, rem(hash, bloom_size), <<1>>)
    state
    # %DetsPlus{state | bloom: Map.put(bloom, rem(hash, bloom_size), true)}
  end

  defp bloom_finalize(state = %DetsPlus{bloom: bloom, bloom_size: bloom_size}) do
    {:ok, binary} = :file.pread(bloom, 0, bloom_size)
    :file.close(bloom)
    %DetsPlus{state | bloom: binary}
  end

  defp bloom_lookup(%DetsPlus{bloom_size: bloom_size, bloom: bloom}, hash) do
    :binary.at(bloom, rem(hash, bloom_size)) == 1
  end

  defp table_offset(%DetsPlus{table_offsets: nil}, _table_idx) do
    nil
  end

  defp table_offset(%DetsPlus{table_offsets: offsets}, table_idx) do
    Map.get(offsets, table_idx)
  end

  defp file_put_entry(
         state = %DetsPlus{fp: fp, slot_counts: slot_counts},
         hash,
         entry,
         offset
       ) do
    size = byte_size(entry)
    table_idx = rem(hash, 256)

    slot_count = Map.get(slot_counts, table_idx)
    slot = rem(div(hash, 256), slot_count)
    base_offset = table_offset(state, table_idx)
    file_put_entry_probe(fp, base_offset, slot, slot_count, offset)

    :ok =
      PagedFile.pwrite(
        fp,
        offset,
        <<size::unsigned-size(@entry_size_size_bits), entry::binary()>>
      )

    offset + size + @entry_size_size
  end

  # recursivley retry next slot if current slot is used already
  defp file_put_entry_probe(fp, base_offset, slot, slot_count, value) do
    slot = rem(slot, slot_count)
    # There could be some analytics here for number of probings worst case etc...

    # probe batching to improve file io
    batch_size = min(32, slot_count - slot)

    {slot, probe} =
      case PagedFile.pread(fp, base_offset + slot * @slot_size, @slot_size * batch_size) do
        # eof means the file has not been extended yet, so safe to write
        :eof ->
          {slot, true}

        {:ok, data} ->
          data = data <> :binary.copy(<<0>>, @slot_size * batch_size - byte_size(data))
          # all zeros means there is no entry address stored yet
          for(<<x::unsigned-size(@slot_size_bits) <- data>>, do: x == 0)
          |> Enum.with_index()
          |> Enum.find(fn {x, _index} -> x end)
          |> case do
            nil -> {slot, false}
            {true, idx} -> {slot + idx, true}
          end
      end

    if probe do
      :ok =
        PagedFile.pwrite(
          fp,
          base_offset + slot * @slot_size,
          <<value::unsigned-size(@slot_size_bits)>>
        )
    else
      file_put_entry_probe(fp, base_offset, slot + batch_size, slot_count, value)
    end
  end

  defp file_lookup(%DetsPlus{file_entries: 0}, _key), do: []

  defp file_lookup(state = %DetsPlus{slot_counts: slot_counts}, key) do
    hash = do_hash(state, key)
    table_idx = rem(hash, 256)
    slot_count = Map.get(slot_counts, table_idx, 0)

    if bloom_lookup(state, hash) and slot_count > 0 do
      # if slot_count > 0 do
      slot = rem(div(hash, 256), slot_count)

      {ret, _n} =
        file_lookup_slot_loop(state, key, table_offset(state, table_idx), slot, slot_count)

      # if n > 2 do
      #   IO.inspect({ret, n, slot_count})
      # end

      ret
    else
      []
    end
  end

  defp batch_read(fp, point, count) do
    {:ok, data} = PagedFile.pread(fp, point, @slot_size * count)
    data = data <> :binary.copy(<<0>>, @slot_size * count - byte_size(data))
    for <<offset::unsigned-size(@slot_size_bits) <- data>>, do: offset
  end

  @batch_size 32
  defp file_lookup_slot_loop(
         state = %DetsPlus{fp: fp},
         key,
         base_offset,
         slot,
         slot_count,
         n \\ 0
       ) do
    slot = rem(slot, slot_count)
    point = base_offset + slot * @slot_size
    batch_size = min(@batch_size, slot_count - slot)
    offsets = batch_read(fp, point, batch_size)

    if hd(offsets) == 0 do
      {[], n}
    else
      # a zero is an indication of the end
      offsets = Enum.take_while(offsets, fn x -> x != 0 end)
      len = length(offsets)

      {:ok, sizes} = PagedFile.pread(fp, Enum.zip(offsets, List.duplicate(@entry_size_size, len)))
      sizes = Enum.map(sizes, fn <<size::unsigned-size(@entry_size_size_bits)>> -> size end)

      offsets = Enum.map(offsets, fn offset -> offset + @entry_size_size end)
      {:ok, entries} = PagedFile.pread(fp, Enum.zip(offsets, sizes))

      Enum.find_value(entries, fn entry ->
        entry = :erlang.binary_to_term(entry)

        if do_key(state, entry) == key do
          entry
        end
      end)
      |> case do
        nil ->
          if len < batch_size do
            {[], n}
          else
            file_lookup_slot_loop(state, key, base_offset, slot + batch_size, slot_count, n + 1)
          end

        entry ->
          {[entry], n}
      end
    end
  end

  defp do_key(%DetsPlus{keypos: keypos}, tuple) do
    elem(tuple, keypos - 1)
  end

  defp do_hash(%DetsPlus{}, key) do
    <<slot::unsigned-size(@hash_size_bits)>> = :crypto.hash(:sha256, :erlang.term_to_binary(key))

    slot
  end

  defp do_string(atom) when is_atom(atom), do: Atom.to_string(atom)
  defp do_string(list) when is_list(list), do: List.to_string(list)
  defp do_string(string) when is_binary(string), do: string

  defp file_open(filename) do
    # opts =
    #   if String.ends_with?(filename, ".gz") do
    #     [:read, :read_ahead, :binary, :compressed]
    #   else
    #     [:read, :read_ahead, :binary]
    #   end
    opts = [page_size: 1_000_000, max_pages: 1000]
    {:ok, fp} = PagedFile.open(filename, opts)
    fp
  end
end