anchor.diff:  init(_Args) ->
       timer:sleep(timer:seconds(2)),% todo wait for couchdb
       process_flag(trap_exit, true),
  -    LastSeq = get_last_seq(),
  -    start_changes_stream(LastSeq),
  -    {ok, #{provider_callbacks => #{}, last_seq => LastSeq}}.
  +
  +    CacheMaxSize = cache_size_limit(),
  +    ?info("Cache max size ~p", [CacheMaxSize]),
  +
  +    CurrentSeq = last_seq(),
  +    ?info("Starting changes stream using changes from ~p", [CurrentSeq]),
  +    start_changes_stream(CurrentSeq),
  +
  +    {ok, #{
  +        provider_callbacks => #{},
  +        cache => gb_trees:empty(),
  +        user_to_provider => #{}
  +    }}.
  +
   
   handle(healthcheck) ->
  -    ok;
  +    case couchdb_datastore_driver:db_run(couchbeam_changes, follow_once, [], 30) of
  +        {ok, _, _} -> ok;
  +        _ -> {error, couchbeam_not_reachible}
  +    end;
  +
   handle({provider_subscribe, ProviderID, Callback, From}) ->
       ?info("Request ~p", [{provider_subscribe, ProviderID, Callback, From}]),
       add_subscription(ProviderID, Callback),
       fetch_history(ProviderID, Callback, From);
  +
   handle(_Request) ->
       ?log_bad_request(_Request).
   
  
positive.code: 
  -spec common_change_callback(Seq :: non_neg_integer(), datastore:document() | stream_ended, model_behaviour:model_type() | undefined) -> ok.
  common_change_callback(_Seq, stream_ended, _Type) ->
      Seq = worker_host:state_get(?MODULE, last_seq),
      ?error("Changes stream broken - restarting at ~p", [Seq]),
      start_changes_stream(Seq),
      ok;
  common_change_callback(Seq, Doc, Type) ->
      cleanup_expired(),
      Callbacks = worker_host:state_get(?MODULE, provider_callbacks),
      update_last_common_seq(Seq),
      handle_change(Seq, Doc, Type, Callbacks).
  
  update_last_common_seq(Seq) ->
      worker_host:state_update(?MODULE, last_seq, fun(Val) ->
          case Val < Seq of
              true -> Seq;
              false -> Val
          end
      end).
  
  handle_change(Seq, Doc, Type, Callbacks) ->
      NoOp = fun(_Seq, _Doc, _Type) -> ok end,
      Providers = get_providers(Seq, Doc, Type),
      ?info("Dispatching {seq=~p, doc=~p, type=~p} to ~p having ~p", [Seq, Doc, Type, Providers, Callbacks]),
      lists:foreach(fun(Provider) ->
          {Callback, _} = maps:get(Provider, Callbacks, NoOp),
          spawn(fun() -> Callback(Seq, Doc, Type) end)
      end, Providers).
  
  cleanup_expired() ->
      worker_host:state_update(?MODULE, provider_callbacks, fun(Callbacks) ->
          Now = now_seconds(),
          maps:filter(fun(_ID, {_Callback, ExpiresAt}) ->
  
negative.code:     end),
      ok.
  
  cache_size_limit() ->
      application:get_env(?APP_Name, subscription_cache_size, 100).
  
  now_seconds() ->
      erlang:system_time(seconds).
  
  