-module(temporal_sdk_scope). % elp:ignore W0012 W0040 -moduledoc false. -export([ init_ctx/1, get_members/1, get_local_members/1, start/1, build_ets_ids/2, ets_scope_id/2, pg_scope_id/2 ]). -spec init_ctx(temporal_sdk_api:context()) -> temporal_sdk_api:context(). init_ctx( #{ worker_opts := #{namespace := Namespace}, task_opts := #{workflow_id := WorkflowId, run_id := RunId}, workflow_scope := {Scope, ShardSize} } = ApiContext ) -> ShardId = shard_id(ShardSize, Namespace, WorkflowId, RunId), ApiContext#{ workflow_scope := #{ scope => Scope, shard_size => ShardSize, shard_id => ShardId, ets_scope_id => ets_scope_id(Scope, ShardId), pg_scope_id => pg_scope_id(Scope, ShardId) } }. -spec get_members(ApiContext :: temporal_sdk_api:context()) -> ScopeMembers :: [pid()]. get_members(#{ enable_single_distributed_workflow_execution := EnableSingleDistWorkflowExecution, worker_opts := #{namespace := Namespace}, task_opts := #{workflow_id := WorkflowId}, workflow_scope := #{pg_scope_id := PgScopeId} }) -> case EnableSingleDistWorkflowExecution of true -> pg:get_members(PgScopeId, {Namespace, WorkflowId}); false -> pg:get_local_members(PgScopeId, {Namespace, WorkflowId}) end. -spec get_local_members(ApiContext :: temporal_sdk_api:context()) -> ScopeMembers :: [pid()]. get_local_members(#{ worker_opts := #{namespace := Namespace}, task_opts := #{workflow_id := WorkflowId}, workflow_scope := #{pg_scope_id := PgScopeId} }) -> pg:get_local_members(PgScopeId, {Namespace, WorkflowId}). -spec start(ApiContext :: temporal_sdk_api:context()) -> ok. start(#{ worker_opts := #{namespace := Namespace}, task_opts := #{workflow_id := WorkflowId}, workflow_scope := #{ets_scope_id := EtsScopeId, pg_scope_id := PgScopeId} }) -> temporal_sdk_utils_recycled_int:new(EtsScopeId), pg:join(PgScopeId, {Namespace, WorkflowId}, self()). build_ets_ids(#{workflow_scope := #{scope := Scope, shard_id := ShardId}}, TableId) -> History = temporal_sdk_utils_path:atom_path([?MODULE, ets, Scope, ShardId, TableId, history]), Index = temporal_sdk_utils_path:atom_path([?MODULE, ets, Scope, ShardId, TableId, index]), {History, Index}. ets_scope_id(Scope, ShardId) -> temporal_sdk_utils_path:atom_path([?MODULE, ets, Scope, ShardId]). pg_scope_id(Scope, ShardId) -> temporal_sdk_utils_path:atom_path([?MODULE, pg, Scope, ShardId]). shard_id(ShardSize, Namespace, WorkflowId, RunId) -> erlang:adler32([Namespace, WorkflowId, RunId]) rem ShardSize + 1.