% Copyright (c) 2025 Onomondo ApS & sysmocom - s.f.m.c. GmbH. All rights reserved. % % SPDX-License-Identifier: AGPL-3.0-only % % Author: Philipp Maier / sysmocom - s.f.m.c. GmbH -module(mnesia_db_work). -include_lib("stdlib/include/qlc.hrl"). -include("mnesia_db_rest.hrl"). -include("mnesia_db_work.hrl"). % work functions, to be called by the eIM code (from inside) -export([fetch/2, pickup/2, update/2, bind/2, finish/3]). % debugging -export([dump/0]). % Start working on an order by creating a an entry in the work table and marking it's status as "work". After calling % this it is the callers responsibility to handle the work item and call rest_finish_order when the work is done. fetch(EidValue, Pid) -> % One process can only work on one work item at a time. The API user must call finish before the next work % item can be processed. If there is already a pending work item under the given PID, forcefully finish this work % item. TransPidExists = fun() -> Q = qlc:q([X || X <- mnesia:table(work), X#work.pid == Pid]), WorkPresent = qlc:e(Q), case WorkPresent of [] -> false; _ -> true end end, {atomic, PidExists} = mnesia:transaction(TransPidExists), case PidExists of true -> finish(Pid, [{[{procedureError, stuckOrder}]}], none); false -> ok end, % Read the next pending REST resource from the rest table and create a related work item. The work item is then % in progress. Trans = fun() -> Q = qlc:q([ X || X <- mnesia:table(rest), X#rest.eidValue == EidValue, X#rest.status == new, X#rest.facility =/= euicc ]), Rows = qlc:e(Q), case Rows of [Row | _] -> % Create an entry in the work table WorkRow = #work{ pid = Pid, resourceId = Row#rest.resourceId, transactionId = none, eidValue = Row#rest.eidValue, order = Row#rest.order, state = none }, case mnesia:write(WorkRow) of ok -> % We are now working on this order ok = mnesia_db_rest:trans_set_status(Row#rest.resourceId, work, [], none), Row; _ -> error end; [] -> none; _ -> error end end, {atomic, Result} = mnesia:transaction(Trans), case Result of {rest, _, Facility, _, Order, _, _, _, _} -> logger:info( "Work: fetching new work item,~nEidValue=~p, Pid=~p, Facility=~p,~nOrder=~p~n", [EidValue, Pid, Facility, Order] ), {Facility, Order}; none -> logger:info("Work: no work item in database,~nEidValue=~p, Pid=~p~n", [EidValue, Pid]), none; _ -> logger:error("Work: cannot fetch work item, database error,~nEidValue=~p, Pid=~p~n", [ EidValue, Pid ]), error end. % Bind a work item to a TransactionId. The transactionId has to be a unique identifier that can be used as a secondary % key to find a work item in the databse. The binding works in two directions. The work item is first searched by its % pid, when found, the transactionId is updated. In case the pid has become invalid, then the work item is searched % again by the transactionId and when found, the pid is updated. This function can be called any time after fetch % was called before. It can also be called multiple times. bind(Pid, TransactionId) -> % Transaction to update the TransactionId. This is the normal case. A work item starts without having a % TransactionId assigned. As soon as a (new) TransactionId becomes known, it is updated using this Transaction. TransUpdateTrnsId = fun() -> Q = qlc:q([X || X <- mnesia:table(work), X#work.pid == Pid]), Rows = qlc:e(Q), case Rows of [Row | _] -> mnesia:write(Row#work{transactionId = TransactionId}); [] -> none; _ -> error end end, % Transaction to update the PID. This is a corner case that comes into play in case the PID is lost (the % process/connection handling this work item has died). We then try to find the work item by the TransactionId % and update its PID. TransUpdatePid = fun() -> Q = qlc:q([X || X <- mnesia:table(work), X#work.transactionId == TransactionId]), Rows = qlc:e(Q), case Rows of [Row | _] -> mnesia:write(Row#work{pid = Pid}); [] -> none; _ -> error end end, {atomic, Result} = mnesia:transaction(TransUpdateTrnsId), case Result of ok -> logger:info("Work: bound work item to transactionId,~nPid=~p, TransactionId=~p~n", [ Pid, TransactionId ]), ok; none -> {atomic, UpdatePidResult} = mnesia:transaction(TransUpdatePid), case UpdatePidResult of ok -> logger:info("Work: bound work item to PID,~nPid=~p, TransactionId=~p~n", [ Pid, TransactionId ]), ok; none -> logger:error( "Work: cannot bind work item, transactionId nor PID found,~nPid=~p, TransactionId=~p~n", [Pid, TransactionId] ), error; _ -> logger:error( "Work: cannot bind work item, database error,~nPid=~p, TransactionId=~p, TransUpdatePid~n", [Pid, TransactionId] ), error end; _ -> logger:error( "Work: cannot bind work item, database error,~nPid=~p, TransactionId=~p, TransUpdateTrnsId~n", [Pid, TransactionId] ), error end. % Pickup a work item that is in progress. This function can be called any time after fetch was called % before. It can also be called multiple times. pickup(Pid, TransactionId) -> % In case a TransactionId is provied, bind the PID to this TransactionId, WorkBound = case TransactionId of none -> ok; _ -> bind(Pid, TransactionId) end, % Lookup the work state by the given PID case WorkBound of ok -> Trans = fun() -> Q = qlc:q([ {X#work.eidValue, X#work.order, X#work.state} || X <- mnesia:table(work), X#work.pid == Pid ]), qlc:e(Q) end, {atomic, Result} = mnesia:transaction(Trans), case Result of [{EidValue, Order, State} | _] -> {EidValue, Order, State}; [] -> logger:error( "Work: no work item found under specified Pid, already finished?, not fetched?,~nPid=~p~n", [Pid] ), none; _ -> logger:error("Work: cannot pick up work item, database error,~nPid=~p~n", [Pid]), error end; _ -> WorkBound end. % Update a work item that is in progress. This function updates the state (any user defined term) of the work item. % This function can be called any time after fetch was called before. It can also be called multiple times. update(Pid, State) -> Trans = fun() -> Q = qlc:q([X || X <- mnesia:table(work), X#work.pid == Pid]), Rows = qlc:e(Q), case Rows of [Row | _] -> mnesia:write(Row#work{state = State}); [] -> error; _ -> error end end, {atomic, Result} = mnesia:transaction(Trans), case Result of ok -> logger:info("Work: updating work item,~nPid=~p, State=~p~n", [Pid, State]), ok; _ -> logger:error("Work: cannot update, database error,~nPid=~p, State=~p~n", [Pid, State]), error end. % Finish an order that has been worked on. This removes the related entry from the work table and sets the status in % the rest table to "done". finish(Pid, Outcome, Debuginfo) -> Trans = fun() -> Q = qlc:q([X#work.resourceId || X <- mnesia:table(work), X#work.pid == Pid]), Rows = qlc:e(Q), case Rows of [] -> error; [ResourceId | _] -> Oid = {work, Pid}, ok = mnesia:delete(Oid), ok = mnesia_db_rest:trans_set_status(ResourceId, done, Outcome, Debuginfo) end end, {atomic, Result} = mnesia:transaction(Trans), case Result of ok -> logger:info("Work: finishing work item,~nPid=~p, Outcome=~p~n", [Pid, Outcome]), ok; _ -> logger:error( "Work: cannot finish work item, database error,~nPid=~p, Outcome=~p~n", [Pid, Outcome] ), error end. % Dump all currently pending work items (for debugging, to be called from console) dump() -> Trans = fun() -> Q = qlc:q([X || X <- mnesia:table(work)]), qlc:e(Q) end, {atomic, Result} = mnesia:transaction(Trans), Result.