anchor.diff:  %% -*- erlang-indent-level: 4;indent-tabs-mode: nil; fill-column: 92-*-
   %% ex: ts=4 sw=4 et
  -%% Copyright 2012 Opscode, Inc. All Rights Reserved.
  +%% Copyright 2012-2013 Opscode, Inc. All Rights Reserved.
   %%
   %% This file is provided to you under the Apache License,
   %% Version 2.0 (the "License"); you may not use this file
  
positive.code: -type org_info() :: {object_id(), binary()}.
  
  -export([
           reindex/2
          ]).
  
  %% @doc Sends all indexed items (clients, data bag items,
  
negative.code:                        NameIdDict), %% All dict values will be unique anyway
      BatchSize = envy:get(chef_wm, bulk_fetch_batch_size, pos_integer),
      batch_reindex(Ctx, AllIds, BatchSize, OrgInfo, Index, NameIdDict).
  
  %% @doc Reindex the objects with the specified `Ids' in the given `Index'.
  -spec reindex_by_id(DbContext :: chef_db:db_context(),
                      OrgInfo :: org_info(),
                      Index :: index(),
                      Ids :: [binary()]) -> ok.
  reindex_by_id(Ctx, {OrgId, _OrgName} = OrgInfo, Index, Ids) ->
      %% This is overkill, but workable until we have more robust reindexing
      NameIdDict = chef_db:create_name_id_dict(Ctx, Index, OrgId),
      {ok, BatchSize} = application:get_env(chef_wm, bulk_fetch_batch_size),
      batch_reindex(Ctx, Ids, BatchSize, OrgInfo, Index, NameIdDict).
  
  -spec reindex_by_name(DbContext :: chef_db:db_context(),
                        OrgInfo :: org_info(),
                        Index :: index(),
                        Names :: [binary()]) -> ok.
  reindex_by_name(Ctx, {OrgId, _OrgName} = OrgInfo, Index, Names) ->
      NameIdDict = chef_db:create_name_id_dict(Ctx, Index, OrgId),
      Ids = lists:foldl(
              fun(Name, Acc) ->
                      case dict:find(Name, NameIdDict) of
                          {ok, Id} ->
                              [Id | Acc];
                          error ->
                              error_logger:warning_msg("skipping: no id found for name ~p", [Name]),
                              Acc
                      end
              end, [], Names),
      {ok, BatchSize} = application:get_env(chef_wm, bulk_fetch_batch_size),
      batch_reindex(Ctx, Ids, BatchSize, OrgInfo, Index, NameIdDict).
  
  -spec ez_reindex_by_name(binary(), index(), [binary()]) -> ok.
  ez_reindex_by_name(OrgName, Index, Names) ->
      Ctx = make_context(),
      OrgId = chef_db:fetch_org_id(Ctx, OrgName),
      OrgInfo = {OrgId, OrgName},
      reindex_by_name(Ctx, OrgInfo, Index, Names).
  
  -spec ez_reindex_by_id(binary(), index(), [<<_:256>>]) -> ok.
  ez_reindex_by_id(OrgName, Index, Ids) ->
      Ctx = make_context(),
      OrgId = chef_db:fetch_org_id(Ctx, OrgName),
      OrgInfo = {OrgId, OrgName},
      reindex_by_id(Ctx, OrgInfo, Index, Ids).
  
  -spec ez_name_id_dict(binary(), index()) -> dict().
  ez_name_id_dict(OrgName, Index) ->
      Ctx = make_context(),
      OrgId = chef_db:fetch_org_id(Ctx, OrgName),
      chef_db:create_name_id_dict(Ctx, Index, OrgId).
  
  -spec ez_all_ids(binary(), index()) -> [binary()].
  ez_all_ids(OrgName, Index) ->
      NameIdDict = ez_name_id_dict(OrgName, Index),
      dict:fold(fun(_K, V, Acc) -> [V | Acc] end,
                [], NameIdDict).
  
  make_context() ->
      Time = string:join([ integer_to_list(I) || I <- tuple_to_list(os:timestamp())], "-"),
      ReqId = erlang:iolist_to_binary([atom_to_list(node()),
                                       "-chef_reindex-",
                                       Time
                                      ]),
      chef_db:make_context(ReqId).
  
  %% @doc Reindex objects listed in file `File'.
  %%
  %% `File' should be the path to a text file with the following format:
  %%  ```
  %%      OrgName\tIndex\tBagName\tObjectId\n
  %%  '''
  %%
  %% Where `OrgName' is the name of the organization, `Index' is one of `client',
  %% `environment', `node', `role', or `data_bag'. The value of `BagName' is ignored for all
  %% index types except for `data_bag' where the value provided is used as the data bag
  %% name. This is required so that we can distinguish between the `node' index and a data bag
  %% named `"node"'.
  %%
  %% The parser expects tab delimited lines and only reads the first four fields. This means
  %% that files contain additional columns and those columns will be ignored. This is useful
  %% if you'd like the file to contain the object name. The file is read into memory,
  %% processed into a per org per index structure and then executed. This is not meant for
  %% enormous input files.
  %%
  ez_reindex_from_file(File) ->
      {ok, Bin} = file:read_file(File),
      Lines = re:split(Bin, <<"\n">>),
      LinesWithNum = lists:zip(Lines, lists:seq(1, length(Lines))),
      OrgDict = parse_lines(LinesWithNum),
      error_logger:info_msg("reindexing data for ~p orgs~n", [dict:size(OrgDict)]),
      dict:map(
        fun(OrgName, IndexDict) ->
                error_logger:info_msg("processing org: ~s~n", [OrgName]),
                Ctx = make_context(),
                OrgId = chef_db:fetch_org_id(Ctx, OrgName),
                OrgInfo = {OrgId, OrgName},
                dict:map(
                  fun(RealIndex, Ids) ->
                          error_logger:info_msg("org: ~s, index: ~p (~p items)~n", [OrgName, RealIndex, length(Ids)]),
                          reindex_by_id(Ctx, OrgInfo, RealIndex, Ids)
                  end, IndexDict),
                error_logger:info_msg("completed org: ~s~n", [OrgName])
        end, OrgDict),
      ok.
  
  parse_lines(LinesWithNum) ->
      OrgDict = lists:foldl(
                  fun(LineInfo, OrgDict) ->
                          case parse_line(LineInfo) of
                              skip ->
                                  OrgDict;
                          {OrgName, Index, Id} ->
                                  dict:append(OrgName, {Index, Id}, OrgDict)
                          end
                  end, dict:new(), LinesWithNum),
      %% OrgDict now maps OrgName to [{Index, Id}], so now we take a second pass to convert
      %% those lists to dicts mapping Index to [Id].
      dict:from_list(
        [ {OrgName,
         lists:foldl(fun({Index, Id}, IndexDict) ->
                             dict:append(Index, Id, IndexDict)
                     end, dict:new(), IndexIds)}
          || {OrgName, IndexIds} <- dict:to_list(OrgDict) ]).
  
  parse_line({<<>>, _Num}) ->
      skip;
  parse_line({Line, Num}) ->
      try
          [OrgName, Index, BagName, Id | _ ] = re:split(Line, <<"\t">>),
          RealIndex = normalize_index(Index, BagName),
          {OrgName, RealIndex, Id}
      catch
          error:_Why ->
              erlang:error({parse_error, {line, Num}, Line})
      end.
  
  normalize_index(<<"data_bag">>, BagName) ->
      BagName;
  normalize_index(Index, _) when Index == <<"client">>;
                                 Index == <<"environment">>;
                                 Index == <<"node">>;
                                 Index == <<"role">> ->
      erlang:binary_to_atom(Index, utf8);
  normalize_index(Index, _) ->
      erlang:error({bad_index, Index}).
  
  
  %% @doc Recursively batch-process a list of database IDs by retrieving
  %% the serialized_object data from the database, processing them as
  