anchor.diff:  %%% Internal functions
   %%%===================================================================
   
  +update_cache(Seq, Doc, Type) ->
  +    worker_host:state_update(?MODULE, cache, fun(Cache) ->
  +        UpdatedCache = gb_trees:insert(Seq, {Doc, Type}, Cache),
  +        Size = gb_trees:size(Cache),
  +        Limit = cache_size_limit(),
  +        case Size > Limit of
  +            true ->
  +                {_, _, FinalCache} = gb_trees:take_smallest(UpdatedCache),
  +                FinalCache;
  +            false ->
  +                ?warning("Cache not saturated - size ~p of ~p", [Size, Limit]),
  +                UpdatedCache
  +        end
  +    end).
  +
   add_subscription(ID, Callback) ->
       TTL = application:get_env(?APP_Name, subscription_ttl_seconds, 120),
       ExpiresAt = now_seconds() + TTL,
  
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: init(_Args) ->
      timer:sleep(timer:seconds(2)),% todo wait for couchdb
      process_flag(trap_exit, true),
  
      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) ->
      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).
  
  