anchor.diff:          Result -> Result
       end.
   
  -%% @doc Verifies that all checksums in the given sandbox are marked as uploaded, and if so,
  -%% deletes the sandbox from the database.
  -%%
  -%% This most closely corresponds to the "delete_X" functions for other Chef objects.
  --spec commit_sandbox(#context{}, #chef_sandbox{}) -> {ok, 1 | 2} |
  -                                                     not_found |
  -                                                     {error, {need_upload, [binary()]}} |
  -                                                     {error, term()}.
  -commit_sandbox(#context{}=Ctx, #chef_sandbox{}=Sandbox) ->
  -    case non_uploaded_checksums(Ctx, Sandbox) of
  -        [] ->
  -            %% Good to go!
  -            remove_sandbox(Ctx, Sandbox);
  -        Checksums when is_list(Checksums) ->
  -            %% Oops, you've still got some work to do
  -            {error, {need_upload, Checksums}};
  -        {error, Reason} ->
  -            {error, Reason}
  -    end.
  -
  -%% @doc Deletes a sandbox from the database.
  -%%
  -%% This is not an exported function, because it is a considered a "private" step in other
  -%% processes.  Currently, this is only used for committing a sandbox, but in the future, it
  -%% will be used for purging stale sandboxes as well.
  --spec remove_sandbox(#context{}, #chef_sandbox{}) -> {ok, 1 | 2} |
  -                                                     not_found |
  -                                                     {error, any()}.
  -remove_sandbox(#context{}=Ctx, #chef_sandbox{}=Sandbox) ->
  -    delete_object(Ctx, delete_sandbox, Sandbox).
  -
  -%% @doc Returns a list of all the checksums in the given sandbox that have not yet been
  -%% marked as uploaded.
  --spec non_uploaded_checksums(#context{}, #chef_sandbox{}) -> [binary()] | {error, term()}.
  -non_uploaded_checksums(#context{reqid=ReqId}, #chef_sandbox{id=SandboxId,
  -                                                            org_id=OrgId}) ->
  -    ?SH_TIME(ReqId, chef_sql, non_uploaded_checksums, (SandboxId, OrgId)).
  -
  -%% @doc Given an OrgId and a list of checksums, marks them all as having been uploaded to the system
  --spec mark_checksums_as_uploaded(#context{}, binary(), [binary()]) -> ok | {error, term()}.
  -mark_checksums_as_uploaded(#context{reqid=ReqId}, OrgId, Checksums) ->
  -    ?SH_TIME(ReqId, chef_sql, mark_checksums_as_uploaded, (OrgId, Checksums)).
  -
  --spec node_record_to_authz_id(any(), any()) -> id().
  -node_record_to_authz_id(_Context, NodeRecord) ->
  -    NodeRecord#chef_node.authz_id.
  -
  --spec is_user_in_org(#context{}, binary(), binary()) -> boolean() | {error, _}.
  -is_user_in_org(#context{reqid = ReqId, otto_connection = S}=Ctx, User, OrgName) ->
  -    case fetch(#chef_user{username = User}, Ctx) of
  -        #chef_user{id = UserId} ->
  -            ?SH_TIME(ReqId, chef_otto, is_user_in_org, (S, UserId, OrgName));
  -        not_found ->
  -            false;
  -        {error, Why} ->
  -            {error, Why}
  -    end.
  -
  -connect() ->
  -    chef_otto:connect().
  -
  --spec bulk_get(#context{}, binary(), chef_type(), [binary()]) ->
  -                      [binary()|ej:json_object()] | {error, _}.
  -%% @doc Return a list of JSON/gzip'd JSON as binary corresponding to the specified list of
  -%% IDs.
  -bulk_get(#context{reqid = ReqId}, _OrgName, node, Ids) ->
  -    bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_nodes, (Ids)));
  -bulk_get(#context{reqid = ReqId}, _OrgName, role, Ids) ->
  -    bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_roles, (Ids)));
  -bulk_get(#context{reqid = ReqId}, _OrgName, data_bag_item, Ids) ->
  -    bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_data_bag_items, (Ids)));
  -bulk_get(#context{reqid = ReqId}, _OrgName, environment, Ids) ->
  -    bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_environments, (Ids)));
  -bulk_get(#context{reqid = ReqId}, OrgName, client, Ids) ->
  -    ClientRecords = bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_clients, (Ids))),
  -    [chef_client:assemble_client_ejson(C, OrgName) || #chef_client{}=C <- ClientRecords];
  -
  -% TODO: Can this come out now?
  -bulk_get(Ctx, OrgName, Type, Ids) ->
  -    bulk_get_couchdb(Ctx, OrgName, Type, Ids).
  -
  -bulk_get_result({ok, not_found}) ->
  -    [];
  -bulk_get_result({ok, L}) when is_list(L) ->
  -    L;
  -bulk_get_result({error, _Why}=Error) ->
  -    Error.
  -
  -bulk_get_couchdb(#context{reqid = ReqId, otto_connection = S}=Ctx, OrgName, _Type, Ids) ->
  -    %% presently searching for non-nodes goes to couchdb
  -    case fetch_org_id(Ctx, OrgName) of
  -        not_found ->
  -            {error, {not_found, org}};
  -        OrgId ->
  -            DbName = chef_otto:dbname(OrgId),
  -            ?SH_TIME(ReqId, chef_otto, bulk_get, (S, DbName, Ids))
  -    end.
  -
  --spec data_bag_exists(#context{}, binary(), binary()) -> boolean().
  -%% @doc Return true if data bag `DataBag' exists in org `OrgName' and false otherwise.
  -data_bag_exists(#context{}=Ctx, OrgId, DataBag) ->
  -    case fetch(#chef_data_bag{org_id = OrgId, name = DataBag}, Ctx) of
  -        #chef_data_bag{} -> true;
  -        not_found -> false
  -    end.
  -
  --spec environment_exists(#context{}, <<_:256>>, binary()) -> boolean().
  -%% @doc Return true if environment `EnvName' exists in org `OrgId' and false otherwise.
  -environment_exists(#context{}=Ctx, OrgId, EnvName) ->
  -    %% FIXME: we should implement a specialized environment exists function
  -    case fetch(#chef_environment{org_id = OrgId, name = EnvName}, Ctx) of
  -        #chef_environment{} -> true;
  -        _ -> false
  -    end.
  -
  --spec create(object_rec(), #context{}, object_id()) -> ok | {conflict, term()} | {error, term()}.
  -create(#chef_cookbook_version{} = Record, DbContext, ActorId) ->
  -    create_object(DbContext, create_cookbook_version, Record, ActorId);
  -create(ObjectRec0, #context{reqid = ReqId}, ActorId) ->
  -    ObjectRec = chef_object:set_created(ObjectRec0, ActorId),
  -    QueryName = chef_object:create_query(ObjectRec),
  -    case stats_hero:ctime(ReqId, {chef_sql, create_object},
  -                          fun() -> chef_sql:create_object(QueryName, ObjectRec) end) of
  -        {ok, 1} -> ok;
  -        {conflict, Msg}-> {conflict, Msg};
  -        {error, Why} -> {error, Why}
  +-spec fetch(object_rec(),
  +            DbContext :: #context{}) ->
  +                   object_rec() |
  +                   not_found |
  +                   {error, term()}.
  +fetch(ObjectRec, #context{reqid = ReqId}) ->
  +    RecordType = element(1, ObjectRec), %% record type is not the same as type name :(
  +    QueryName = chef_object:find_query(ObjectRec),
  +    Keys = chef_object:fields_for_fetch(ObjectRec),
  +    RecordFields = chef_object:record_fields(ObjectRec),
  +    case stats_hero:ctime(ReqId, {chef_sql, fetch_object},
  +                          fun() -> chef_sql:fetch_object(Keys, RecordType, QueryName, RecordFields) end) of
  +        {ok, not_found} -> not_found;
  +        {ok, Object} -> Object;
  +        {error, _Why} = Error -> Error
       end.
   
   -spec list(#context{}, object_rec()) -> {ok, [binary()]} | {error, _}.
  
positive.code: make_context(ReqId, Darklaunch, OttoServer) ->
      #context{reqid = ReqId, darklaunch = Darklaunch, otto_connection = OttoServer}.
  
  darklaunch_from_context(#context{darklaunch = Darklaunch}) ->
      Darklaunch.
  
  -spec count_user_admins(#context{}) -> integer() | {error, term()}.
  count_user_admins(#context{reqid = ReqId}) ->
    case stats_hero:ctime(ReqId, {chef_sql, count_user_admins},
                          fun() -> chef_sql:count_user_admins() end) of
         {ok, Count} -> Count;
         Other -> Other
    end.
  
  -spec user_record_to_authz_id(#context{}, #chef_user{} | not_found) -> id().
  user_record_to_authz_id(#context{}, #chef_user{} = UserRecord) ->
      UserRecord#chef_user.authz_id;
  user_record_to_authz_id(#context{}, not_found) ->
      %% FIXME: is this what we want here?
      erlang:error({error, not_found}).
  
  -spec fetch_org_id(#context{}, binary() | ?OSC_ORG_NAME) -> not_found | binary().
  fetch_org_id(_, ?OSC_ORG_NAME) ->
      ?OSC_ORG_ID;
  fetch_org_id(#context{reqid = ReqId,
                        otto_connection = Server}, OrgName) when is_binary(OrgName) ->
      case chef_cache:get(org_guid, OrgName) of
          Error when Error =:= not_found orelse Error =:= no_cache ->
              case ?SH_TIME(ReqId, chef_otto, fetch_org_id, (Server, OrgName)) of
                  not_found ->
                      not_found;
                  Guid ->
                      chef_cache:put(org_guid, OrgName, Guid),
                      Guid
              end;
          {ok, Guid} ->
              Guid
      end.
  
  
  client_record_to_authz_id(_Context, ClientRecord) ->
      ClientRecord#chef_client.authz_id.
  
  %% @doc Given a name and an org, find either a user or a client and return a
  %% #chef_user{} or #chef_client{} record.
  %%
  %% Looks for a client first, then a user. We search for the client first for two reasons: 1)
  %% more API requests come from clients; 2) prevents a DoS attack cleverly chosen user names
  %% in OHC breaking clients with matching names.
  %%
  %% We examine the user/client record and return the certificate containing the actor's
  %% public key.  Some legacy records are present in the database that do not have
  %% certificates and have only a public key.  For these cases, we return the key tagged with
  %% 'key' instead of 'cert'.
  -spec fetch_requestor(#context{},
                        binary(),
                        binary()) -> #chef_client{} | #chef_user{} |
                                     %% TODO: fix chef_wm so we can just return 'not_found'
                                     {'not_found', 'client'}.
  fetch_requestor(Context, OrgId, ClientName) ->
      case fetch(#chef_client{org_id = OrgId, name = ClientName}, Context) of
          not_found ->
              case fetch(#chef_user{username = ClientName}, Context) of
                  not_found ->
                      %% back compat for now until we update chef_wm
                      {not_found, client};
                  #chef_user{} = User ->
                      User
              end;
          #chef_client{} = Client ->
              Client
      end.
  
  %% Currently, this is only used internally, but will be useful for future enhancements to
  %% the sandboxes protocol that allow for partial progress monitoring (e.g., upload a few
  %% files, retrieve the current state of the sandbox, upload a few more, etc.)
  -spec fetch_sandbox(DbContext::#context{},
                      OrgName :: binary() | {id, binary()},
                      SandboxId::object_id()) -> #chef_sandbox{} |
                                                 not_found |
                                                 {error, any()}.
  fetch_sandbox(#context{reqid = ReqId} = Ctx, {id, OrgId}, SandboxId) ->
      case stats_hero:ctime(ReqId, {chef_sql, fetch_sandbox},
                            fun() -> chef_sql:fetch_sandbox(OrgId, SandboxId) end) of
          {ok, not_found} -> not_found;
          {ok, Object} -> Object;
          {error, _Why} = Error -> Error
      end;
  fetch_sandbox(#context{}=Ctx, OrgName, SandboxId) ->
      case fetch_org_id(Ctx, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              fetch_sandbox(Ctx, {id, OrgId}, SandboxId)
      end.
  
  %% @doc Saves sandbox information for a new sandbox in the database, and returns a
  %% chef_sandbox record representing the new sandbox.  This is a different pattern from other
  %% Chef objects, because the creation of a record for the object requires data that the
  %% caller does not have access to (the 'upload' state of the checksums must be retrieved
  %% from the database).
  
  -spec make_sandbox(DbContext :: #context{},
                     OrgName :: binary(),
                     ActorId :: object_id(),
                     Checksums :: [binary()]) -> #chef_sandbox{} |
                                                 not_found |
                                                 {conflict, term()} |
                                                 {error, term()}.
  make_sandbox(#context{}=Ctx, OrgName, ActorId, Checksums) ->
      case fetch_org_id(Ctx, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              Id = chef_object_base:make_org_prefix_id(OrgId),
              %% TempSandbox doesn't know if the checksums have been uploaded yet or not
              TempSandbox = #chef_sandbox{id=Id,
                                          org_id=OrgId,
                                          checksums = [{C, false} || C <- Checksums]},
              %% TODO: ActorId isn't actually needed in the code, but it is if we want to use common code paths
              case create_object(Ctx, create_sandbox, TempSandbox, ActorId) of
                  ok ->
                      %% this sandbox will know if the checksums have been uploaded
                      fetch_sandbox(Ctx, OrgName, Id);
                  {conflict, Msg} ->
                      {conflict, Msg};
                  {error, Why} ->
                      {error, Why}
              end
      end.
  
  -spec cookbook_exists(DbContext :: #context{},
                        OrgName :: binary(),
                        CookbookName :: binary()) ->
                               boolean() | {error, term()}.
  cookbook_exists(#context{reqid=ReqId} = DbContext, OrgName, CookbookName) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              ?SH_TIME(ReqId, chef_sql, cookbook_exists, (OrgId, CookbookName))
      end.
  
  -spec fetch_cookbook_versions(#context{}, binary()) -> {not_found, org} |
                                                         [versioned_cookbook()] |
                                                         {error, any()}.
  %% @doc Return a list of all cookbook names and versions in an org
  fetch_cookbook_versions(#context{} = Ctx, OrgName) ->
      fetch_objects(Ctx, fetch_cookbook_versions, OrgName).
  
  -spec fetch_cookbook_versions(#context{}, binary(), binary()) -> {not_found, org} |
                                                                   [versioned_cookbook()] |
                                                                   {error, any()}.
  %% @doc Return a list of all cookbook names and versions in an org
  fetch_cookbook_versions(#context{} = Ctx, OrgName, CookbookName) ->
      fetch_objects(Ctx, fetch_cookbook_versions, OrgName, CookbookName).
  
  -spec fetch_cookbook_version(DbContext :: #context{},
                               OrgName :: binary(),
                               VersionedCookbook :: versioned_cookbook()) -> #chef_cookbook_version{} |
                                                                             {cookbook_exists, object_id()} |
                                                                             not_found |
                                                                             {error, term()}.
  fetch_cookbook_version(#context{reqid = ReqId} = Ctx, OrgName, VersionedCookbook) ->
      case fetch_org_id(Ctx, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              stats_hero:ctime(ReqId, {chef_sql, fetch_cookbook_version},
                               fun() -> chef_sql:fetch_cookbook_version(OrgId,
                                                                        VersionedCookbook)
                               end)
      end.
  
  -spec fetch_latest_cookbook_version(Ctx::#context{},
                                      OrgName::binary(),
                                      CookbookName::binary()) ->
                                      #chef_cookbook_version{} |
                                      not_found |
                                      {error, term()}.
  %% @doc Return the latest version of the requested cookbook
  fetch_latest_cookbook_version(#context{reqid=ReqId} = Ctx, OrgName, CookbookName) ->
      case fetch_org_id(Ctx, OrgName) of
        not_found ->
          not_found;
        OrgId ->
           stats_hero:ctime(ReqId, {chef_sql, fetch_latest_cookbook_version},
                            fun() -> chef_sql:fetch_latest_cookbook_version(OrgId, CookbookName)
                            end)
      end.
  
  %% @doc Retrieve a list of `{Name, Verison}' tuples for the latest version of each cookbook in
  %% an organization.  Version information is returned as a binary string (e.g., `<<"1.0.0">>')
  %% instead of the tuple form (e.g. `{1,0,0}') that is used elsewhere.
  -spec fetch_latest_cookbook_versions(DbContext :: #context{},
                                       OrgName :: binary()) ->
                                              [{CookbookName :: binary(),
                                                VersionString :: binary()}] |
                                              {not_found, org} |
                                              {error, Reason :: term()}.
  fetch_latest_cookbook_versions(#context{}=DbContext, OrgName) ->
      fetch_latest_cookbook_versions(DbContext, OrgName, 1).
  
  %% @doc Same as {@link fetch_latest_cookbook_versions/2}, but allows you to specify how many
  %% versions for which to fetch information.  For instance, setting `NumberOfVersions' to `3'
  %% will retrieve the `{Name, Version}' pairs for the three most recent versions of each
  %% cookbook.
  %%
  %% If fewer than `NumberOfVersions' versions exist for a given cookbook, all
  %% versions are represented in the output.
  -spec fetch_latest_cookbook_versions(DbContext :: #context{},
                                       OrgName :: binary(),
                                       NumberOfVersions :: non_neg_integer()) ->
                                              [{CookbookName :: binary(),
                                                VersionString :: binary()}] |
                                              {not_found, org} |
                                              {error, Reason :: term()}.
  fetch_latest_cookbook_versions(#context{reqid=ReqId}=DbContext, OrgName, NumberOfVersions) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              {not_found, org};
          OrgId ->
              case ?SH_TIME(ReqId, chef_sql, fetch_latest_cookbook_versions, (OrgId, all, NumberOfVersions)) of
                  {ok, VersionInfo} ->
                      VersionInfo;
                  {error, Error} ->
                      {error, Error}
              end
      end.
  
  %% @doc Retrieves the list of all recipes from the latest version of all an organization's
  %% cookbooks and returns their cookbook-qualified names.
  -spec fetch_latest_cookbook_recipes(DbContext :: #context{},
                                      OrgName :: binary()) ->
                                             [CookbookQualifiedRecipeName :: binary()] |
                                             {not_found, org} |
                                             {error, Reason :: term()}.
  fetch_latest_cookbook_recipes(#context{reqid=ReqId}=DbContext, OrgName) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              {not_found, org};
          OrgId ->
              case ?SH_TIME(ReqId, chef_sql, fetch_latest_cookbook_recipes, (OrgId)) of
                  {ok, Results} ->
                      Results;
                  {error, Error} ->
                      {error, Error}
              end
      end.
  
  %% @doc Retrieve all cookbook dependency information for an organization as a list of
  %% Depsolver dependency sets.
  %%
  %% See the corresponding function in the chef_sql module for more information.
  -spec fetch_all_cookbook_version_dependencies(DbContext :: #context{},
                                                OrgName :: binary()) -> [depsolver:dependency_set()] |
                                                                        {error, term()}.
  fetch_all_cookbook_version_dependencies(#context{reqid=ReqId}=DbContext, OrgName) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              case ?SH_TIME(ReqId, chef_sql, fetch_all_cookbook_version_dependencies, (OrgId)) of
                  {ok, Results} ->
                      Results;
                  {error, Error} ->
                      {error, Error}
              end
      end.
  
  %% @doc Retrieve cookbook versions, subject to any constraints imposed by the specified
  %% environment.  `NumVersions' is either an integer or the atom `all', and indicates the
  %% maximum number of versions that should be returned that satisfy the constraints for each
  %% cookbook
  %%
  %% If the given CookbookName is the atom 'all', information for all cookbooks is retrieved;
  %% if it is a cookbook name (i.e., a binary), then information on just that cookbook will be
  %% returned.
  -spec fetch_environment_filtered_cookbook_versions(DbContext :: #context{},
                                                     OrgName :: binary(),
                                                     EnvName :: binary(),
                                                     CookbookName :: binary() | all,
                                                     NumVersions :: all | non_neg_integer()) ->
                                                            [{CookbookName :: binary(), [Version :: binary()]}] |
                                                            {error, term()}.
  fetch_environment_filtered_cookbook_versions(#context{reqid=ReqId}=DbContext, OrgName, EnvName, CookbookName, NumVersions) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              case fetch(#chef_environment{org_id = OrgId, name = EnvName}, DbContext) of
                  #chef_environment{} = Environment ->
                      case ?SH_TIME(ReqId, chef_sql, fetch_environment_filtered_cookbook_versions, (OrgId, Environment, CookbookName, NumVersions)) of
                          {ok, Results} ->
                              Results;
                          {error, Error} ->
                              {error, Error}
                      end;
                  {error, Error} -> {error, Error}
              end
      end.
  
  fetch_environment_filtered_recipes(#context{reqid=ReqId}=DbContext, OrgName, EnvName) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              case fetch(#chef_environment{org_id = OrgId, name = EnvName}, DbContext) of
                  #chef_environment{} = Environment ->
                      case ?SH_TIME(ReqId, chef_sql, fetch_environment_filtered_recipes, (OrgId, Environment)) of
                          {ok, Results} ->
                              Results;
                          {error, Error} ->
                              {error, Error}
                      end;
                  {error, Error} -> {error, Error}
              end
      end.
  
  
  -spec fetch_roles(#context{}, binary()) -> {not_found, org} |
                                             [binary()] |
                                             {error, any()}.
  %% @doc Return a list of all role names in an org
  fetch_roles(#context{} = Ctx, OrgName) ->
      fetch_objects(Ctx, fetch_roles, OrgName).
  
  
  -spec fetch_data_bag_item_ids(#context{}, binary() | {id, object_id()},
                                binary()) -> [binary()] |
                                             {not_found, org} |
                                             {error, _}.
  %% @doc Returns list of data_bag_item names in `DataBagName' for `OrgName'.
  fetch_data_bag_item_ids(#context{} = Ctx, OrgName, DataBagName) ->
      fetch_objects(Ctx, fetch_data_bag_item_ids, OrgName, DataBagName).
  
  -spec delete(#context{}, object_rec()) -> {ok, 1 | 2} | not_found | {error, _}.
  delete(#context{reqid = ReqId} = Ctx,
  
negative.code:         {error, Error} -> {error, Error}
      end.
  
  darklaunch_from_context(#context{darklaunch = Darklaunch}) ->
      Darklaunch.
  
  -spec count_user_admins(#context{}) -> integer() | {error, term()}.
  count_user_admins(#context{reqid = ReqId}) ->
    case stats_hero:ctime(ReqId, {chef_sql, count_user_admins},
                          fun() -> chef_sql:count_user_admins() end) of
         {ok, Count} -> Count;
         Other -> Other
    end.
  
  -spec user_record_to_authz_id(#context{}, #chef_user{} | not_found) -> id().
  user_record_to_authz_id(#context{}, #chef_user{} = UserRecord) ->
      UserRecord#chef_user.authz_id;
  user_record_to_authz_id(#context{}, not_found) ->
      %% FIXME: is this what we want here?
      erlang:error({error, not_found}).
  
  -spec fetch_org_id(#context{}, binary() | ?OSC_ORG_NAME) -> not_found | binary().
  fetch_org_id(_, ?OSC_ORG_NAME) ->
      ?OSC_ORG_ID;
  fetch_org_id(#context{reqid = ReqId,
                        otto_connection = Server}, OrgName) when is_binary(OrgName) ->
      case chef_cache:get(org_guid, OrgName) of
          Error when Error =:= not_found orelse Error =:= no_cache ->
              case ?SH_TIME(ReqId, chef_otto, fetch_org_id, (Server, OrgName)) of
                  not_found ->
                      not_found;
                  Guid ->
                      chef_cache:put(org_guid, OrgName, Guid),
                      Guid
              end;
          {ok, Guid} ->
              Guid
      end.
  
  client_record_to_authz_id(_Context, ClientRecord) ->
      ClientRecord#chef_client.authz_id.
  
  %% @doc Given a name and an org, find either a user or a client and return a
  %% #chef_user{} or #chef_client{} record.
  %%
  %% Looks for a client first, then a user. We search for the client first for two reasons: 1)
  %% more API requests come from clients; 2) prevents a DoS attack cleverly chosen user names
  %% in OHC breaking clients with matching names.
  %%
  %% We examine the user/client record and return the certificate containing the actor's
  %% public key.  Some legacy records are present in the database that do not have
  %% certificates and have only a public key.  For these cases, we return the key tagged with
  %% 'key' instead of 'cert'.
  -spec fetch_requestor(#context{},
                        binary(),
                        binary()) -> #chef_client{} | #chef_user{} |
                                     %% TODO: fix chef_wm so we can just return 'not_found'
                                     {'not_found', 'client'}.
  fetch_requestor(Context, OrgId, ClientName) ->
      case fetch(#chef_client{org_id = OrgId, name = ClientName}, Context) of
          not_found ->
              case fetch(#chef_user{username = ClientName}, Context) of
                  not_found ->
                      %% back compat for now until we update chef_wm
                      {not_found, client};
                  #chef_user{} = User ->
                      User
              end;
          #chef_client{} = Client ->
              Client
      end.
  
  %% Currently, this is only used internally, but will be useful for future enhancements to
  %% the sandboxes protocol that allow for partial progress monitoring (e.g., upload a few
  %% files, retrieve the current state of the sandbox, upload a few more, etc.)
  -spec fetch_sandbox(DbContext::#context{},
                      OrgName :: binary() | {id, binary()},
                      SandboxId::object_id()) -> #chef_sandbox{} |
                                                 not_found |
                                                 {error, any()}.
  fetch_sandbox(#context{reqid = ReqId} = Ctx, {id, OrgId}, SandboxId) ->
      case stats_hero:ctime(ReqId, {chef_sql, fetch_sandbox},
                            fun() -> chef_sql:fetch_sandbox(OrgId, SandboxId) end) of
          {ok, not_found} -> not_found;
          {ok, Object} -> Object;
          {error, _Why} = Error -> Error
      end;
  fetch_sandbox(#context{}=Ctx, OrgName, SandboxId) ->
      case fetch_org_id(Ctx, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              fetch_sandbox(Ctx, {id, OrgId}, SandboxId)
      end.
  
  %% @doc Saves sandbox information for a new sandbox in the database, and returns a
  %% chef_sandbox record representing the new sandbox.  This is a different pattern from other
  %% Chef objects, because the creation of a record for the object requires data that the
  %% caller does not have access to (the 'upload' state of the checksums must be retrieved
  %% from the database).
  
  -spec make_sandbox(DbContext :: #context{},
                     OrgName :: binary(),
                     ActorId :: object_id(),
                     Checksums :: [binary()]) -> #chef_sandbox{} |
                                                 not_found |
                                                 {conflict, term()} |
                                                 {error, term()}.
  make_sandbox(#context{}=Ctx, OrgName, ActorId, Checksums) ->
      case fetch_org_id(Ctx, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              Id = chef_object_base:make_org_prefix_id(OrgId),
              %% TempSandbox doesn't know if the checksums have been uploaded yet or not
              TempSandbox = #chef_sandbox{id=Id,
                                          org_id=OrgId,
                                          checksums = [{C, false} || C <- Checksums]},
              %% TODO: ActorId isn't actually needed in the code, but it is if we want to use common code paths
              case create_object(Ctx, create_sandbox, TempSandbox, ActorId) of
                  ok ->
                      %% this sandbox will know if the checksums have been uploaded
                      fetch_sandbox(Ctx, OrgName, Id);
                  {conflict, Msg} ->
                      {conflict, Msg};
                  {error, Why} ->
                      {error, Why}
              end
      end.
  
  -spec cookbook_exists(DbContext :: #context{},
                        OrgName :: binary(),
                        CookbookName :: binary()) ->
                               boolean() | {error, term()}.
  cookbook_exists(#context{reqid=ReqId} = DbContext, OrgName, CookbookName) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              ?SH_TIME(ReqId, chef_sql, cookbook_exists, (OrgId, CookbookName))
      end.
  
  -spec fetch_cookbook_versions(#context{}, binary()) -> {not_found, org} |
                                                         [versioned_cookbook()] |
                                                         {error, any()}.
  %% @doc Return a list of all cookbook names and versions in an org
  fetch_cookbook_versions(#context{} = Ctx, OrgName) ->
      fetch_objects(Ctx, fetch_cookbook_versions, OrgName).
  
  -spec fetch_cookbook_versions(#context{}, binary(), binary()) -> {not_found, org} |
                                                                   [versioned_cookbook()] |
                                                                   {error, any()}.
  %% @doc Return a list of all cookbook names and versions in an org
  fetch_cookbook_versions(#context{} = Ctx, OrgName, CookbookName) ->
      fetch_objects(Ctx, fetch_cookbook_versions, OrgName, CookbookName).
  
  -spec fetch_cookbook_version(DbContext :: #context{},
                               OrgName :: binary(),
                               VersionedCookbook :: versioned_cookbook()) -> #chef_cookbook_version{} |
                                                                             {cookbook_exists, object_id()} |
                                                                             not_found |
                                                                             {error, term()}.
  fetch_cookbook_version(#context{reqid = ReqId} = Ctx, OrgName, VersionedCookbook) ->
      case fetch_org_id(Ctx, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              stats_hero:ctime(ReqId, {chef_sql, fetch_cookbook_version},
                               fun() -> chef_sql:fetch_cookbook_version(OrgId,
                                                                        VersionedCookbook)
                               end)
      end.
  
  -spec fetch_latest_cookbook_version(Ctx::#context{},
                                      OrgName::binary(),
                                      CookbookName::binary()) ->
                                      #chef_cookbook_version{} |
                                      not_found |
                                      {error, term()}.
  %% @doc Return the latest version of the requested cookbook
  fetch_latest_cookbook_version(#context{reqid=ReqId} = Ctx, OrgName, CookbookName) ->
      case fetch_org_id(Ctx, OrgName) of
        not_found ->
          not_found;
        OrgId ->
           stats_hero:ctime(ReqId, {chef_sql, fetch_latest_cookbook_version},
                            fun() -> chef_sql:fetch_latest_cookbook_version(OrgId, CookbookName)
                            end)
      end.
  
  %% @doc Retrieve a list of `{Name, Verison}' tuples for the latest version of each cookbook in
  %% an organization.  Version information is returned as a binary string (e.g., `<<"1.0.0">>')
  %% instead of the tuple form (e.g. `{1,0,0}') that is used elsewhere.
  -spec fetch_latest_cookbook_versions(DbContext :: #context{},
                                       OrgName :: binary()) ->
                                              [{CookbookName :: binary(),
                                                VersionString :: binary()}] |
                                              {not_found, org} |
                                              {error, Reason :: term()}.
  fetch_latest_cookbook_versions(#context{}=DbContext, OrgName) ->
      fetch_latest_cookbook_versions(DbContext, OrgName, 1).
  
  %% @doc Same as {@link fetch_latest_cookbook_versions/2}, but allows you to specify how many
  %% versions for which to fetch information.  For instance, setting `NumberOfVersions' to `3'
  %% will retrieve the `{Name, Version}' pairs for the three most recent versions of each
  %% cookbook.
  %%
  %% If fewer than `NumberOfVersions' versions exist for a given cookbook, all
  %% versions are represented in the output.
  -spec fetch_latest_cookbook_versions(DbContext :: #context{},
                                       OrgName :: binary(),
                                       NumberOfVersions :: non_neg_integer()) ->
                                              [{CookbookName :: binary(),
                                                VersionString :: binary()}] |
                                              {not_found, org} |
                                              {error, Reason :: term()}.
  fetch_latest_cookbook_versions(#context{reqid=ReqId}=DbContext, OrgName, NumberOfVersions) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              {not_found, org};
          OrgId ->
              case ?SH_TIME(ReqId, chef_sql, fetch_latest_cookbook_versions, (OrgId, all, NumberOfVersions)) of
                  {ok, VersionInfo} ->
                      VersionInfo;
                  {error, Error} ->
                      {error, Error}
              end
      end.
  
  %% @doc Retrieves the list of all recipes from the latest version of all an organization's
  %% cookbooks and returns their cookbook-qualified names.
  -spec fetch_latest_cookbook_recipes(DbContext :: #context{},
                                      OrgName :: binary()) ->
                                             [CookbookQualifiedRecipeName :: binary()] |
                                             {not_found, org} |
                                             {error, Reason :: term()}.
  fetch_latest_cookbook_recipes(#context{reqid=ReqId}=DbContext, OrgName) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              {not_found, org};
          OrgId ->
              case ?SH_TIME(ReqId, chef_sql, fetch_latest_cookbook_recipes, (OrgId)) of
                  {ok, Results} ->
                      Results;
                  {error, Error} ->
                      {error, Error}
              end
      end.
  
  %% @doc Retrieve all cookbook dependency information for an organization as a list of
  %% Depsolver dependency sets.
  %%
  %% See the corresponding function in the chef_sql module for more information.
  -spec fetch_all_cookbook_version_dependencies(DbContext :: #context{},
                                                OrgName :: binary()) -> [depsolver:dependency_set()] |
                                                                        {error, term()}.
  fetch_all_cookbook_version_dependencies(#context{reqid=ReqId}=DbContext, OrgName) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              case ?SH_TIME(ReqId, chef_sql, fetch_all_cookbook_version_dependencies, (OrgId)) of
                  {ok, Results} ->
                      Results;
                  {error, Error} ->
                      {error, Error}
              end
      end.
  
  %% @doc Retrieve cookbook versions, subject to any constraints imposed by the specified
  %% environment.  `NumVersions' is either an integer or the atom `all', and indicates the
  %% maximum number of versions that should be returned that satisfy the constraints for each
  %% cookbook
  %%
  %% If the given CookbookName is the atom 'all', information for all cookbooks is retrieved;
  %% if it is a cookbook name (i.e., a binary), then information on just that cookbook will be
  %% returned.
  -spec fetch_environment_filtered_cookbook_versions(DbContext :: #context{},
                                                     OrgName :: binary(),
                                                     EnvName :: binary(),
                                                     CookbookName :: binary() | all,
                                                     NumVersions :: all | non_neg_integer()) ->
                                                            [{CookbookName :: binary(), [Version :: binary()]}] |
                                                            {error, term()}.
  fetch_environment_filtered_cookbook_versions(#context{reqid=ReqId}=DbContext, OrgName, EnvName, CookbookName, NumVersions) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              case fetch(#chef_environment{org_id = OrgId, name = EnvName}, DbContext) of
                  #chef_environment{} = Environment ->
                      case ?SH_TIME(ReqId, chef_sql, fetch_environment_filtered_cookbook_versions, (OrgId, Environment, CookbookName, NumVersions)) of
                          {ok, Results} ->
                              Results;
                          {error, Error} ->
                              {error, Error}
                      end;
                  {error, Error} -> {error, Error}
              end
      end.
  
  fetch_environment_filtered_recipes(#context{reqid=ReqId}=DbContext, OrgName, EnvName) ->
      case fetch_org_id(DbContext, OrgName) of
          not_found ->
              not_found;
          OrgId ->
              case fetch(#chef_environment{org_id = OrgId, name = EnvName}, DbContext) of
                  #chef_environment{} = Environment ->
                      case ?SH_TIME(ReqId, chef_sql, fetch_environment_filtered_recipes, (OrgId, Environment)) of
                          {ok, Results} ->
                              Results;
                          {error, Error} ->
                              {error, Error}
                      end;
                  {error, Error} -> {error, Error}
              end
      end.
  
  
  -spec fetch_roles(#context{}, binary()) -> {not_found, org} |
                                             [binary()] |
                                             {error, any()}.
  %% @doc Return a list of all role names in an org
  fetch_roles(#context{} = Ctx, OrgName) ->
      fetch_objects(Ctx, fetch_roles, OrgName).
  
  
  -spec fetch_data_bag_item_ids(#context{}, binary() | {id, object_id()},
                                binary()) -> [binary()] |
                                             {not_found, org} |
                                             {error, _}.
  %% @doc Returns list of data_bag_item names in `DataBagName' for `OrgName'.
  fetch_data_bag_item_ids(#context{} = Ctx, OrgName, DataBagName) ->
      fetch_objects(Ctx, fetch_data_bag_item_ids, OrgName, DataBagName).
  
  %% @doc Verifies that all checksums in the given sandbox are marked as uploaded, and if so,
  %% deletes the sandbox from the database.
  %%
  %% This most closely corresponds to the "delete_X" functions for other Chef objects.
  -spec commit_sandbox(#context{}, #chef_sandbox{}) -> {ok, 1 | 2} |
                                                       not_found |
                                                       {error, {need_upload, [binary()]}} |
                                                       {error, term()}.
  commit_sandbox(#context{}=Ctx, #chef_sandbox{}=Sandbox) ->
      case non_uploaded_checksums(Ctx, Sandbox) of
          [] ->
              %% Good to go!
              remove_sandbox(Ctx, Sandbox);
          Checksums when is_list(Checksums) ->
              %% Oops, you've still got some work to do
              {error, {need_upload, Checksums}};
          {error, Reason} ->
              {error, Reason}
      end.
  
  %% @doc Deletes a sandbox from the database.
  %%
  %% This is not an exported function, because it is a considered a "private" step in other
  %% processes.  Currently, this is only used for committing a sandbox, but in the future, it
  %% will be used for purging stale sandboxes as well.
  -spec remove_sandbox(#context{}, #chef_sandbox{}) -> {ok, 1 | 2} |
                                                       not_found |
                                                       {error, any()}.
  remove_sandbox(#context{}=Ctx, #chef_sandbox{}=Sandbox) ->
      delete_object(Ctx, delete_sandbox, Sandbox).
  
  %% @doc Returns a list of all the checksums in the given sandbox that have not yet been
  %% marked as uploaded.
  -spec non_uploaded_checksums(#context{}, #chef_sandbox{}) -> [binary()] | {error, term()}.
  non_uploaded_checksums(#context{reqid=ReqId}, #chef_sandbox{id=SandboxId,
                                                              org_id=OrgId}) ->
      ?SH_TIME(ReqId, chef_sql, non_uploaded_checksums, (SandboxId, OrgId)).
  
  %% @doc Given an OrgId and a list of checksums, marks them all as having been uploaded to the system
  -spec mark_checksums_as_uploaded(#context{}, binary(), [binary()]) -> ok | {error, term()}.
  mark_checksums_as_uploaded(#context{reqid=ReqId}, OrgId, Checksums) ->
      ?SH_TIME(ReqId, chef_sql, mark_checksums_as_uploaded, (OrgId, Checksums)).
  
  -spec node_record_to_authz_id(any(), any()) -> id().
  node_record_to_authz_id(_Context, NodeRecord) ->
      NodeRecord#chef_node.authz_id.
  
  -spec is_user_in_org(#context{}, binary(), binary()) -> boolean() | {error, _}.
  is_user_in_org(#context{reqid = ReqId, otto_connection = S}=Ctx, User, OrgName) ->
      case fetch(#chef_user{username = User}, Ctx) of
          #chef_user{id = UserId} ->
              ?SH_TIME(ReqId, chef_otto, is_user_in_org, (S, UserId, OrgName));
          not_found ->
              false;
          {error, Why} ->
              {error, Why}
      end.
  
  connect() ->
      chef_otto:connect().
  
  -spec bulk_get(#context{}, binary(), chef_type(), [binary()]) ->
                        [binary()|ej:json_object()] | {error, _}.
  %% @doc Return a list of JSON/gzip'd JSON as binary corresponding to the specified list of
  %% IDs.
  bulk_get(#context{reqid = ReqId}, _OrgName, node, Ids) ->
      bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_nodes, (Ids)));
  bulk_get(#context{reqid = ReqId}, _OrgName, role, Ids) ->
      bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_roles, (Ids)));
  bulk_get(#context{reqid = ReqId}, _OrgName, data_bag_item, Ids) ->
      bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_data_bag_items, (Ids)));
  bulk_get(#context{reqid = ReqId}, _OrgName, environment, Ids) ->
      bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_environments, (Ids)));
  bulk_get(#context{reqid = ReqId}, OrgName, client, Ids) ->
      ClientRecords = bulk_get_result(?SH_TIME(ReqId, chef_sql, bulk_get_clients, (Ids))),
      [chef_client:assemble_client_ejson(C, OrgName) || #chef_client{}=C <- ClientRecords];
  
  % TODO: Can this come out now?
  bulk_get(Ctx, OrgName, Type, Ids) ->
      bulk_get_couchdb(Ctx, OrgName, Type, Ids).
  
  bulk_get_result({ok, not_found}) ->
      [];
  bulk_get_result({ok, L}) when is_list(L) ->
      L;
  bulk_get_result({error, _Why}=Error) ->
      Error.
  
  bulk_get_couchdb(#context{reqid = ReqId, otto_connection = S}=Ctx, OrgName, _Type, Ids) ->
      %% presently searching for non-nodes goes to couchdb
      case fetch_org_id(Ctx, OrgName) of
          not_found ->
              {error, {not_found, org}};
          OrgId ->
              DbName = chef_otto:dbname(OrgId),
              ?SH_TIME(ReqId, chef_otto, bulk_get, (S, DbName, Ids))
      end.
  
  -spec data_bag_exists(#context{}, binary(), binary()) -> boolean().
  %% @doc Return true if data bag `DataBag' exists in org `OrgName' and false otherwise.
  data_bag_exists(#context{}=Ctx, OrgId, DataBag) ->
      case fetch(#chef_data_bag{org_id = OrgId, name = DataBag}, Ctx) of
          #chef_data_bag{} -> true;
          not_found -> false
      end.
  
  -spec environment_exists(#context{}, <<_:256>>, binary()) -> boolean().
  %% @doc Return true if environment `EnvName' exists in org `OrgId' and false otherwise.
  environment_exists(#context{}=Ctx, OrgId, EnvName) ->
      %% FIXME: we should implement a specialized environment exists function
      case fetch(#chef_environment{org_id = OrgId, name = EnvName}, Ctx) of
          #chef_environment{} -> true;
          _ -> false
      end.
  
  
  %% -------------------------------------
  %% private functions
  