%% @doc Bounded exact-user-scope version-2 memory adapter. %% %% Each `{user, App, User}' principal is assigned one supervised adapter %% process. The default is volatile ETS; `adk_memory_mnesia' may be selected %% for durable storage while retaining independent per-scope call execution. -module(adk_memory_sharded). -behaviour(adk_memory_service). -export([start_link/1, stop/1, capabilities/1, status/1, add_entry/4, add_entry/5, add_events/5, add_events/6, add_session_to_memory/5, search/4, search/5, delete_entry/3, delete_entry/4, delete_session/3, delete_session/4, delete_user/2, delete_user/3]). -define(DEFAULT_TIMEOUT_MS, 5000). -spec start_link(map()) -> {ok, adk_scope_shard_router:handle()} | {error, term()}. start_link(Config) when is_map(Config) -> Allowed = [adapter, adapter_config, max_active_scopes, max_router_queue], Unknown = lists:sort(maps:keys(maps:without(Allowed, Config))), Adapter = maps:get(adapter, Config, adk_memory_ets), AdapterConfig = maps:get(adapter_config, Config, #{}), RouterOptions = maps:with( [max_active_scopes, max_router_queue], Config), case {Unknown, is_atom(Adapter), is_map(AdapterConfig)} of {[], true, true} -> adk_scope_shard_router:start_link( memory, Adapter, AdapterConfig, RouterOptions); {[_ | _], _, _} -> {error, {invalid_memory_sharded_config, {unknown_keys, Unknown}}}; {[], false, _} -> {error, {invalid_memory_sharded_config, adapter}}; {[], _, false} -> {error, {invalid_memory_sharded_config, adapter_config}} end; start_link(_Config) -> {error, invalid_memory_sharded_config}. -spec stop(term()) -> ok | {error, term()}. stop(Handle) -> adk_scope_shard_router:stop(Handle). -spec capabilities(term()) -> map() | {error, term()}. capabilities(Handle) -> case adk_scope_shard_router:capabilities(Handle) of {ok, Capabilities} -> Capabilities; {error, _} = Error -> Error end. -spec status(term()) -> {ok, map()} | {error, term()}. status(Handle) -> adk_scope_shard_router:status(Handle). -spec add_entry(term(), adk_memory_service:scope(), adk_memory_service:entry_input(), map()) -> {ok, adk_memory_service:entry()} | {error, term()}. add_entry(Handle, Scope, Input, Options) -> add_entry(Handle, Scope, Input, Options, #{}). -spec add_entry(term(), adk_memory_service:scope(), adk_memory_service:entry_input(), map(), adk_memory_service:call_options()) -> {ok, adk_memory_service:entry()} | {error, term()}. add_entry(Handle, Scope, Input, Options, CallOptions) -> deadline_call( Handle, Scope, CallOptions, fun(Adapter, Worker, Adjusted) -> Adapter:add_entry(Worker, Scope, Input, Options, Adjusted) end). -spec add_events(term(), adk_memory_service:scope(), binary(), [adk_event:event()], map()) -> {ok, adk_memory_service:ingest_result()} | {error, term()}. add_events(Handle, Scope, SessionId, Events, Options) -> add_events(Handle, Scope, SessionId, Events, Options, #{}). -spec add_events(term(), adk_memory_service:scope(), binary(), [adk_event:event()], map(), adk_memory_service:call_options()) -> {ok, adk_memory_service:ingest_result()} | {error, term()}. add_events(Handle, Scope, SessionId, Events, Options, CallOptions) -> deadline_call( Handle, Scope, CallOptions, fun(Adapter, Worker, Adjusted) -> Adapter:add_events(Worker, Scope, SessionId, Events, Options, Adjusted) end). -spec add_session_to_memory(term(), adk_memory_service:scope(), binary(), [adk_event:event()], map()) -> {ok, adk_memory_service:ingest_result()} | {error, term()}. add_session_to_memory(Handle, Scope, SessionId, Events, Options) -> add_events(Handle, Scope, SessionId, Events, Options). -spec search(term(), adk_memory_service:scope(), adk_memory_service:query(), map()) -> {ok, [adk_memory_service:hit()]} | {error, term()}. search(Handle, Scope, Query, Options) -> search(Handle, Scope, Query, Options, #{}). -spec search(term(), adk_memory_service:scope(), adk_memory_service:query(), map(), adk_memory_service:call_options()) -> {ok, [adk_memory_service:hit()]} | {error, term()}. search(Handle, Scope, Query, Options, CallOptions) -> deadline_call( Handle, Scope, CallOptions, fun(Adapter, Worker, Adjusted) -> Adapter:search(Worker, Scope, Query, Options, Adjusted) end). -spec delete_entry(term(), adk_memory_service:scope(), binary()) -> ok | {error, term()}. delete_entry(Handle, Scope, Id) -> delete_entry(Handle, Scope, Id, #{}). -spec delete_entry(term(), adk_memory_service:scope(), binary(), adk_memory_service:call_options()) -> ok | {error, term()}. delete_entry(Handle, Scope, Id, CallOptions) -> deadline_call( Handle, Scope, CallOptions, fun(Adapter, Worker, Adjusted) -> Adapter:delete_entry(Worker, Scope, Id, Adjusted) end). -spec delete_session(term(), adk_memory_service:scope(), binary()) -> ok | {error, term()}. delete_session(Handle, Scope, SessionId) -> delete_session(Handle, Scope, SessionId, #{}). -spec delete_session(term(), adk_memory_service:scope(), binary(), adk_memory_service:call_options()) -> ok | {error, term()}. delete_session(Handle, Scope, SessionId, CallOptions) -> deadline_call( Handle, Scope, CallOptions, fun(Adapter, Worker, Adjusted) -> Adapter:delete_session(Worker, Scope, SessionId, Adjusted) end). -spec delete_user(term(), adk_memory_service:scope()) -> ok | {error, term()}. delete_user(Handle, Scope) -> delete_user(Handle, Scope, #{}). -spec delete_user(term(), adk_memory_service:scope(), adk_memory_service:call_options()) -> ok | {error, term()}. delete_user(Handle, Scope, CallOptions) -> deadline_call( Handle, Scope, CallOptions, fun(Adapter, Worker, Adjusted) -> Adapter:delete_user(Worker, Scope, Adjusted) end). deadline_call(Handle, Scope, CallOptions, Function) -> case adk_memory_contract:validate_scope(Scope) of {ok, CanonicalScope} -> case validate_call_options(CallOptions) of {ok, Timeout, Deadline} -> case adk_scope_shard_router:resolve( Handle, CanonicalScope, Timeout) of {ok, Adapter, Worker} -> case remaining_timeout(Deadline) of 0 -> {error, timeout}; Remaining -> safe_apply( fun() -> Function(Adapter, Worker, #{timeout_ms => Remaining}) end) end; {error, _} = Error -> Error end; {error, _} = Error -> Error end; {error, _} = Error -> Error end. validate_call_options(CallOptions) when is_map(CallOptions) -> Unknown = lists:sort( maps:keys(maps:without([timeout_ms], CallOptions))), Timeout = maps:get(timeout_ms, CallOptions, ?DEFAULT_TIMEOUT_MS), case {Unknown, is_integer(Timeout) andalso Timeout > 0 andalso Timeout =< 60000} of {[], true} -> {ok, Timeout, erlang:monotonic_time(millisecond) + Timeout}; {[_ | _], _} -> {error, {invalid_memory_call_options, {unknown_keys, Unknown}}}; {[], false} -> {error, invalid_memory_call_timeout} end; validate_call_options(_CallOptions) -> {error, invalid_memory_call_options}. remaining_timeout(Deadline) -> erlang:max(0, Deadline - erlang:monotonic_time(millisecond)). safe_apply(Function) -> try Function() of Reply -> Reply catch _:_ -> {error, memory_shard_adapter_failed} end.