
##! Functions to interface with broker's distributed data store.

%%{
#include "zeek/broker/Manager.h"
#include "zeek/broker/Store.h"
#include "zeek/broker/Data.h"
#include "zeek/Trigger.h"

static zeek::Broker::detail::StoreHandleVal* to_store_handle(zeek::Val* h)
	{
	auto rval = dynamic_cast<zeek::Broker::detail::StoreHandleVal*>(h);
	return rval && rval->have_store ? rval : nullptr;
	}
%%}

module Broker;

type Broker::QueryResult: record;

type Broker::BackendOptions: record;

enum BackendType %{
	MEMORY,
	SQLITE,
%}

function Broker::__create_master%(id: string, b: BackendType,
                                  options: BackendOptions &default = BackendOptions()%): opaque of Broker::Store
	%{
	zeek::Broker::Manager::ScriptScopeGuard ssg;
	auto name = id->CheckString();
	auto rval = broker_mgr->LookupStore(name);

	if ( rval )
		return ValPtr{zeek::NewRef{}, rval};

	auto e = static_cast<BifEnum::Broker::BackendType>(b->AsEnum());
	auto type = zeek::Broker::detail::to_backend_type(e);
	auto opts = zeek::Broker::detail::to_backend_options(type, options->AsRecordVal());

	ValPtr store{zeek::AdoptRef{}, broker_mgr->MakeMaster(name, type, std::move(opts))};

	if ( ! store )
		{
		zeek::emit_builtin_error(zeek::util::fmt("Could not create Broker master store '%s'", name));
		return make_intrusive<zeek::Broker::detail::StoreHandleVal>();
		}

	return store;
	%}

function Broker::__create_clone%(id: string, resync_interval: interval,
                                 stale_interval: interval,
                                 mutation_buffer_interval: interval%): opaque of Broker::Store
	%{
	zeek::Broker::Manager::ScriptScopeGuard ssg;
	auto name = id->CheckString();
	auto rval = broker_mgr->LookupStore(name);

	if ( rval )
		return ValPtr{zeek::NewRef{}, rval};

	ValPtr store {zeek::AdoptRef{},
	              broker_mgr->MakeClone(name, resync_interval,
	                                    stale_interval,
	                                    mutation_buffer_interval)};

	if ( ! store )
		{
		zeek::emit_builtin_error(zeek::util::fmt("Could not create clone of Broker store '%s'", name));
		return make_intrusive<zeek::Broker::detail::StoreHandleVal>();
		}

	return store;
	%}

function Broker::__is_closed%(h: opaque of Broker::Store%): bool
	%{
	zeek::Broker::Manager::ScriptScopeGuard ssg;
	auto handle = dynamic_cast<zeek::Broker::detail::StoreHandleVal*>(h);

	if ( ! handle )
		zeek::detail::emit_builtin_exception("invalid Broker store handle", h);

	if ( ! handle->have_store )
		return zeek::val_mgr->True();

	return zeek::val_mgr->Bool(! broker_mgr->LookupStore(handle->store.name()));
	%}

function Broker::__close%(h: opaque of Broker::Store%): bool
	%{
	zeek::Broker::Manager::ScriptScopeGuard ssg;
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return val_mgr->False();
		}

	if ( ! handle->have_store )
		{
		return val_mgr->False();
		}

	return zeek::val_mgr->Bool(broker_mgr->CloseStore(handle->store.name()));
	%}

function Broker::__store_name%(h: opaque of Broker::Store%): string
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		zeek::detail::emit_builtin_exception("invalid Broker store handle", h);

	return make_intrusive<zeek::StringVal>(handle->store.name());
	%}

function Broker::__exists%(h: opaque of Broker::Store,
                           k: any%): Broker::QueryResult
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::Broker::detail::query_result();
		}

	auto key = zeek::Broker::detail::val_to_data(k);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::Broker::detail::query_result();
		}

	auto trigger = frame->GetTrigger();

	if ( ! trigger )
		{
		zeek::emit_builtin_error("Broker queries can only be called inside when-condition");
		return zeek::Broker::detail::query_result();
		}

	auto timeout = trigger->TimeoutValue();

	if ( timeout < 0 )
		{
		zeek::emit_builtin_error("Broker queries must specify a timeout block");
		return zeek::Broker::detail::query_result();
		}

	frame->SetDelayed();
	trigger->Hold();

	auto cb = new zeek::Broker::detail::StoreQueryCallback(trigger, frame->GetTriggerAssoc(),
	                                               handle->store);
	auto req_id = handle->proxy.exists(std::move(*key));
	broker_mgr->TrackStoreQuery(handle, req_id, cb);

	return nullptr;
	%}

function Broker::__get%(h: opaque of Broker::Store,
                        k: any%): Broker::QueryResult
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::Broker::detail::query_result();
		}

	auto key = zeek::Broker::detail::val_to_data(k);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::Broker::detail::query_result();
		}

	auto trigger = frame->GetTrigger();

	if ( ! trigger )
		{
		zeek::emit_builtin_error("Broker queries can only be called inside when-condition");
		return zeek::Broker::detail::query_result();
		}

	auto timeout = trigger->TimeoutValue();

	if ( timeout < 0 )
		{
		zeek::emit_builtin_error("Broker queries must specify a timeout block");
		return zeek::Broker::detail::query_result();
		}

	frame->SetDelayed();
	trigger->Hold();

	auto cb = new zeek::Broker::detail::StoreQueryCallback(trigger, frame->GetTriggerAssoc(),
	                                               handle->store);
	auto req_id = handle->proxy.get(std::move(*key));
	broker_mgr->TrackStoreQuery(handle, req_id, cb);

	return nullptr;
	%}

function Broker::__put_unique%(h: opaque of Broker::Store,
                               k: any, v: any, e: interval%): Broker::QueryResult
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::Broker::detail::query_result();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto val = zeek::Broker::detail::val_to_data(v);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::Broker::detail::query_result();
		}

	if ( ! val )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for value argument", v);
		return zeek::Broker::detail::query_result();
		}

	auto trigger = frame->GetTrigger();

	if ( ! trigger )
		{
		zeek::emit_builtin_error("Broker queries can only be called inside when-condition");
		return zeek::Broker::detail::query_result();
		}

	auto timeout = trigger->TimeoutValue();

	if ( timeout < 0 )
		{
		zeek::emit_builtin_error("Broker queries must specify a timeout block");
		return zeek::Broker::detail::query_result();
		}

	frame->SetDelayed();
	trigger->Hold();

	auto cb = new zeek::Broker::detail::StoreQueryCallback(trigger, frame->GetTriggerAssoc(),
	                                               handle->store);

	auto req_id = handle->proxy.put_unique(std::move(*key), std::move(*val),
	                                       zeek::Broker::detail::convert_expiry(e));
	broker_mgr->TrackStoreQuery(handle, req_id, cb);

	return nullptr;
	%}

function Broker::__get_index_from_value%(h: opaque of Broker::Store,
                                         k: any, i: any%): Broker::QueryResult
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::Broker::detail::query_result();
		}

	auto key = zeek::Broker::detail::val_to_data(k);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::Broker::detail::query_result();
		}

	auto index = zeek::Broker::detail::val_to_data(i);

	if ( ! index )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for index argument", i);
		return zeek::Broker::detail::query_result();
		}

	auto trigger = frame->GetTrigger();

	if ( ! trigger )
		{
		zeek::emit_builtin_error("Broker queries can only be called inside when-condition");
		return zeek::Broker::detail::query_result();
		}

	auto timeout = trigger->TimeoutValue();

	if ( timeout < 0 )
		{
		zeek::emit_builtin_error("Broker queries must specify a timeout block");
		return zeek::Broker::detail::query_result();
		}

	frame->SetDelayed();
	trigger->Hold();

	auto cb = new zeek::Broker::detail::StoreQueryCallback(trigger, frame->GetTriggerAssoc(),
						       handle->store);
	auto req_id = handle->proxy.get_index_from_value(std::move(*key),
	                                                 std::move(*index));
	broker_mgr->TrackStoreQuery(handle, req_id, cb);

	return nullptr;
	%}

function Broker::__keys%(h: opaque of Broker::Store%): Broker::QueryResult
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::Broker::detail::query_result();
		}

	if ( ! handle->have_store )
		{
		zeek::emit_builtin_error("cannot retrieve keys from a closed store", h);
		return zeek::Broker::detail::query_result();
		}

	auto trigger = frame->GetTrigger();

	if ( ! trigger )
		{
		zeek::emit_builtin_error("Broker queries can only be called inside when-condition");
		return zeek::Broker::detail::query_result();
		}

	auto timeout = trigger->TimeoutValue();

	if ( timeout < 0 )
		{
		zeek::emit_builtin_error("Broker queries must specify a timeout block");
		return zeek::Broker::detail::query_result();
		}

	frame->SetDelayed();
	trigger->Hold();

	auto cb = new zeek::Broker::detail::StoreQueryCallback(trigger, frame->GetTriggerAssoc(),
	                                             handle->store);
	auto req_id = handle->proxy.keys();
	broker_mgr->TrackStoreQuery(handle, req_id, cb);

	return nullptr;
	%}

function Broker::__put%(h: opaque of Broker::Store,
                        k: any, v: any, e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto val = zeek::Broker::detail::val_to_data(v);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	if ( ! val )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for value argument", v);
		return zeek::val_mgr->False();
		}

	handle->store.put(std::move(*key), std::move(*val), zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__erase%(h: opaque of Broker::Store, k: any%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	handle->store.erase(std::move(*key));
	return zeek::val_mgr->True();
	%}

function Broker::__increment%(h: opaque of Broker::Store, k: any, a: any,
                              e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto amount = zeek::Broker::detail::val_to_data(a);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	if ( ! amount )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for amount argument", a);
		return zeek::val_mgr->False();
		}

	handle->store.increment(std::move(*key), std::move(*amount),
	                        zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__decrement%(h: opaque of Broker::Store, k: any, a: any,
                              e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto amount = zeek::Broker::detail::val_to_data(a);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	if ( ! amount )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for amount argument", a);
		return zeek::val_mgr->False();
		}

	handle->store.decrement(std::move(*key), std::move(*amount), zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__append%(h: opaque of Broker::Store, k: any, s: any,
                           e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto str = zeek::Broker::detail::val_to_data(s);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	if ( ! str )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for str argument", s);
		return zeek::val_mgr->False();
		}

	handle->store.append(std::move(*key), std::move(*str), zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__insert_into_set%(h: opaque of Broker::Store, k: any, i: any,
                                    e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto idx = zeek::Broker::detail::val_to_data(i);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	if ( ! idx )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for index argument", i);
		return zeek::val_mgr->False();
		}

	handle->store.insert_into(std::move(*key), std::move(*idx),
	                          zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__insert_into_table%(h: opaque of Broker::Store, k: any,
                                      i: any, v: any, e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto idx = zeek::Broker::detail::val_to_data(i);
	auto val = zeek::Broker::detail::val_to_data(v);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	if ( ! idx )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for index argument", i);
		return zeek::val_mgr->False();
		}

	if ( ! val )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for value argument", v);
		return zeek::val_mgr->False();
		}

	handle->store.insert_into(std::move(*key), std::move(*idx),
	                          std::move(*val), zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__remove_from%(h: opaque of Broker::Store, k: any, i: any,
                                e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto idx = zeek::Broker::detail::val_to_data(i);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	if ( ! idx )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for index argument", i);
		return zeek::val_mgr->False();
		}

	handle->store.remove_from(std::move(*key), std::move(*idx),
	                          zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__push%(h: opaque of Broker::Store, k: any, v: any,
                         e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);
	auto val = zeek::Broker::detail::val_to_data(v);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	if ( ! val )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for value argument", v);
		return zeek::val_mgr->False();
		}

	handle->store.push(std::move(*key), std::move(*val), zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__pop%(h: opaque of Broker::Store, k: any, e: interval%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	auto key = zeek::Broker::detail::val_to_data(k);

	if ( ! key )
		{
		zeek::emit_builtin_error("invalid Broker data conversion for key argument", k);
		return zeek::val_mgr->False();
		}

	handle->store.pop(*key, zeek::Broker::detail::convert_expiry(e));
	return zeek::val_mgr->True();
	%}

function Broker::__clear%(h: opaque of Broker::Store%): bool
	%{
	auto handle = to_store_handle(h);

	if ( ! handle )
		{
		zeek::emit_builtin_error("invalid Broker store handle", h);
		return zeek::val_mgr->False();
		}

	handle->store.clear();
	return zeek::val_mgr->True();
	%}
