odi

%%% Copyright (C) 2013 - Aleksandr Mescheryakov.  All rights reserved.

-module(odi).

-export([connect/4,
         close/1,
         db_open/5,
         db_create/5,
         db_close/1,
         db_exist/3,
         db_reload/1,
         db_delete/3,
         db_size/1,
         db_countrecords/1,
         datacluster_add/3,
         datacluster_remove/2,
         record_create/5,
         record_load/4,
         record_update/6,
         record_update/7,
         record_delete/4,
         query/4,
         query/5,
         command/2,
         command/4,
         script/3,
         script/5,
         live_query/3,
         live_query/4,
         tx_commit/4]).

-include("../include/odi.hrl").

-define(DEFAULT_TIMEOUT, infinity).

%% -- client interface --

-type rid()::{ClusterId::integer(), ClusterPosition::integer()}.
%% A record ID


-type record()::{Class::string(), Data::#{string() => any()}}.  %% TODO: more details
%% A record


-type record_type()::raw|flat|document.


-type mode()::sync|async|no_response.


-type fetched_record()::
    {Key::true|rid(), document, Version::integer(), Class::string(), Data::map()} |
    {Key::true|rid(), raw, Version::integer(), Class::raw, Data::binary()}.
%% The result of a record fetch


-type tx_operation()::
    {update, Rid::rid(), RecordType::record_type(), Version::integer(), UpdateContent::boolean(), Record::record()} |
    {delete, Rid::rid(), RecordType::record_type(), Version::integer()} |
    {create, Rid::rid(), RecordType::record_type(), Record::record()}.
%% A transaction operation


-type error()::{error, [{ExceptionClass::binary(), ExceptionMessage::binary()}]}.
%% Result type for an error


-export_type([tx_operation/0]).


-spec connect(Host::string(), Username::string(), Password::string(),
              Opts::[{timeout, Timeout::integer()} | {port, Port::integer()}]) -> {ok, Con::pid()} | error().
%% @doc Connect to a server.
%%
%% This is the first operation requested by the client when it needs to work with the server instance without openning a database.
connect(Host, Username, Password, Opts) ->
    {ok, C} = odi_sock:start_link(),
    {call(C, {connect, Host, Username, Password, Opts}), C}.


-spec db_open(Host::string(), Dbname::string(), Username::string(), Password::string(),
              Opts::[{timeout, Timeout::integer()} | {port, Port::integer()}]) ->
      {Clusters::[{ClusterName::string(), ClusterId::integer()}], Con::pid()} | error().
%% @doc Open a database
%%
%% This is the first operation the client should call. It opens a database on the remote OrientDB Server.
db_open(Host, DBName, Username, Password, Opts) ->
    {ok, C} = odi_sock:start_link(),
    {call(C, {db_open, Host, DBName, Username, Password, Opts}), C}.


-spec db_create(C::pid(), DatabaseName::string(), DatabaseType::string(), StorageType::string(),
                BackupPath::string()) -> ok | error().
%% @doc Creates a database in the remote OrientDB server instance.
%%
%% Works in connect-mode.
db_create(C, DatabaseName, DatabaseType, StorageType, BackupPath) ->
    call(C, {db_create, DatabaseName, DatabaseType, StorageType, BackupPath}).


-spec db_close(C::pid()) -> {stop, closed}.
%% @doc Closes the database and the network connection to the OrientDB Server instance.
db_close(C) ->
    call(C, {db_close}).


-spec db_exist(C::pid(), DatabaseName::string(), StorageType::string()) -> boolean() | error().
%% @doc Asks if a database exists in the OrientDB Server instance.
%%
%% It returns true (non-zero) or false (zero). Works in connect-mode.
db_exist(C, DatabaseName, StorageType) ->
    call(C, {db_exist, DatabaseName, StorageType}).


-spec db_reload(C::pid()) -> [{ClusterName::string(), ClusterId::integer()}] | error().
%% @doc Reloads database information.
db_reload(C) ->
    call(C, {db_reload}).


-spec db_delete(C::pid(), DatabaseName::string(), StorageType::string()) -> ok | error().
%% @doc Removes a database from the OrientDB Server instance.
%%
%% Works in connect-mode.
db_delete(C, DatabaseName, ServerStorageType) ->
    call(C, {db_delete, DatabaseName, ServerStorageType}).


-spec db_size(C::pid()) -> integer() | error().
%% @doc Returns size of the opened database.
db_size(C) ->
    call(C, {db_size}).


-spec db_countrecords(C::pid()) -> integer() | error().
%% @doc Asks for the number of records in a database in the OrientDB Server instance.
db_countrecords(C) ->
    call(C, {db_countrecords}).


-spec datacluster_add(C::pid(), Name::string(), ClusterId::integer()) -> integer() | error().
%% @doc Add a new data cluster.
datacluster_add(C, Name, ClusterId) ->
    call(C, {datacluster_add, Name, ClusterId}).


-spec datacluster_remove(C::pid(), ClusterId::integer()) -> boolean() | error().
%% @doc Remove a cluster.
datacluster_remove(C, ClusterId) ->
    call(C, {datacluster_remove, ClusterId}).


-spec record_create(C::pid(), ClusterId::integer(), RecordContent::binary(), RecordType::record_type(), Mode::mode()) ->
    {ClusterId::integer(), ClusterPosition::integer(), RecordVersion::integer(),
     [{Uuid::integer(), UpdatedFileId::integer(), UpdatePageIndex::integer(), UpdatedPageOffset::integer()}]} |
    error().
%% @doc Create a new record.
%%
%% Returns the position in the cluster of the new record.
record_create(C, ClusterId, {Class, Fields}, document, Mode) ->
    {RecordBin, _} = odi_record_binary:encode_record(Class, Fields, 0),
    call(C, {record_create, ClusterId, RecordBin, document, Mode});
record_create(C, ClusterId, RecordContent, RecordType, Mode) ->
    call(C, {record_create, ClusterId, RecordContent, RecordType, Mode}).


-spec record_load(C::pid(), {ClusterId::integer(), RecordPosition::integer()}, FetchPlan::string()|default,
    IgnoreCache::boolean()) -> [fetched_record()] | error().
%% @doc Load a record by RecordID, according to a fetch plan.
record_load(C, {ClusterId, RecordPosition}, FetchPlan, IgnoreCache) ->
    FetchPlan2 = case FetchPlan of default -> "*:1"; _ -> FetchPlan end,
    call(C, {record_load, ClusterId, RecordPosition, FetchPlan2, IgnoreCache}).


-spec record_update(C::pid(), RID::rid(), UpdateContent::boolean(), Record::record(),
                    OldRecordVersion::integer(), Mode::mode()) ->
    {RecordVersion::integer(),
     [{Uuid::integer(), UpdatedFileId::integer(), UpdatePageIndex::integer(), UpdatedPageOffset::integer()}]} | error().
%% @doc Update a record. Returns the new record's version.
record_update(C, RID, UpdateContent, {Class, Fields}, OldRecordVersion, Mode) ->
    {RecordBin, _} = odi_record_binary:encode_record(Class, Fields, 0),
    record_update(C, RID, UpdateContent, RecordBin, OldRecordVersion, document, Mode).


-spec record_update(C::pid(), RID::rid(), UpdateContent::boolean(), Record::binary(),
                    OldRecordVersion ::integer(), RecordType::record_type(), Mode::mode()) ->
    {OldRecordVersion ::integer(),
     [{Uuid::integer(), UpdatedFileId::integer(), UpdatePageIndex::integer(), UpdatedPageOffset::integer()}]} | error().
%% @doc Update a record. Returns the new record's version.
record_update(C, {ClusterId, ClusterPosition}, UpdateContent, RecordContent, OldRecordVersion, RecordType,
              Mode) when is_binary(RecordContent) ->
    call(C, {record_update, ClusterId, ClusterPosition, UpdateContent, RecordContent, OldRecordVersion, RecordType, Mode}).


%% @doc Delete a record by its RecordID.
-spec record_delete(C::pid(), RID::rid(), RecordVersion::integer(), Mode::mode()) -> boolean() | error().
%%
%% During the optimistic transaction the record will be deleted only if the versions match.
%% Returns true if has been deleted otherwise false.
record_delete(C, {ClusterId, ClusterPosition}, RecordVersion, Mode) ->
    call(C, {record_delete, ClusterId, ClusterPosition, RecordVersion, Mode}).


-spec query(C::pid(), SQL::string(), Limit::integer(), FetchPlan::string()|default) ->
    {Results::[fetched_record()], Cached::[fetched_record()]} | error().
%% @doc SQL query (SELECT or TRAVERSE).
query(C, SQL, Limit, FetchPlan) ->
    query(C, SQL, Limit, FetchPlan, null).


-spec query(C::pid(), SQL::string(), Limit::integer(), FetchPlan::string()|default,
            Params::#{string()=>any()} | null) ->
    {Results::[fetched_record()], Cached::[fetched_record()]} | error().
%% @doc SQL query with parameters (SELECT or TRAVERSE).
query(C, SQL, Limit, FetchPlan, Params) ->
    FetchPlan2 = case FetchPlan of default -> "*:1"; _ -> FetchPlan end,
    call(C, {command, {select, SQL, Limit, FetchPlan2, Params}, sync}).


-spec command(C::pid(), SQL::string()) -> [fetched_record()] | error().
%% @doc Syncronous SQL command.
command(C, SQL) ->
    command(C, SQL, null, null).


-spec command(C::pid(), SQL::string(), SimpleParams::#{string() => any()} | null,
              ComplexParams::#{string() => {embedded_list, []}} | null) -> [fetched_record()] | error().
%% @doc Syncronous SQL command with parameters.
command(C, SQL, SimpleParams, ComplexParams) ->
    {Results, []} = call(C, {command, {command, SQL, SimpleParams, ComplexParams}, sync}),
    Results.


-spec script(C::pid(), Language::string(), Code::string()) ->
    {[fetched_record()], [fetched_record()]} | error().
%% @doc Syncronous script.
script(C, Language, Code) ->
    script(C, Language, Code, null, null).


-spec script(C::pid(), Language::string(), Code::string(), SimpleParams::#{string() => any()} | null,
             ComplexParams::#{string() => {embedded_list, []}} | null) ->
    {[fetched_record()], [fetched_record()]} | error().
%% @doc Syncronous script with parameters.
script(C, Language, Code, SimpleParams, ComplexParams) ->
    call(C, {command, {script, Language, Code, SimpleParams, ComplexParams}, sync}).


-spec live_query(C::pid(), SQL::string(),
                 CallBack::fun((live | live_unsubscription,
                                {loaded|updated|deleted|created, fetched_record()} | {}) -> any())) ->
    {ok, Token::integer()} | error().
%% @doc Live SELECT query.
live_query(C, SQL, CallBack) ->
    live_query(C, SQL, null, CallBack).


-spec live_query(C::pid(), SQL::string(), Params::#{string() => any()} | null,
    CallBack::fun((live, {loaded|updated|deleted|created, fetched_record()}) -> any())) ->
    {ok, Token::integer()} | error().
%% @doc Live SELECT query with parameters.
live_query(C, SQL, Params, CallBack) ->
    case call(C, {command, {live, SQL, -1, "", Params, CallBack}, live}) of
        {[{_Rid, document, 0, _Class, #{"token" := {integer, Token}}}],[]} -> {ok, Token};
        Other -> Other
    end.


-spec tx_commit(C::pid(), TxId::integer(), UsingLog::boolean(), Operations::[tx_operation()]) ->
  {CreatedRecords::[{ClientSpecifiedRid::rid(), ActualRid::rid()}],
   UpdatedRecords::[{UpdatedRid::rid(), NewRecordVersion::integer()}],
   CollectionChanges::[{Uuid::number(), UpdatedFileId::number(), UpdatedPageIndex::number(),
                        UpdatedPageOffset ::number()}]} | error().
%% @doc Commits a transaction.
%%
%% This operation flushes all the given changes to the server side in a single transaction.
tx_commit(C, TxId, UsingTxLog, Operations) ->
    call(C, {tx_commit, TxId, UsingTxLog, lists:map(fun encode_operation_record/1, Operations)}).


-spec close(C::pid()) -> ok.
%% @doc Close the connection.
close(C) ->
    odi_sock:close(C).



%% -- internal functions --

call(C, Command) ->
    case gen_server:call(C, Command, infinity) of
        Error = {error, _} -> Error;
        {R} -> R;
        R -> R
    end.


encode_operation_record({update, Rid, RecordType, Version, UpdateContent,
                         {Class, Fields}}) ->
    {RecordBin, _} = odi_record_binary:encode_record(Class, Fields, 0),
    {update, Rid, RecordType, Version, UpdateContent,
        RecordBin};
encode_operation_record({delete, _Rid, _RecordType, _Version} = Record) ->
  Record;
encode_operation_record({create, Rid, RecordType, {Class, Fields}}) ->
    {RecordBin, _} = odi_record_binary:encode_record(Class, Fields, 0),
    {create, Rid, RecordType, RecordBin}.