unfund sync (#971)

* remove device from config service when it can no longer be found on console

* eui and skf are taken care of in device update

* fix test now that skf defaults to enabled

* don't upload skf of unfunded org

* add/remove eui for funding/unfunding

* add/remove skfs on ws message

* initialize with the zero_dc endpoint

* remove intitial reconcile in favor of default no balance org

* add cli for refetching unfunded

* update table

* fix calling cli

* fix typo, eunit tests, wait until devaddr work has devaddrs

* typo

* don't remove unfunded org more than once

it takes a while and can really build up

* classic of classics

* allow multiple attempts to remove unfunded orgs.

* returns boolean

* add option to wait for reconcile on startup

we want to wait for the unfunded list of orgs to be fetched, but that
worker sits under another superviser before this worker, and we can't be
sure everything is ready to go by the time we get here. So we wait, just
a little, but we wait.

* better add/remove on update

consider EUI and SKF separately.

* skf being enabled is not a given

* move devaddr_int_nwk_key to router_device

* always remove when requested, only add when funded

* updating devices will handle adding eui for new unjoined

this also gives us a copy of the device saved, so we can reactivate
unjoined devices on refunding an org

* don't drop devaddrs when the device is updated and nothing changes

* remove devaddrs and nwk_s_keys from config service when app_eui/dev_eui
is changed

* remove device old EUI if they change

* make to send device EUI if we think it has changed and the device is
still active

* update clause size

* fix dialyzer

* fix keys argument to make_skf_removes

* eat add message from device intiialization

* go back to remaining pairs

we need to remove any possible combinations that may have arisen from
the devaddrs and keys.

* make skf delete match config-service

ignore max_copies value

* upgrade proto to latest

config/test.config

@@ -12,6 +12,8 @@
             host => "localhost",
             port => 8085,
             devaddr_enabled => "true",
+            eui_enabled => "true",
+            skf_enabled => "true",
             route_id => "test-route-id"
         }},
         {config_service_max_timeout_attempt, 5}

rebar.lock

@@ -90,7 +90,7 @@
  {<<"hackney">>,{pkg,<<"hackney">>,<<"1.18.1">>},0},
  {<<"helium_proto">>,
   {git,"https://github.com/helium/proto.git",
-       {ref,"91cd0c3953e8e233a86e6c67120a6df9b9b1cb40"}},
+       {ref,"c10a73fc89d895f439bc82efbbc8cd11ed880546"}},
   0},
  {<<"hpack">>,
   {git,"https://github.com/novalabsxyz/hpack.git",

src/apis/router_console_api.erl

@@ -22,6 +22,7 @@
     get_downlink_url/2,
     get_org/1,
     get_orgs/0,
+    get_unfunded_org_ids/0,
     org_manual_update_router_dc/2,
     organizations_burned/3,
     evict_org_from_cache/1,
@@ -167,6 +168,11 @@ get_orgs() ->
     {Endpoint, Token} = token_lookup(),
     get_orgs(Endpoint, Token, [], undefined).
 
+-spec get_unfunded_org_ids() -> {ok, list()} | {error, any()}.
+get_unfunded_org_ids() ->
+    {Endpoint, Token} = token_lookup(),
+    get_unfunded_orgs(Endpoint, Token).
+
 -spec org_manual_update_router_dc(OrgID :: binary(), Balance :: non_neg_integer()) ->
     ok | {error, any()}.
 org_manual_update_router_dc(OrgID, Balance) ->
@@ -478,6 +484,15 @@ init(Args) ->
     Token = get_token(Endpoint, Secret),
     ok = token_insert(Endpoint, DownlinkEndpoint, Token),
     {ok, DB, CF} = router_db:get_devices(),
+
+    case get_unfunded_org_ids() of
+        {ok, OrgIDs} ->
+            lager:info("inserting ~p unfunded orgs", [erlang:length(OrgIDs)]),
+            [router_console_dc_tracker:add_unfunded(OrgID) || OrgID <- OrgIDs];
+        {error, Err} ->
+            lager:error("fetching unfunded orgs failed: ~p", [Err])
+    end,
+
     _ = erlang:send_after(?TOKEN_CACHE_TIME, self(), refresh_token),
     {ok, P} = load_pending_burns(DB),
     Inflight = maybe_spawn_pending_burns(P, []),
@@ -657,6 +672,28 @@ get_orgs(Endpoint, Token, AccOrgs, ResourceID) ->
             {error, Other}
     end.
 
+-spec get_unfunded_orgs(Endpoint :: binary(), Token :: binary()) -> {ok, list()} | {error, any()}.
+get_unfunded_orgs(Endpoint, Token) ->
+    Url = <<Endpoint/binary, "/api/router/organizations/zero_dc">>,
+    Opts = [
+        with_body,
+        {pool, ?POOL},
+        {connect_timeout, timer:seconds(2)},
+        {recv_timeout, timer:seconds(2)}
+    ],
+    Start = erlang:system_time(millisecond),
+    case hackney:get(Url, [{<<"Authorization">>, <<"Bearer ", Token/binary>>}], <<>>, Opts) of
+        {ok, 200, _Headers, Body} ->
+            End = erlang:system_time(millisecond),
+            ok = router_metrics:console_api_observe(get_unfunded_orgs, ok, End - Start),
+            #{<<"data">> := OrgIDs} = jsx:decode(Body, [return_maps]),
+            {ok, OrgIDs};
+        Other ->
+            End = erlang:system_time(millisecond),
+            ok = router_metrics:console_api_observe(get_unfunded_orgs, error, End - Start),
+            {error, Other}
+    end.
+
 -spec convert_channel(Device :: router_device:device(), Pid :: pid(), JSONChannel :: map()) ->
     false | {true, router_channel:channel()}.
 convert_channel(Device, Pid, #{<<"type">> := <<"http">>} = JSONChannel) ->

src/apis/router_console_dc_tracker.erl

@@ -15,7 +15,12 @@
     refill/3,
     has_enough_dc/2,
     charge/2,
-    current_balance/1
+    current_balance/1,
+    %%
+    add_unfunded/1,
+    remove_unfunded/1,
+    list_unfunded/0,
+    reset_unfunded_from_api/0
 ]).
 
 %% ------------------------------------------------------------------
@@ -51,6 +56,7 @@
 
 -define(SERVER, ?MODULE).
 -define(ETS, router_console_dc_tracker_ets).
+-define(UNFUNDED_ETS, router_console_dc_tracker_unfunded_ets).
 
 -record(state, {
     pubkey_bin :: libp2p_crypto:pubkey_bin()
@@ -66,10 +72,35 @@ start_link(Args) ->
 -spec init_ets() -> ok.
 init_ets() ->
     ?ETS = ets:new(?ETS, [public, named_table, set]),
+    ?UNFUNDED_ETS = ets:new(?UNFUNDED_ETS, [public, named_table, set]),
+    ok.
+
+-spec add_unfunded(OrgID :: binary()) -> boolean().
+add_unfunded(OrgID) ->
+    Update = ets:update_counter(?UNFUNDED_ETS, OrgID, {2, 1}, {default, 0}),
+    Max = router_utils:get_env_int(max_unfunded_remove_retry, 15),
+    ShouldRemove = Update < Max,
+    ShouldRemove.
+
+-spec remove_unfunded(OrgID :: binary()) -> ok.
+remove_unfunded(OrgID) ->
+    true = ets:delete(?UNFUNDED_ETS, OrgID),
+    ok.
+
+-spec list_unfunded() -> [binary()].
+list_unfunded() ->
+    [OrgID || {OrgID, _} <- ets:tab2list(?UNFUNDED_ETS)].
+
+-spec reset_unfunded_from_api() -> ok.
+reset_unfunded_from_api() ->
+    true = ets:delete_all_objects(?UNFUNDED_ETS),
+    {ok, OrgIDs} = router_console_api:get_unfunded_org_ids(),
+    true = ets:insert(?UNFUNDED_ETS, [{ID, 0} || ID <- OrgIDs]),
     ok.
 
 -spec refill(OrgID :: binary(), Nonce :: non_neg_integer(), Balance :: non_neg_integer()) -> ok.
 refill(OrgID, Nonce, Balance) ->
+    ok = ?MODULE:remove_unfunded(OrgID),
     case lookup(OrgID) of
         {error, not_found} ->
             lager:info("refilling ~p with ~p @ epoch ~p", [OrgID, Balance, Nonce]),
@@ -290,15 +321,19 @@ insert(OrgID, Balance, Nonce) ->
 %% ------------------------------------------------------------------
 -ifdef(TEST).
 
+delete_ets() ->
+    ets:delete(?ETS),
+    ets:delete(?UNFUNDED_ETS).
+
 refill_test() ->
-    _ = ets:new(?ETS, [public, named_table, set]),
+    ok = init_ets(),
     OrgID = <<"ORG_ID">>,
     Nonce = 1,
     Balance = 100,
     ?assertEqual({error, not_found}, lookup(OrgID)),
     ?assertEqual(ok, refill(OrgID, Nonce, Balance)),
     ?assertEqual({ok, Balance, Nonce}, lookup(OrgID)),
-    ets:delete(?ETS),
+    delete_ets(),
     ok.
 
 setup_meck() ->
@@ -320,7 +355,7 @@ teardown_meck() ->
     ok.
 
 has_enough_dc_test() ->
-    _ = ets:new(?ETS, [public, named_table, set]),
+    ok = init_ets(),
     ok = setup_meck(),
 
     OrgID = <<"ORG_ID">>,
@@ -333,11 +368,11 @@ has_enough_dc_test() ->
     ?assertEqual(ok, refill(OrgID, Nonce, Balance)),
     ?assertEqual({ok, OrgID, 0, 1}, has_enough_dc(OrgID, PayloadSize)),
 
-    ets:delete(?ETS),
+    delete_ets(),
     ok = teardown_meck().
 
 charge_test() ->
-    _ = ets:new(?ETS, [public, named_table, set]),
+    ok = init_ets(),
     ok = setup_meck(),
 
     OrgID = <<"ORG_ID">>,
@@ -351,11 +386,11 @@ charge_test() ->
     ?assertEqual({ok, 0, 1}, charge(OrgID, PayloadSize)),
     ?assertEqual({error, {not_enough_dc, 0, Balance}}, charge(OrgID, PayloadSize)),
 
-    ets:delete(?ETS),
+    delete_ets(),
     ok = teardown_meck().
 
 current_balance_test() ->
-    _ = ets:new(?ETS, [public, named_table, set]),
+    ok = init_ets(),
     meck:new(router_console_api, [passthrough]),
     meck:expect(router_console_api, get_org, fun(_OrgID) -> {error, 0} end),
     OrgID = <<"ORG_ID">>,
@@ -366,22 +401,22 @@ current_balance_test() ->
     ?assertEqual({100, 1}, current_balance(OrgID)),
     ?assert(meck:validate(router_console_api)),
     meck:unload(router_console_api),
-    ets:delete(?ETS),
+    delete_ets(),
     ok.
 
 lookup_balance_less_than_test() ->
-    _ = ets:new(?ETS, [public, named_table, set]),
+    ok = init_ets(),
     OrgID = <<"ORG_ID">>,
     Nonce = 1,
     Balance = 100,
     ?assertEqual(ok, refill(OrgID, Nonce, Balance)),
     ?assertEqual(ok, refill(<<"ANOTHER_ORG">>, 1, 999)),
     ?assertEqual([{OrgID, {Balance, Nonce}}], lookup_balance_less_than(500)),
-    ets:delete(?ETS),
+    delete_ets(),
     ok.
 
 lookup_nonce_test() ->
-    _ = ets:new(?ETS, [public, named_table, set]),
+    ok = init_ets(),
 
     ?assertEqual(ok, refill(<<"ONE">>, 1, 1)),
     ?assertEqual(ok, refill(<<"TWO">>, 2, 2)),
@@ -390,11 +425,11 @@ lookup_nonce_test() ->
     ?assertEqual([{<<"ONE">>, {1, 1}}, {<<"THREE">>, {2, 1}}], lookup_nonce(1)),
     ?assertEqual([{<<"TWO">>, {2, 2}}], lookup_nonce(2)),
 
-    ets:delete(?ETS),
+    delete_ets(),
     ok.
 
 lookup_balance_test() ->
-    _ = ets:new(?ETS, [public, named_table, set]),
+    ok = init_ets(),
 
     ?assertEqual(ok, refill(<<"ONE">>, 1, 1)),
     ?assertEqual(ok, refill(<<"TWO">>, 2, 2)),
@@ -403,7 +438,7 @@ lookup_balance_test() ->
     ?assertEqual([{<<"ONE">>, {1, 1}}], lookup_balance(1)),
     ?assertEqual([{<<"TWO">>, {2, 2}}, {<<"THREE">>, {2, 1}}], lookup_balance(2)),
 
-    ets:delete(?ETS),
+    delete_ets(),
     ok.
 
 -endif.