TEST_F(SchedulerTests, test_config_results_purge) {
  // Set a query time for now (time is only important relative to a week ago).
  auto query_time = osquery::getUnixTime();
  setDatabaseValue(kPersistentSettings, "timestamp.test_query",
                   std::to_string(query_time));
  // Store a meaningless saved query interval splay.
  setDatabaseValue(kPersistentSettings, "interval.test_query", "11");
  // Store meaningless query differential results.
  setDatabaseValue(kQueries, "test_query", "{}");

  // We do not need "THE" config instance.
  // We only need to trigger a 'purge' event, this occurs when configuration
  // content is updated by a plugin or on load.
  Config::getInstance().purge();

  // Nothing should have been purged.
  {
    std::string content;
    getDatabaseValue(kPersistentSettings, "timestamp.test_query", content);
    EXPECT_FALSE(content.empty());
  }

  {
    std::string content;
    getDatabaseValue(kPersistentSettings, "interval.test_query", content);
    EXPECT_FALSE(content.empty());
  }

  {
    std::string content;
    getDatabaseValue(kQueries, "test_query", content);
    EXPECT_FALSE(content.empty());
  }

  // Update the timestamp to have run a week and a day ago.
  query_time -= (84600 * (7 + 1));
  setDatabaseValue(kPersistentSettings, "timestamp.test_query",
                   std::to_string(query_time));

  // Trigger another purge.
  Config::getInstance().purge();
  // Now ALL 'test_query' related storage will have been purged.
  {
    std::string content;
    getDatabaseValue(kPersistentSettings, "timestamp.test_query", content);
    EXPECT_TRUE(content.empty());
  }

  {
    std::string content;
    getDatabaseValue(kPersistentSettings, "interval.test_query", content);
    EXPECT_TRUE(content.empty());
  }

  {
    std::string content;
    getDatabaseValue(kQueries, "test_query", content);
    EXPECT_TRUE(content.empty());
  }
}

void Config::recordQueryStart(const std::string& name) {
  // There should only ever be a single executing query in the schedule.
  setDatabaseValue(kPersistentSettings, kExecutingQuery, name);
  // Store the time this query name last executed for later results eviction.
  // When configuration updates occur the previous schedule is searched for
  // 'stale' query names, aka those that have week-old or longer last execute
  // timestamps. Offending queries have their database results purged.
  setDatabaseValue(
      kPersistentSettings, "timestamp." + name, std::to_string(getUnixTime()));
}

Status EventSubscriberPlugin::recordEvent(EventID& eid, EventTime time) {
  Status status;
  std::string time_value = boost::lexical_cast<std::string>(time);

  // The record is identified by the event type then module name.
  std::string index_key = "indexes." + dbNamespace();
  std::string record_key = "records." + dbNamespace();
  // The list key includes the list type (bin size) and the list ID (bin).
  std::string list_key;
  std::string list_id;

  for (const auto& time_list : kEventTimeLists) {
    // The list_id is the MOST-Specific key ID, the bin for this list.
    // If the event time was 13 and the time_list is 5 seconds, lid = 2.
    list_id = boost::lexical_cast<std::string>(time / time_list);
    // The list name identifies the 'type' of list.
    list_key = boost::lexical_cast<std::string>(time_list);
    // list_key = list_key + "." + list_id;

    {
      WriteLock lock(event_record_lock_);
      // Append the record (eid, unix_time) to the list bin.
      std::string record_value;
      status = getDatabaseValue(
          kEvents, record_key + "." + list_key + "." + list_id, record_value);

      if (record_value.length() == 0) {
        // This is a new list_id for list_key, append the ID to the indirect
        // lookup for this list_key.
        std::string index_value;
        status =
            getDatabaseValue(kEvents, index_key + "." + list_key, index_value);
        if (index_value.length() == 0) {
          // A new index.
          index_value = list_id;
        } else {
          index_value += "," + list_id;
        }
        status =
            setDatabaseValue(kEvents, index_key + "." + list_key, index_value);
        record_value = eid + ":" + time_value;
      } else {
        // Tokenize a record using ',' and the EID/time using ':'.
        record_value += "," + eid + ":" + time_value;
      }
      status = setDatabaseValue(
          kEvents, record_key + "." + list_key + "." + list_id, record_value);
      if (!status.ok()) {
        LOG(ERROR) << "Could not put Event Record key: " << record_key;
      }
    }
  }

  return Status(0, "OK");
}

TEST_F(DatabaseTests, test_scan_values) {
  setDatabaseValue(kLogs, "1", "0");
  setDatabaseValue(kLogs, "2", "0");
  setDatabaseValue(kLogs, "3", "0");

  std::vector<std::string> keys;
  auto s = scanDatabaseKeys(kLogs, keys);
  EXPECT_TRUE(s.ok());
  EXPECT_GT(keys.size(), 2U);

  keys.clear();
  s = scanDatabaseKeys(kLogs, keys, 2);
  EXPECT_TRUE(s.ok());
  EXPECT_EQ(keys.size(), 2U);
}

Status TLSLoggerPlugin::logStatus(const std::vector<StatusLogLine>& log) {
  for (const auto& item : log) {
    // Convert the StatusLogLine into ptree format, to convert to JSON.
    pt::ptree buffer;
    buffer.put("severity", (google::LogSeverity)item.severity);
    buffer.put("filename", item.filename);
    buffer.put("line", item.line);
    buffer.put("message", item.message);

    // Convert to JSON, for storing a string-representation in the database.
    std::string json;
    try {
      std::stringstream json_output;
      pt::write_json(json_output, buffer, false);
      json = json_output.str();
    } catch (const pt::json_parser::json_parser_error& e) {
      // The log could not be represented as JSON.
      return Status(1, e.what());
    }

    // Store the status line in a backing store.
    if (!json.empty()) {
      json.pop_back();
    }
    auto index = genLogIndex(false, log_index_);
    auto status = setDatabaseValue(kLogs, index, json);
    if (!status.ok()) {
      // Do not continue if any line fails.
      return status;
    }
  }

  return Status(0, "OK");
}

static void DATABASE_store(benchmark::State& state) {
  while (state.KeepRunning()) {
    setDatabaseValue(kPersistentSettings, "benchmark", "1");
  }
  // All benchmarks will share a single database handle.
  deleteDatabaseValue(kPersistentSettings, "benchmark");
}

Status carvePaths(const std::set<std::string>& paths) {
  auto guid = generateNewUUID();

  pt::ptree tree;
  tree.put("carve_guid", guid);
  tree.put("time", getUnixTime());
  tree.put("status", "STARTING");
  tree.put("sha256", "");
  tree.put("size", -1);

  if (paths.size() > 1) {
    tree.put("path", boost::algorithm::join(paths, ","));
  } else {
    tree.put("path", *(paths.begin()));
  }

  std::ostringstream os;
  pt::write_json(os, tree, false);
  auto s = setDatabaseValue(kCarveDbDomain, kCarverDBPrefix + guid, os.str());
  if (!s.ok()) {
    return s;
  } else {
    auto requestId = Distributed::getCurrentRequestId();
    Dispatcher::addService(std::make_shared<Carver>(paths, guid, requestId));
  }
  return s;
}

size_t restoreSplayedValue(const std::string& name, size_t interval) {
  // Attempt to restore a previously-calculated splay.
  std::string content;
  getDatabaseValue(kPersistentSettings, "interval." + name, content);
  if (!content.empty()) {
    // This query name existed before, check the last requested interval.
    auto details = osquery::split(content, ":");
    if (details.size() == 2) {
      long last_interval, last_splay;
      if (safeStrtol(details[0], 10, last_interval) &&
          safeStrtol(details[1], 10, last_splay)) {
        if (last_interval == static_cast<long>(interval) && last_splay > 0) {
          // This is a matching interval, use the previous splay.
          return static_cast<size_t>(last_splay);
        }
      }
    }
  }

  // If the splayed interval was not restored from the database.
  auto splay = splayValue(interval, FLAGS_schedule_splay_percent);
  content = std::to_string(interval) + ":" + std::to_string(splay);
  setDatabaseValue(kPersistentSettings, "interval." + name, content);
  return splay;
}

QueryData EventSubscriberPlugin::genTable(QueryContext& context) {
  // Stop is an unsigned (-1), our end of time equivalent.
  EventTime start = 0, stop = -1;
  if (context.constraints["time"].getAll().size() > 0) {
    // Use the 'time' constraint to optimize backing-store lookups.
    for (const auto& constraint : context.constraints["time"].getAll()) {
      EventTime expr = timeFromRecord(constraint.expr);
      if (constraint.op == EQUALS) {
        stop = start = expr;
        break;
      } else if (constraint.op == GREATER_THAN) {
        start = std::max(start, expr + 1);
      } else if (constraint.op == GREATER_THAN_OR_EQUALS) {
        start = std::max(start, expr);
      } else if (constraint.op == LESS_THAN) {
        stop = std::min(stop, expr - 1);
      } else if (constraint.op == LESS_THAN_OR_EQUALS) {
        stop = std::min(stop, expr);
      }
    }
  } else if (kToolType == OSQUERY_TOOL_DAEMON && FLAGS_events_optimize) {
    // If the daemon is querying a subscriber without a 'time' constraint and
    // allows optimization, only emit events since the last query.
    start = optimize_time_;
    optimize_time_ = getUnixTime() - 1;

    // Store the optimize time such that it can be restored if the daemon is
    // restarted.
    auto index_key = "optimize." + dbNamespace();
    setDatabaseValue(kEvents, index_key, std::to_string(optimize_time_));
  }

  return get(start, stop);
}

Status EventSubscriberPlugin::add(Row& r, EventTime event_time) {
  // Get and increment the EID for this module.
  EventID eid = getEventID();
  // Without encouraging a missing event time, do not support a 0-time.
  r["time"] = std::to_string((event_time == 0) ? getUnixTime() : event_time);
  // Serialize and store the row data, for query-time retrieval.
  std::string data;
  auto status = serializeRowJSON(r, data);
  if (!status.ok()) {
    return status;
  }
  // Then remove the newline.
  if (data.size() > 0 && data.back() == '\n') {
    data.pop_back();
  }

  // Use the last EventID and a checkpoint bucket size to periodically apply
  // buffer eviction. Eviction occurs if the total count exceeds events_max.
  if (last_eid_ % EVENTS_CHECKPOINT == 0) {
    expireCheck();
  }

  // Store the event data.
  std::string event_key = "data." + dbNamespace() + "." + eid;
  status = setDatabaseValue(kEvents, event_key, data);
  // Record the event in the indexing bins, using the index time.
  recordEvent(eid, event_time);
  return status;
}

Status ViewsConfigParserPlugin::update(const std::string& source,
                                       const ParserConfig& config) {
  auto cv = config.find("views");
  if (cv == config.end()) {
    return Status(1);
  }

  auto obj = data_.getObject();
  data_.copyFrom(cv->second.doc(), obj);
  data_.add("views", obj);

  const auto& views = data_.doc()["views"];

  // We use a restricted scope below to change the data structure from
  // an array to a set. This lets us do deletes much more efficiently
  std::vector<std::string> created_views;
  std::set<std::string> erase_views;
  {
    std::vector<std::string> old_views_vec;
    scanDatabaseKeys(kQueries, old_views_vec, kConfigViews);
    for (const auto& view : old_views_vec) {
      erase_views.insert(view.substr(kConfigViews.size()));
    }
  }

  QueryData r;
  for (const auto& view : views.GetObject()) {
    std::string name = view.name.GetString();
    std::string query = view.value.GetString();
    if (query.empty()) {
      continue;
    }

    std::string old_query = "";
    getDatabaseValue(kQueries, kConfigViews + name, old_query);
    erase_views.erase(name);
    if (old_query == query) {
      continue;
    }

    // View has been updated
    osquery::query("DROP VIEW " + name, r);
    auto s = osquery::query("CREATE VIEW " + name + " AS " + query, r);
    if (s.ok()) {
      setDatabaseValue(kQueries, kConfigViews + name, query);
    } else {
      LOG(INFO) << "Error creating view (" << name << "): " << s.getMessage();
    }
  }

  // Any views left are views that don't exist in the new configuration file
  // so we tear them down and remove them from the database.
  for (const auto& old_view : erase_views) {
    osquery::query("DROP VIEW " + old_view, r);
    deleteDatabaseValue(kQueries, kConfigViews + old_view);
  }
  return Status(0, "OK");
}

Status BufferedLogForwarder::addValueWithCount(const std::string& domain,
                                               const std::string& key,
                                               const std::string& value) {
  Status status = setDatabaseValue(domain, key, value);
  if (status.ok()) {
    buffer_count_++;
  }
  return status;
}

void saveScheduleBlacklist(const std::map<std::string, size_t>& blacklist) {
  std::string content;
  for (const auto& query : blacklist) {
    if (!content.empty()) {
      content += ":";
    }
    content += query.first + ":" + std::to_string(query.second);
  }
  setDatabaseValue(kPersistentSettings, kFailedQueries, content);
}

void TablePlugin::setCache(size_t step,
                           size_t interval,
                           const QueryData& results) {
  // Serialize QueryData and save to database.
  std::string content;
  if (!FLAGS_disable_caching && serializeQueryDataJSON(results, content)) {
    last_cached_ = step;
    last_interval_ = interval;
    setDatabaseValue(kQueries, "cache." + getName(), content);
  }
}

void DatabasePluginTests::testReset() {
  RegistryFactory::get().setActive("database", getName());
  setDatabaseValue(kLogs, "reset", "1");
  resetDatabase();

  if ("ephemeral" != getName()) {
    // The ephemeral plugin is special and does not persist after reset.
    std::string value;
    EXPECT_TRUE(getDatabaseValue(kLogs, "reset", value));
    EXPECT_EQ(value, "1");
  }
}

static void DATABASE_store_large(benchmark::State& state) {
  // Serialize the example result set into a string.
  std::string content;
  auto qd = getExampleQueryData(20, 100);
  serializeQueryDataJSON(qd, content);

  while (state.KeepRunning()) {
    setDatabaseValue(kPersistentSettings, "benchmark", content);
  }
  // All benchmarks will share a single database handle.
  deleteDatabaseValue(kPersistentSettings, "benchmark");
}

Status getHostUUID(std::string& ident) {
  // Lookup the host identifier (UUID) previously generated and stored.
  auto status = getDatabaseValue(kPersistentSettings, "hostIdentifier", ident);
  if (ident.size() == 0) {
    // There was no UUID stored in the database, generate one and store it.
    ident = osquery::generateHostUUID();
    VLOG(1) << "Using UUID " << ident << " as host identifier";
    return setDatabaseValue(kPersistentSettings, "hostIdentifier", ident);
  }

  return status;
}

void Config::purge() {
  // The first use of purge is removing expired query results.
  std::vector<std::string> saved_queries;
  scanDatabaseKeys(kQueries, saved_queries);

  const auto& schedule = this->schedule_;
  auto queryExists = [&schedule](const std::string& query_name) {
    for (const auto& pack : schedule->packs_) {
      const auto& pack_queries = pack->getSchedule();
      if (pack_queries.count(query_name)) {
        return true;
      }
    }
    return false;
  };

  RecursiveLock lock(config_schedule_mutex_);
  // Iterate over each result set in the database.
  for (const auto& saved_query : saved_queries) {
    if (queryExists(saved_query)) {
      continue;
    }

    std::string content;
    getDatabaseValue(kPersistentSettings, "timestamp." + saved_query, content);
    if (content.empty()) {
      // No timestamp is set for this query, perhaps this is the first time
      // query results expiration is applied.
      setDatabaseValue(kPersistentSettings,
                       "timestamp." + saved_query,
                       std::to_string(getUnixTime()));
      continue;
    }

    // Parse the timestamp and compare.
    size_t last_executed = 0;
    try {
      last_executed = boost::lexical_cast<size_t>(content);
    } catch (const boost::bad_lexical_cast& /* e */) {
      // Erase the timestamp as is it potentially corrupt.
      deleteDatabaseValue(kPersistentSettings, "timestamp." + saved_query);
      continue;
    }

    if (last_executed < getUnixTime() - 592200) {
      // Query has not run in the last week, expire results and interval.
      deleteDatabaseValue(kQueries, saved_query);
      deleteDatabaseValue(kPersistentSettings, "interval." + saved_query);
      deleteDatabaseValue(kPersistentSettings, "timestamp." + saved_query);
      VLOG(1) << "Expiring results for scheduled query: " << saved_query;
    }
  }
}

QueryData EventSubscriberPlugin::get(EventTime start, EventTime stop) {
  QueryData results;

  // Get the records for this time range.
  auto indexes = getIndexes(start, stop);
  auto records = getRecords(indexes);

  std::string events_key = "data." + dbNamespace();
  std::vector<std::string> mapped_records;
  for (const auto& record : records) {
    if (record.second >= start && (record.second <= stop || stop == 0)) {
      mapped_records.push_back(events_key + "." + record.first);
    }
  }

  if (FLAGS_events_optimize && !records.empty()) {
    // If records were returned save the ordered-last as the optimization EID.
    unsigned long int eidr = 0;
    if (safeStrtoul(records.back().first, 10, eidr)) {
      optimize_eid_ = static_cast<size_t>(eidr);
      auto index_key = "optimize_id." + dbNamespace();
      setDatabaseValue(kEvents, index_key, records.back().first);
    }
  }

  // Select mapped_records using event_ids as keys.
  std::string data_value;
  for (const auto& record : mapped_records) {
    Row r;
    auto status = getDatabaseValue(kEvents, record, data_value);
    if (data_value.length() == 0) {
      // There is no record here, interesting error case.
      continue;
    }
    status = deserializeRowJSON(data_value, r);
    data_value.clear();
    if (status.ok()) {
      results.push_back(std::move(r));
    }
  }

  if (getEventsExpiry() > 0) {
    // Set the expire time to NOW - "configured lifetime".
    // Index retrieval will apply the constraints checking and auto-expire.
    expire_time_ = getUnixTime() - getEventsExpiry();
  }

  return results;
}

void Config::recordQueryPerformance(const std::string& name,
                                    size_t delay,
                                    size_t size,
                                    const Row& r0,
                                    const Row& r1) {
  RecursiveLock lock(config_performance_mutex_);
  if (performance_.count(name) == 0) {
    performance_[name] = QueryPerformance();
  }

  // Grab access to the non-const schedule item.
  auto& query = performance_.at(name);
  BIGINT_LITERAL diff = 0;
  if (!r1.at("user_time").empty() && !r0.at("user_time").empty()) {
    diff = AS_LITERAL(BIGINT_LITERAL, r1.at("user_time")) -
           AS_LITERAL(BIGINT_LITERAL, r0.at("user_time"));
    if (diff > 0) {
      query.user_time += diff;
    }
  }

  if (!r1.at("system_time").empty() && !r0.at("system_time").empty()) {
    diff = AS_LITERAL(BIGINT_LITERAL, r1.at("system_time")) -
           AS_LITERAL(BIGINT_LITERAL, r0.at("system_time"));
    if (diff > 0) {
      query.system_time += diff;
    }
  }

  if (!r1.at("resident_size").empty() && !r0.at("resident_size").empty()) {
    diff = AS_LITERAL(BIGINT_LITERAL, r1.at("resident_size")) -
           AS_LITERAL(BIGINT_LITERAL, r0.at("resident_size"));
    if (diff > 0) {
      // Memory is stored as an average of RSS changes between query executions.
      query.average_memory = (query.average_memory * query.executions) + diff;
      query.average_memory = (query.average_memory / (query.executions + 1));
    }
  }

  query.wall_time += delay;
  query.output_size += size;
  query.executions += 1;
  query.last_executed = getUnixTime();

  // Clear the executing query (remove the dirty bit).
  setDatabaseValue(kPersistentSettings, kExecutingQuery, "");
}