  This Source Code Form is subject to the terms of the Mozilla Public
  License , v. 2.0 . If a copy of the MPL was not distributed with this
  file , You can obtain one at /.
%%
  Copyright © 2022 - 2023 VMware , Inc. or its affiliates .   All rights reserved .
%%

  @doc Khepri import / export callback module using Erlang terms formatted as
%% plain text as its external format.
%%
%% The exported file could be read using {@link file:consult/1} to get back
%% the list of backup items.
%%
%% This callback module takes a filename or an opened file descriptor as its
%% private data passed to {@link khepri:export/4} and {@link khepri:import/3}.
%%
%% Example:
%% ```
  ok = khepri : put(StoreId , " /:stock/:wood / Oak " , 100 ) .
  ok = khepri : put(StoreId , " /:stock/:wood / Mapple " , 55 ) .
  ok =  : export(StoreId , khepri_export_erlang , " export.erl " ) .
%% '''
%%
%% Content of `export.erl':
%% ```
%% {put,[stock,wood,<<"Mapple">>],{p_data,55},#{},#{}}.
%% {put,[stock,wood,<<"Oak">>],{p_data,100},#{},#{}}.
%% '''

-module(khepri_export_erlang).

-behavior(khepri_import_export).

-include_lib("kernel/include/logger.hrl").

%% How many bytes to read from the file descriptor before `read/1' returns a
%% list of backup items. `khepri_import_export' will call this module again to
%% read the next batch.
-define(IMPORT_APPROX_LIMIT_PER_CALL, 100 * 1024). %% In bytes.

-export([open_write/1,
         write/2,
         commit_write/1,
         abort_write/1,

         open_read/1,
         read/1,
         close_read/1]).

-type write_state() :: #{fd := file:io_device(),
                         filename => file:name_all()}.

-type read_state() :: #{fd := file:io_device(),
                        max_chunk := non_neg_integer(),
                        location := erl_anno:location(),
                        start_pos => integer(),
                        filename => file:name_all()}.

-spec open_write(Filename | Fd) -> Ret when
      Filename :: file:name_all(),
      Fd :: file:io_device(),
      Ret :: {ok, WriteState} | {error, any()},
      WriteState :: write_state().
  @private

open_write(Filename) when is_list(Filename) ->
    ?LOG_DEBUG(
       "~s: opening file \"~ts\" for export",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    case file:open(Filename, [write]) of
        {ok, Fd} ->
            State = #{fd => Fd,
                      filename => Filename},
            {ok, State};
        {error, _} = Error ->
            Error
    end;
open_write(Fd) ->
    ?LOG_DEBUG(
       "~s: using existing file descriptor ~p",
       [?MODULE, Fd],
       #{domain => [khepri, import_export, ?MODULE]}),
    State = #{fd => Fd},
    {ok, State}.

-spec write(WriteState, BackupItems) -> Ret when
      WriteState :: write_state(),
      BackupItems :: [khepri_import_export:backup_item()],
      Ret :: {ok, WriteState} | {error, any()}.
  @private

write(#{fd := Fd} = State, [BackupItem | Rest]) ->
    ?LOG_DEBUG(
       "~s: exporting: ~p",
       [?MODULE, BackupItem],
       #{domain => [khepri, import_export, ?MODULE]}),
    Binary = io_lib:format("~0p.~n", [BackupItem]),
    case file:write(Fd, Binary) of
        ok                 -> write(State, Rest);
        {error, _} = Error -> Error
    end;
write(State, []) ->
    ?LOG_DEBUG(
       "~s: finished exporting",
       [?MODULE],
       #{domain => [khepri, import_export, ?MODULE]}),
    {ok, State}.

-spec commit_write(WriteState) -> Ret when
      WriteState :: write_state(),
      Ret :: ok | {error, any()}.
  @private

commit_write(#{fd := Fd, filename := Filename}) ->
    ?LOG_DEBUG(
       "~s: closing file \"~ts\" after export",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    case file:close(Fd) of
        ok                 -> ok;
        {error, _} = Error -> Error
    end;
commit_write(_State) ->
    ?LOG_DEBUG(
       "~s: used existing file descriptor; nothing to close after export",
       [?MODULE],
       #{domain => [khepri, import_export, ?MODULE]}),
    ok.

-spec abort_write(WriteState) -> Ret when
      WriteState :: write_state(),
      Ret :: ok.
  @private

abort_write(#{fd := Fd, filename := Filename}) ->
    ?LOG_DEBUG(
       "~s: aborting export to file \"~ts\"",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    _ = file:close(Fd),
    _ = file:delete(Filename),
    ok;
abort_write(#{fd := Fd}) ->
    ?LOG_DEBUG(
       "~s: aborting export to existing file descriptor ~p",
       [?MODULE, Fd],
       #{domain => [khepri, import_export, ?MODULE]}),
    ok.

-spec open_read(Filename | Fd) -> Ret when
      Filename :: file:name_all(),
      Fd :: file:io_device(),
      Ret :: {ok, ReadState} | {error, any()},
      ReadState :: read_state().
  @private

open_read(Filename) when is_list(Filename) ->
    ?LOG_DEBUG(
       "~s: opening file \"~ts\" for import",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    case file:open(Filename, [read]) of
        {ok, Fd} ->
            _ = epp:set_encoding(Fd),
            State = #{fd => Fd,
                      max_chunk => ?IMPORT_APPROX_LIMIT_PER_CALL,
                      location => 1,
                      filename => Filename},
            {ok, State};
        {error, _} = Error ->
            Error
    end;
open_read(Fd) ->
    ?LOG_DEBUG(
       "~s: using existing file descriptor ~p for import",
       [?MODULE, Fd],
       #{domain => [khepri, import_export, ?MODULE]}),
    State = #{fd => Fd,
              max_chunk => ?IMPORT_APPROX_LIMIT_PER_CALL,
              location => 1},
    {ok, State}.

-spec read(ReadState) -> Ret when
      ReadState :: read_state(),
      Ret :: {ok, BackupItems, ReadState} | {error, any()},
      BackupItems :: [khepri_import_export:backup_item()].
  @private

read(#{fd := Fd} = State) ->
    StartPos = case file:position(Fd, cur) of
                   {ok, Pos} -> Pos;
                   _         -> undefined
               end,
    State1 = State#{start_pos => StartPos},
    do_read(State1, []).

-spec do_read(ReadState, BackupItems) -> Ret when
      ReadState :: read_state(),
      BackupItems :: [khepri_import_export:backup_item()],
      Ret :: {ok, BackupItems, ReadState} | {error, any()}.
  @private

do_read(
  #{fd := Fd,
    location := Location,
    start_pos := StartPos,
    max_chunk := MaxChunk} = State,
  BackupItems) ->
    case io:read(Fd, <<>>, Location) of
        {ok, BackupItem, EndLocation} ->
            ?LOG_DEBUG(
               "~s: importing: ~p",
               [?MODULE, BackupItem],
               #{domain => [khepri, import_export, ?MODULE]}),
            BackupItems1 = [BackupItem | BackupItems],
            State1 = State#{location => EndLocation},

            CurPos = case file:position(Fd, cur) of
                         {ok, Pos} -> Pos;
                         _         -> undefined
                     end,
            MustStop = (StartPos =:= undefined orelse
                        CurPos =:= undefined orelse
                        CurPos - StartPos >= MaxChunk),
            case MustStop of
                false -> do_read(State1, BackupItems1);
                true  -> {ok, lists:reverse(BackupItems1), State1}
            end;
        {eof, EndLocation} ->
            ?LOG_DEBUG(
               "~s: finished importing",
               [?MODULE],
               #{domain => [khepri, import_export, ?MODULE]}),
            State1 = State#{location => EndLocation},
            {ok, lists:reverse(BackupItems), State1};
        {error, _} = Error ->
            Error
    end.

-spec close_read(ReadState) -> Ret when
      ReadState :: read_state(),
      Ret :: ok | {error, any()}.
  @private

close_read(#{fd := Fd, filename := Filename}) ->
    ?LOG_DEBUG(
       "~s: closing file \"~ts\" after import",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    file:close(Fd);
close_read(_State) ->
    ?LOG_DEBUG(
       "~s: used existing file descriptor; nothing to close after import",
       [?MODULE],
       #{domain => [khepri, import_export, ?MODULE]}),
    ok.



 plain text as its external format.

 The exported file could be read using {@link file:consult/1} to get back
 the list of backup items.

 This callback module takes a filename or an opened file descriptor as its
 private data passed to {@link khepri:export/4} and {@link khepri:import/3}.

 Example:
 ```
 '''

 Content of `export.erl':
 ```
 {put,[stock,wood,<<"Mapple">>],{p_data,55},#{},#{}}.
 {put,[stock,wood,<<"Oak">>],{p_data,100},#{},#{}}.
 '''
 How many bytes to read from the file descriptor before `read/1' returns a
 list of backup items. `khepri_import_export' will call this module again to
 read the next batch.
 In bytes.
  This Source Code Form is subject to the terms of the Mozilla Public
  License , v. 2.0 . If a copy of the MPL was not distributed with this
  file , You can obtain one at /.
  Copyright © 2022 - 2023 VMware , Inc. or its affiliates .   All rights reserved .

  @doc Khepri import / export callback module using Erlang terms formatted as
  ok = khepri : put(StoreId , " /:stock/:wood / Oak " , 100 ) .
  ok = khepri : put(StoreId , " /:stock/:wood / Mapple " , 55 ) .
  ok =  : export(StoreId , khepri_export_erlang , " export.erl " ) .

-module(khepri_export_erlang).

-behavior(khepri_import_export).

-include_lib("kernel/include/logger.hrl").


-export([open_write/1,
         write/2,
         commit_write/1,
         abort_write/1,

         open_read/1,
         read/1,
         close_read/1]).

-type write_state() :: #{fd := file:io_device(),
                         filename => file:name_all()}.

-type read_state() :: #{fd := file:io_device(),
                        max_chunk := non_neg_integer(),
                        location := erl_anno:location(),
                        start_pos => integer(),
                        filename => file:name_all()}.

-spec open_write(Filename | Fd) -> Ret when
      Filename :: file:name_all(),
      Fd :: file:io_device(),
      Ret :: {ok, WriteState} | {error, any()},
      WriteState :: write_state().
  @private

open_write(Filename) when is_list(Filename) ->
    ?LOG_DEBUG(
       "~s: opening file \"~ts\" for export",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    case file:open(Filename, [write]) of
        {ok, Fd} ->
            State = #{fd => Fd,
                      filename => Filename},
            {ok, State};
        {error, _} = Error ->
            Error
    end;
open_write(Fd) ->
    ?LOG_DEBUG(
       "~s: using existing file descriptor ~p",
       [?MODULE, Fd],
       #{domain => [khepri, import_export, ?MODULE]}),
    State = #{fd => Fd},
    {ok, State}.

-spec write(WriteState, BackupItems) -> Ret when
      WriteState :: write_state(),
      BackupItems :: [khepri_import_export:backup_item()],
      Ret :: {ok, WriteState} | {error, any()}.
  @private

write(#{fd := Fd} = State, [BackupItem | Rest]) ->
    ?LOG_DEBUG(
       "~s: exporting: ~p",
       [?MODULE, BackupItem],
       #{domain => [khepri, import_export, ?MODULE]}),
    Binary = io_lib:format("~0p.~n", [BackupItem]),
    case file:write(Fd, Binary) of
        ok                 -> write(State, Rest);
        {error, _} = Error -> Error
    end;
write(State, []) ->
    ?LOG_DEBUG(
       "~s: finished exporting",
       [?MODULE],
       #{domain => [khepri, import_export, ?MODULE]}),
    {ok, State}.

-spec commit_write(WriteState) -> Ret when
      WriteState :: write_state(),
      Ret :: ok | {error, any()}.
  @private

commit_write(#{fd := Fd, filename := Filename}) ->
    ?LOG_DEBUG(
       "~s: closing file \"~ts\" after export",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    case file:close(Fd) of
        ok                 -> ok;
        {error, _} = Error -> Error
    end;
commit_write(_State) ->
    ?LOG_DEBUG(
       "~s: used existing file descriptor; nothing to close after export",
       [?MODULE],
       #{domain => [khepri, import_export, ?MODULE]}),
    ok.

-spec abort_write(WriteState) -> Ret when
      WriteState :: write_state(),
      Ret :: ok.
  @private

abort_write(#{fd := Fd, filename := Filename}) ->
    ?LOG_DEBUG(
       "~s: aborting export to file \"~ts\"",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    _ = file:close(Fd),
    _ = file:delete(Filename),
    ok;
abort_write(#{fd := Fd}) ->
    ?LOG_DEBUG(
       "~s: aborting export to existing file descriptor ~p",
       [?MODULE, Fd],
       #{domain => [khepri, import_export, ?MODULE]}),
    ok.

-spec open_read(Filename | Fd) -> Ret when
      Filename :: file:name_all(),
      Fd :: file:io_device(),
      Ret :: {ok, ReadState} | {error, any()},
      ReadState :: read_state().
  @private

open_read(Filename) when is_list(Filename) ->
    ?LOG_DEBUG(
       "~s: opening file \"~ts\" for import",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    case file:open(Filename, [read]) of
        {ok, Fd} ->
            _ = epp:set_encoding(Fd),
            State = #{fd => Fd,
                      max_chunk => ?IMPORT_APPROX_LIMIT_PER_CALL,
                      location => 1,
                      filename => Filename},
            {ok, State};
        {error, _} = Error ->
            Error
    end;
open_read(Fd) ->
    ?LOG_DEBUG(
       "~s: using existing file descriptor ~p for import",
       [?MODULE, Fd],
       #{domain => [khepri, import_export, ?MODULE]}),
    State = #{fd => Fd,
              max_chunk => ?IMPORT_APPROX_LIMIT_PER_CALL,
              location => 1},
    {ok, State}.

-spec read(ReadState) -> Ret when
      ReadState :: read_state(),
      Ret :: {ok, BackupItems, ReadState} | {error, any()},
      BackupItems :: [khepri_import_export:backup_item()].
  @private

read(#{fd := Fd} = State) ->
    StartPos = case file:position(Fd, cur) of
                   {ok, Pos} -> Pos;
                   _         -> undefined
               end,
    State1 = State#{start_pos => StartPos},
    do_read(State1, []).

-spec do_read(ReadState, BackupItems) -> Ret when
      ReadState :: read_state(),
      BackupItems :: [khepri_import_export:backup_item()],
      Ret :: {ok, BackupItems, ReadState} | {error, any()}.
  @private

do_read(
  #{fd := Fd,
    location := Location,
    start_pos := StartPos,
    max_chunk := MaxChunk} = State,
  BackupItems) ->
    case io:read(Fd, <<>>, Location) of
        {ok, BackupItem, EndLocation} ->
            ?LOG_DEBUG(
               "~s: importing: ~p",
               [?MODULE, BackupItem],
               #{domain => [khepri, import_export, ?MODULE]}),
            BackupItems1 = [BackupItem | BackupItems],
            State1 = State#{location => EndLocation},

            CurPos = case file:position(Fd, cur) of
                         {ok, Pos} -> Pos;
                         _         -> undefined
                     end,
            MustStop = (StartPos =:= undefined orelse
                        CurPos =:= undefined orelse
                        CurPos - StartPos >= MaxChunk),
            case MustStop of
                false -> do_read(State1, BackupItems1);
                true  -> {ok, lists:reverse(BackupItems1), State1}
            end;
        {eof, EndLocation} ->
            ?LOG_DEBUG(
               "~s: finished importing",
               [?MODULE],
               #{domain => [khepri, import_export, ?MODULE]}),
            State1 = State#{location => EndLocation},
            {ok, lists:reverse(BackupItems), State1};
        {error, _} = Error ->
            Error
    end.

-spec close_read(ReadState) -> Ret when
      ReadState :: read_state(),
      Ret :: ok | {error, any()}.
  @private

close_read(#{fd := Fd, filename := Filename}) ->
    ?LOG_DEBUG(
       "~s: closing file \"~ts\" after import",
       [?MODULE, Filename],
       #{domain => [khepri, import_export, ?MODULE]}),
    file:close(Fd);
close_read(_State) ->
    ?LOG_DEBUG(
       "~s: used existing file descriptor; nothing to close after import",
       [?MODULE],
       #{domain => [khepri, import_export, ?MODULE]}),
    ok.
