// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#ifndef COBALT_SRC_LOCAL_AGGREGATION_AGGREGATE_STORE_H_
#define COBALT_SRC_LOCAL_AGGREGATION_AGGREGATE_STORE_H_

#include <condition_variable>
#include <memory>
#include <string>
#include <thread>

#include "src/lib/util/consistent_proto_store.h"
#include "src/lib/util/protected_fields.h"
#include "src/local_aggregation/local_aggregation.pb.h"
#include "src/logger/encoder.h"
#include "src/logger/observation_writer.h"
#include "src/logger/project_context.h"
#include "src/logger/status.h"

namespace cobalt::local_aggregation {

const std::chrono::hours kOneDay(24);

// Maximum value of |backfill_days| allowed by the constructor.
constexpr size_t kMaxAllowedBackfillDays = 1000;
// All aggregation windows larger than this number of days are ignored.
constexpr uint32_t kMaxAllowedAggregationDays = 365;
// All hourly aggregation windows larger than this number of hours are ignored.
constexpr uint32_t kMaxAllowedAggregationHours = 23;

// The current version number of the LocalAggregateStore.
constexpr uint32_t kCurrentLocalAggregateStoreVersion = 1;
// The current version number of the AggregatedObservationHistoryStore.
constexpr uint32_t kCurrentObservationHistoryStoreVersion = 0;

// The AggregateStore manages an in-memory store of aggregated Event values, indexed by report,
// day index, and other dimensions specific to the report type (e.g. event code).
//
// When GenerateObservations() is called, this data is used to generate Observations representing
// aggregates of Event values over a day, week, month, etc.
//
// This class also exposes a GarbageCollect*() and Backup*() functionality which deletes
// unnecessary data and backs up the store respectively.
class AggregateStore {
 public:
  // Constructs an AggregateStore.
  //
  // An AggregateStore maintains daily aggregates of Events in a
  // LocalAggregateStore, uses an Encoder to generate Observations for rolling
  // windows ending on a specified day index, and writes the Observations to
  // an ObservationStore using an ObservationWriter.
  //
  // encoder: the singleton instance of an Encoder on the system.
  //
  // local_aggregate_proto_store: A ConsistentProtoStore to be used by the
  // AggregateStore to store snapshots of its in-memory store of event
  // aggregates.
  //
  // obs_history_proto_store: A ConsistentProtoStore to be used by the
  // AggregateStore to store snapshots of its in-memory history of generated
  // Observations.
  //
  // backfill_days: the number of past days for which the AggregateStore
  // generates and sends Observations, in addition to a requested day index.
  // See the comment above GenerateObservations for more detail. The constructor CHECK-fails if a
  // value larger than |kMaxAllowedBackfillDays| is passed.
  AggregateStore(const logger::Encoder* encoder,
                 const logger::ObservationWriter* observation_writer,
                 util::ConsistentProtoStore* local_aggregate_proto_store,
                 util::ConsistentProtoStore* obs_history_proto_store, size_t backfill_days = 0);

  // Given a ProjectContext, MetricDefinition, and ReportDefinition checks whether a key with the
  // same customer, project, metric, and report ID already exists in the LocalAggregateStore. If
  // not, creates and inserts a new key and value. Returns kInvalidArguments if
  // creation of the key or value fails, and kOK otherwise.
  logger::Status MaybeInsertReportConfig(const logger::ProjectContext& project_context,
                                         const MetricDefinition& metric,
                                         const ReportDefinition& report);

  // Updates the LocalAggregateStore to mark the device as active on the |day_index| day for an
  // event with the given |customer_id|, |project_id|, |metric_id|, |report_id| and |event_code|.
  // Expects that MaybeInsertReportConfig() has been called previously for the ids being passed.
  // Returns kInvalidArguments if the operation fails, and kOK otherwise.
  logger::Status SetActive(uint32_t customer_id, uint32_t project_id, uint32_t metric_id,
                           uint32_t report_id, uint64_t event_code, uint32_t day_index);

  // Updates the LocalAggregateStore by adding |value| to the current daily aggregate in the bucket
  // indexed by |customer_id|, |project_id|, |metric_id|, |report_id|, |component|, |event_code| and
  // |day_index|. Expects that MaybeInsertReportConfig() has been called previously for the ids
  // being passed. Returns kInvalidArguments if the operation fails, and kOK otherwise.
  logger::Status UpdateNumericAggregate(uint32_t customer_id, uint32_t project_id,
                                        uint32_t metric_id, uint32_t report_id,
                                        const std::string& component, uint64_t event_code,
                                        uint32_t day_index, int64_t value);

  // Writes a snapshot of the LocalAggregateStore to
  // |local_aggregate_proto_store_|.
  logger::Status BackUpLocalAggregateStore();

  // Writes a snapshot of |obs_history_|to |obs_history_proto_store_|.
  logger::Status BackUpObservationHistory();

  // Removes from the LocalAggregateStore all daily aggregates that are too
  // old to contribute to their parent report's largest rolling window on the
  // day which is |backfill_days| before |day_index_utc| (if the parent
  // MetricDefinitions' time zone policy is UTC) or which is |backfill_days|
  // before |day_index_local| (if the parent MetricDefinition's time zone policy
  // is LOCAL). If |day_index_local| is 0, then we set |day_index_local| =
  // |day_index_utc|.
  //
  // If the time zone policy of a report's parent metric is UTC (resp., LOCAL)
  // and if day_index is the largest value of the |day_index_utc| (resp.,
  // |day_index_local|) argument with which GarbageCollect() has been called,
  // then the LocalAggregateStore contains the data needed to generate
  // Observations for that report for day index (day_index + k) for any k >= 0.
  logger::Status GarbageCollect(uint32_t day_index_utc, uint32_t day_index_local = 0u);

  // Generates one or more Observations for all of the registered locally
  // aggregated reports known to this AggregateStore, for rolling windows
  // ending on specified day indices.
  //
  // Each MetricDefinition specifies a time zone policy, which determines the
  // day index for which an Event associated with that MetricDefinition is
  // logged. For all MetricDefinitions whose Events are logged with respect to
  // UTC, this method generates Observations for rolling windows ending on
  // |final_day_index_utc|. For all MetricDefinitions whose Events are logged
  // with respect to local time, this method generates Observations for rolling
  // windows ending on |final_day_index_local|. If |final_day_index_local| is
  // 0, then we set |final_day_index_local| = |final_day_index_utc|.
  //
  // The generated Observations are written to the |observation_writer| passed
  // to the constructor.
  //
  // This class maintains a history of generated Observations and this method
  // additionally performs backfill: Observations are also generated for
  // rolling windows ending on any day in the interval [final_day_index -
  // backfill_days, final_day_index] (where final_day_index is either
  // final_day_index_utc or final_day_index_local, depending on the time zone
  // policy of the associated MetricDefinition), if this history indicates they
  // were not previously generated. Does not generate duplicate Observations
  // when called multiple times with the same day index.
  //
  // Observations are not generated for aggregation windows larger than
  // |kMaxAllowedAggregationDays|. Hourly windows are not yet supported.
  logger::Status GenerateObservations(uint32_t final_day_index_utc,
                                      uint32_t final_day_index_local = 0u);

 private:
  friend class AggregateStoreTest;
  friend class EventAggregatorTest;
  friend class EventAggregatorManagerTest;
  friend class TestEventAggregatorManager;

  // Make a LocalAggregateStore which is empty except that its version number is set to |version|.
  LocalAggregateStore MakeNewLocalAggregateStore(
      uint32_t version = kCurrentLocalAggregateStoreVersion);

  // Make an AggregatedObservationHistoryStore which is empty except that its version number is set
  // to |version|.
  AggregatedObservationHistoryStore MakeNewObservationHistoryStore(
      uint32_t version = kCurrentObservationHistoryStoreVersion);

  // The LocalAggregateStore or AggregatedObservationHistoryStore may need to be changed in ways
  // which are structurally but not semantically backwards-compatible. In other words, the meaning
  // to the AggregateStore of a field in the LocalAggregateStore might change. An example is that
  // we might deprecate one field while introducing a new one.
  //
  // The MaybeUpgrade*Store methods allow the AggregateStore to update the contents of its stored
  // protos from previously meaningful values to currently meaningful values. (For example, a
  // possible implementation would move the contents of a deprecated field to the replacement
  // field.)
  //
  // These methods are called by the AggregateStore constructor immediately after reading in stored
  // protos from disk in order to ensure that proto contents have the expected semantics.
  //
  // The method first checks the version number of the store. If the version number is equal to
  // |kCurrentLocalAggregateStoreVersion| or |kCurrentObservationHistoryStoreVersion|
  // (respectively), returns an OK status. Otherwise, if it is possible to upgrade the store to the
  // current version, does so and returns an OK status. If not, logs an error and returns
  // kInvalidArguments. If a non-OK status is returned, the caller should discard the contents of
  // |store| and replace it with an empty store at the current version. The MakeNew*Store() methods
  // may be used to create the new store.
  logger::Status MaybeUpgradeLocalAggregateStore(LocalAggregateStore* store);
  logger::Status MaybeUpgradeObservationHistoryStore(AggregatedObservationHistoryStore* store);

  // Returns the most recent day index for which an Observation was generated
  // for a given UNIQUE_N_DAY_ACTIVES report, event code, and day-based aggregation window,
  // according to |obs_history_|. Returns 0 if no Observation has been generated
  // for the given arguments.
  uint32_t UniqueActivesLastGeneratedDayIndex(const std::string& report_key, uint32_t event_code,
                                              uint32_t aggregation_days) const;

  // Returns the most recent day index for which an Observation was generated for a given
  // PER_DEVICE_NUMERIC_STATS or PER_DEVICE_HISTOGRAM report, component, event code, and day-based
  // aggregation window, according to |obs_history_|. Returns 0 if no Observation has been generated
  // for the given arguments.
  uint32_t PerDeviceNumericLastGeneratedDayIndex(const std::string& report_key,
                                                 const std::string& component, uint32_t event_code,
                                                 uint32_t aggregation_days) const;

  // Returns the most recent day index for which a
  // ReportParticipationObservation was generated for a given report, according
  // to |obs_history_|. Returns 0 if no Observation has been generated for the
  // given arguments.
  uint32_t ReportParticipationLastGeneratedDayIndex(const std::string& report_key) const;

  // For a fixed report of type UNIQUE_N_DAY_ACTIVES, generates an Observation
  // for each event code of the parent metric, for each day-based aggregation window of the
  // report ending on |final_day_index|, unless an Observation with those parameters was generated
  // in the past. Also generates Observations for days in the backfill period if needed. Writes the
  // Observations to an ObservationStore via the ObservationWriter that was passed to the
  // constructor.
  //
  // Observations are not generated for aggregation windows larger than
  // |kMaxAllowedAggregationDays|. Hourly windows are not yet supported.
  logger::Status GenerateUniqueActivesObservations(logger::MetricRef metric_ref,
                                                   const std::string& report_key,
                                                   const ReportAggregates& report_aggregates,
                                                   uint32_t num_event_codes,
                                                   uint32_t final_day_index);

  // Helper method called by GenerateUniqueActivesObservations() to generate
  // and write a single Observation.
  logger::Status GenerateSingleUniqueActivesObservation(logger::MetricRef metric_ref,
                                                        const ReportDefinition* report,
                                                        uint32_t obs_day_index, uint32_t event_code,
                                                        const OnDeviceAggregationWindow& window,
                                                        bool was_active) const;

  // For a fixed report of type PER_DEVICE_NUMERIC_STATS or PER_DEVICE_HISTOGRAM, generates a
  // PerDeviceNumericObservation and PerDeviceHistogramObservation respectively for each
  // tuple (component, event code, aggregation_window) for which a numeric event was logged for that
  // event code and component during the window of that size ending on |final_day_index|, unless an
  // Observation with those parameters has been generated in the past. The value of the Observation
  // is the sum, max, or min (depending on the aggregation_type field of the report definition) of
  // all numeric events logged for that report during the window. Also generates observations for
  // days in the backfill period if needed.
  //
  // In addition to PerDeviceNumericObservations or PerDeviceHistogramObservation , generates
  // a ReportParticipationObservation for |final_day_index| and any needed days in the backfill
  // period. These ReportParticipationObservations are used by the report generators to infer the
  // fleet-wide number of devices for which the sum of numeric events associated to each tuple
  // (component, event code, window size) was zero.
  //
  // Observations are not generated for aggregation windows larger than
  // |kMaxAllowedAggregationWindowSize|.
  logger::Status GenerateObsFromNumericAggregates(logger::MetricRef metric_ref,
                                                  const std::string& report_key,
                                                  const ReportAggregates& report_aggregates,
                                                  uint32_t final_day_index);

  // Helper method called by GenerateObsFromNumericAggregates() to generate and write a single
  // Observation with value |value|. The method will produce a PerDeviceNumericObservation or
  // PerDeviceHistogramObservation  depending on whether the report type is
  // PER_DEVICE_NUMERIC_STATS or PER_DEVICE_HISTOGRAM respectively.
  logger::Status GenerateSinglePerDeviceNumericObservation(
      logger::MetricRef metric_ref, const ReportDefinition* report, uint32_t obs_day_index,
      const std::string& component, uint32_t event_code, const OnDeviceAggregationWindow& window,
      int64_t value) const;

  // Helper method called by GenerateObsFromNumericAggregates() to generate and write a single
  // Observation with value |value|.
  logger::Status GenerateSinglePerDeviceHistogramObservation(
      logger::MetricRef metric_ref, const ReportDefinition* report, uint32_t obs_day_index,
      const std::string& component, uint32_t event_code, const OnDeviceAggregationWindow& window,
      int64_t value) const;

  // Helper method called by GenerateObsFromNumericAggregates() to generate and write a single
  // ReportParticipationObservation.
  logger::Status GenerateSingleReportParticipationObservation(logger::MetricRef metric_ref,
                                                              const ReportDefinition* report,
                                                              uint32_t obs_day_index) const;

  LocalAggregateStore CopyLocalAggregateStore() {
    auto local_aggregate_store = protected_aggregate_store_.lock()->local_aggregate_store;
    return local_aggregate_store;
  }

  struct AggregateStoreFields {
    LocalAggregateStore local_aggregate_store;
  };

  const logger::Encoder* encoder_;
  const logger::ObservationWriter* observation_writer_;
  util::ConsistentProtoStore* local_aggregate_proto_store_;
  util::ConsistentProtoStore* obs_history_proto_store_;
  util::ProtectedFields<AggregateStoreFields> protected_aggregate_store_;
  // Not protected by a mutex. Should only be accessed by the Event Aggregator's |worker_thread_|.
  AggregatedObservationHistoryStore obs_history_;
  size_t backfill_days_ = 0;
};

}  // namespace cobalt::local_aggregation

#endif  // COBALT_SRC_LOCAL_AGGREGATION_AGGREGATE_STORE_H_