// 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.

#include "logger/event_aggregator.h"

#include <algorithm>
#include <map>
#include <string>
#include <utility>
#include <vector>

#include "algorithms/rappor/rappor_config_helper.h"
#include "config/metric_definition.pb.h"
#include "config/packed_event_codes.h"
#include "logger/project_context.h"
#include "util/datetime_util.h"
#include "util/proto_util.h"
#include "util/status.h"

namespace cobalt {

using google::protobuf::RepeatedField;
using rappor::RapporConfigHelper;
using util::ConsistentProtoStore;
using util::SerializeToBase64;
using util::StatusCode;
using util::SystemClock;
using util::TimeToDayIndex;

namespace logger {

namespace {

// Creates an AggregationConfig from a ProjectContext, MetricDefinition, and
// ReportDefinition and populates the aggregation_config field of a specified
// ReportAggregates. Also sets the type of the ReportAggregates based on the
// ReportDefinition's type.
bool PopulateReportAggregates(const ProjectContext& project_context,
                              const MetricDefinition& metric,
                              const ReportDefinition& report,
                              ReportAggregates* report_aggregates) {
  if (report.window_size_size() == 0) {
    LOG(ERROR) << "Report must have at least one window size.";
    return false;
  }
  AggregationConfig* aggregation_config =
      report_aggregates->mutable_aggregation_config();
  *aggregation_config->mutable_project() = project_context.project();
  *aggregation_config->mutable_metric() =
      *project_context.GetMetric(metric.id());
  *aggregation_config->mutable_report() = report;

  std::vector<uint32_t> window_sizes;
  for (const uint32_t window_size : report.window_size()) {
    if (window_size == 0 ||
        window_size > EventAggregator::kMaxAllowedAggregationWindowSize) {
      LOG(ERROR) << "Window size must be positive and cannot exceed "
                 << EventAggregator::kMaxAllowedAggregationWindowSize;
      return false;
    }
    window_sizes.push_back(window_size);
  }
  std::sort(window_sizes.begin(), window_sizes.end());
  for (const uint32_t window_size : window_sizes) {
    aggregation_config->add_window_size(window_size);
  }

  switch (report.report_type()) {
    case ReportDefinition::UNIQUE_N_DAY_ACTIVES: {
      report_aggregates->set_allocated_unique_actives_aggregates(
          new UniqueActivesReportAggregates);
      return true;
    }
    case ReportDefinition::PER_DEVICE_NUMERIC_STATS: {
      report_aggregates->set_allocated_numeric_aggregates(
          new PerDeviceNumericAggregates);
      return true;
    }
    default:
      return false;
  }
}

// Populates a ReportAggregationKey proto message and then populates a string
// with the base64 encoding of the serialized proto.
bool PopulateReportKey(uint32_t customer_id, uint32_t project_id,
                       uint32_t metric_id, uint32_t report_id,
                       std::string* key) {
  ReportAggregationKey key_data;
  key_data.set_customer_id(customer_id);
  key_data.set_project_id(project_id);
  key_data.set_metric_id(metric_id);
  key_data.set_report_id(report_id);
  return SerializeToBase64(key_data, key);
}

// Given a ProjectContext, MetricDefinition, and ReportDefinition and a pointer
// to the LocalAggregateStore, 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. The caller should hold
// the mutex protecting the LocalAggregateStore.
Status MaybeInsertReportConfigLocked(const ProjectContext& project_context,
                                     const MetricDefinition& metric,
                                     const ReportDefinition& report,
                                     LocalAggregateStore* store) {
  std::string key;
  if (!PopulateReportKey(project_context.project().customer_id(),
                         project_context.project().project_id(), metric.id(),
                         report.id(), &key)) {
    return kInvalidArguments;
  }
  ReportAggregates report_aggregates;
  if (store->by_report_key().count(key) == 0) {
    if (!PopulateReportAggregates(project_context, metric, report,
                                  &report_aggregates)) {
      return kInvalidArguments;
    }
    (*store->mutable_by_report_key())[key] = report_aggregates;
  }
  return kOK;
}

RepeatedField<uint32_t> UnpackEventCodesProto(uint64_t packed_event_codes) {
  RepeatedField<uint32_t> fields;
  for (auto code : config::UnpackEventCodes(packed_event_codes)) {
    *fields.Add() = code;
  }
  return fields;
}

}  // namespace

EventAggregator::EventAggregator(
    const Encoder* encoder, const ObservationWriter* observation_writer,
    ConsistentProtoStore* local_aggregate_proto_store,
    ConsistentProtoStore* obs_history_proto_store, const size_t backfill_days,
    const std::chrono::seconds aggregate_backup_interval,
    const std::chrono::seconds generate_obs_interval,
    const std::chrono::seconds gc_interval)
    : encoder_(encoder),
      observation_writer_(observation_writer),
      local_aggregate_proto_store_(local_aggregate_proto_store),
      obs_history_proto_store_(obs_history_proto_store) {
  CHECK_LE(aggregate_backup_interval.count(), generate_obs_interval.count())
      << "aggregate_backup_interval must be less than or equal to "
         "generate_obs_interval";
  CHECK_LE(aggregate_backup_interval.count(), gc_interval.count())
      << "aggregate_backup_interval must be less than or equal to gc_interval";
  CHECK_LE(backfill_days, kMaxAllowedBackfillDays)
      << "backfill_days must be less than or equal to "
      << kMaxAllowedBackfillDays;
  aggregate_backup_interval_ = aggregate_backup_interval;
  generate_obs_interval_ = generate_obs_interval;
  gc_interval_ = gc_interval;
  backfill_days_ = backfill_days;
  auto locked = protected_aggregate_store_.lock();
  auto restore_aggregates_status =
      local_aggregate_proto_store_->Read(&(locked->local_aggregate_store));
  switch (restore_aggregates_status.error_code()) {
    case StatusCode::OK: {
      VLOG(4) << "Read LocalAggregateStore from disk.";
      break;
    }
    case StatusCode::NOT_FOUND: {
      VLOG(4) << "No file found for local_aggregate_proto_store. Proceeding "
                 "with empty LocalAggregateStore. File will be created on "
                 "first snapshot of the LocalAggregateStore.";
      break;
    }
    default: {
      LOG(ERROR)
          << "Read to local_aggregate_proto_store failed with status code: "
          << restore_aggregates_status.error_code()
          << "\nError message: " << restore_aggregates_status.error_message()
          << "\nError details: " << restore_aggregates_status.error_details()
          << "\nProceeding with empty LocalAggregateStore.";
      locked->local_aggregate_store = LocalAggregateStore();
    }
  }
  auto restore_history_status = obs_history_proto_store_->Read(&obs_history_);
  switch (restore_history_status.error_code()) {
    case StatusCode::OK: {
      VLOG(4) << "Read AggregatedObservationHistoryStore from disk.";
      break;
    }
    case StatusCode::NOT_FOUND: {
      VLOG(4) << "No file found for obs_history_proto_store. Proceeding "
                 "with empty AggregatedObservationHistoryStore. File will be "
                 "created on first snapshot of the "
                 "AggregatedObservationHistoryStore.";
      break;
    }
    default: {
      LOG(ERROR) << "Read to obs_history_proto_store failed with status code: "
                 << restore_history_status.error_code() << "\nError message: "
                 << restore_history_status.error_message()
                 << "\nError details: "
                 << restore_history_status.error_details()
                 << "\nProceeding with empty AggregatedObservationHistory.";
      obs_history_ = AggregatedObservationHistoryStore();
    }
  }
  clock_.reset(new SystemClock());
}

void EventAggregator::Start() {
  auto locked = protected_shutdown_flag_.lock();
  locked->shut_down = false;
  std::thread t([this] { this->Run(); });
  worker_thread_ = std::move(t);
}

// TODO(pesk): update the EventAggregator's view of a Metric
// or ReportDefinition when appropriate.
Status EventAggregator::UpdateAggregationConfigs(
    const ProjectContext& project_context) {
  auto locked = protected_aggregate_store_.lock();
  Status status;
  for (const auto& metric : project_context.metrics()) {
    switch (metric.metric_type()) {
      case MetricDefinition::EVENT_OCCURRED: {
        for (const auto& report : metric.reports()) {
          switch (report.report_type()) {
            case ReportDefinition::UNIQUE_N_DAY_ACTIVES: {
              status = MaybeInsertReportConfigLocked(
                  project_context, metric, report,
                  &(locked->local_aggregate_store));
              if (status != kOK) {
                return status;
              }
            }
            default:
              continue;
          }
        }
      }
      case MetricDefinition::EVENT_COUNT:
      case MetricDefinition::ELAPSED_TIME:
      case MetricDefinition::FRAME_RATE:
      case MetricDefinition::MEMORY_USAGE: {
        for (const auto& report : metric.reports()) {
          switch (report.report_type()) {
            case ReportDefinition::PER_DEVICE_NUMERIC_STATS: {
              status = MaybeInsertReportConfigLocked(
                  project_context, metric, report,
                  &(locked->local_aggregate_store));
              if (status != kOK) {
                return status;
              }
            }
            default:
              continue;
          }
        }
      }
      default:
        continue;
    }
  }
  return kOK;
}

Status EventAggregator::LogUniqueActivesEvent(uint32_t report_id,
                                              EventRecord* event_record) {
  if (!event_record->event->has_occurrence_event()) {
    LOG(ERROR) << "EventAggregator::LogUniqueActivesEvent can only "
                  "accept OccurrenceEvents.";
    return kInvalidArguments;
  }
  std::string key;
  if (!PopulateReportKey(event_record->metric->customer_id(),
                         event_record->metric->project_id(),
                         event_record->metric->id(), report_id, &key)) {
    return kInvalidArguments;
  }
  auto locked = protected_aggregate_store_.lock();
  auto aggregates =
      locked->local_aggregate_store.mutable_by_report_key()->find(key);
  if (aggregates ==
      locked->local_aggregate_store.mutable_by_report_key()->end()) {
    LOG(ERROR) << "The Local Aggregate Store received an unexpected key.";
    return kInvalidArguments;
  }
  if (!aggregates->second.has_unique_actives_aggregates()) {
    LOG(ERROR) << "The local aggregates for this report key are not of type "
                  "UniqueActivesReportAggregates.";
    return kInvalidArguments;
  }
  (*(*aggregates->second.mutable_unique_actives_aggregates()
          ->mutable_by_event_code())[event_record->event->occurrence_event()
                                         .event_code()]
        .mutable_by_day_index())[event_record->event->day_index()]
      .mutable_activity_daily_aggregate()
      ->set_activity_indicator(true);
  return kOK;
}

Status EventAggregator::LogCountEvent(uint32_t report_id,
                                      EventRecord* event_record) {
  if (event_record->event->type_case() != Event::kCountEvent) {
    LOG(ERROR) << "EventAggregator: LogCountEvent can only accept "
                  "CountEvents.";
    return kInvalidArguments;
  }
  std::string key;
  if (!PopulateReportKey(event_record->metric->customer_id(),
                         event_record->metric->project_id(),
                         event_record->metric->id(), report_id, &key)) {
    return kInvalidArguments;
  }
  const CountEvent& count_event = event_record->event->count_event();
  return LogNumericEvent(
      key, event_record->event->day_index(), count_event.component(),
      config::PackEventCodes(count_event.event_code()), count_event.count());
}

Status EventAggregator::LogElapsedTimeEvent(uint32_t report_id,
                                            EventRecord* event_record) {
  if (event_record->event->type_case() != Event::kElapsedTimeEvent) {
    LOG(ERROR) << "EventAggregator: LogElapsedTimeEvent can only accept "
                  "ElapsedTimeEvents.";
    return kInvalidArguments;
  }
  std::string key;
  if (!PopulateReportKey(event_record->metric->customer_id(),
                         event_record->metric->project_id(),
                         event_record->metric->id(), report_id, &key)) {
    return kInvalidArguments;
  }
  const ElapsedTimeEvent& elapsed_time_event =
      event_record->event->elapsed_time_event();
  return LogNumericEvent(
      key, event_record->event->day_index(), elapsed_time_event.component(),
      config::PackEventCodes(elapsed_time_event.event_code()),
      elapsed_time_event.elapsed_micros());
}

Status EventAggregator::LogFrameRateEvent(uint32_t report_id,
                                          EventRecord* event_record) {
  if (event_record->event->type_case() != Event::kFrameRateEvent) {
    LOG(ERROR) << "EventAggregator: LogFrameRateEvent can only accept "
                  "FrameRateEvents.";
    return kInvalidArguments;
  }
  std::string key;
  if (!PopulateReportKey(event_record->metric->customer_id(),
                         event_record->metric->project_id(),
                         event_record->metric->id(), report_id, &key)) {
    return kInvalidArguments;
  }
  const FrameRateEvent& frame_rate_event =
      event_record->event->frame_rate_event();
  return LogNumericEvent(key, event_record->event->day_index(),
                         frame_rate_event.component(),
                         config::PackEventCodes(frame_rate_event.event_code()),
                         frame_rate_event.frames_per_1000_seconds());
}

Status EventAggregator::LogMemoryUsageEvent(uint32_t report_id,
                                            EventRecord* event_record) {
  if (event_record->event->type_case() != Event::kMemoryUsageEvent) {
    LOG(ERROR) << "EventAggregator: LogMemoryUsageEvent can only accept "
                  "MemoryUsageEvents.";
    return kInvalidArguments;
  }
  std::string key;
  if (!PopulateReportKey(event_record->metric->customer_id(),
                         event_record->metric->project_id(),
                         event_record->metric->id(), report_id, &key)) {
    return kInvalidArguments;
  }
  const MemoryUsageEvent& memory_usage_event =
      event_record->event->memory_usage_event();
  return LogNumericEvent(
      key, event_record->event->day_index(), memory_usage_event.component(),
      config::PackEventCodes(memory_usage_event.event_code()),
      memory_usage_event.bytes());
}

Status EventAggregator::LogNumericEvent(const std::string& report_key,
                                        uint32_t day_index,
                                        const std::string& component,
                                        uint64_t event_code, int64_t value) {
  auto locked = protected_aggregate_store_.lock();
  auto aggregates =
      locked->local_aggregate_store.mutable_by_report_key()->find(report_key);
  if (aggregates ==
      locked->local_aggregate_store.mutable_by_report_key()->end()) {
    LOG(ERROR) << "The Local Aggregate Store received an unexpected key.";
    return kInvalidArguments;
  }
  if (!aggregates->second.has_numeric_aggregates()) {
    LOG(ERROR) << "The local aggregates for this report key are not of a "
                  "compatible type.";
    return kInvalidArguments;
  }
  auto aggregates_by_day = (*(*aggregates->second.mutable_numeric_aggregates()
                                   ->mutable_by_component())[component]
                                 .mutable_by_event_code())[event_code]
                               .mutable_by_day_index();
  bool first_event_today =
      ((*aggregates_by_day).find(day_index) == aggregates_by_day->end());
  auto day_aggregate =
      (*aggregates_by_day)[day_index].mutable_numeric_daily_aggregate();
  const auto& aggregation_type =
      aggregates->second.aggregation_config().report().aggregation_type();
  switch (aggregation_type) {
    case ReportDefinition::SUM:
      day_aggregate->set_value(value + day_aggregate->value());
      return kOK;
    case ReportDefinition::MAX:
      day_aggregate->set_value(std::max(value, day_aggregate->value()));
      return kOK;
    case ReportDefinition::MIN:
      if (first_event_today) {
        day_aggregate->set_value(value);
      } else {
        day_aggregate->set_value(std::min(value, day_aggregate->value()));
      }
      return kOK;
    default:
      LOG(ERROR) << "Unexpected aggregation type " << aggregation_type;
      return kInvalidArguments;
  }
}

Status EventAggregator::GenerateObservationsNoWorker(
    uint32_t final_day_index_utc, uint32_t final_day_index_local) {
  if (worker_thread_.joinable()) {
    LOG(ERROR) << "GenerateObservationsNoWorker() was called while "
                  "worker thread was running.";
    return kOther;
  }
  return GenerateObservations(final_day_index_utc, final_day_index_local);
}

Status EventAggregator::BackUpLocalAggregateStore() {
  // Lock, copy the LocalAggregateStore, and release the lock. Write the copy
  // to |local_aggregate_proto_store_|.
  auto local_aggregate_store = CopyLocalAggregateStore();
  auto status = local_aggregate_proto_store_->Write(local_aggregate_store);
  if (!status.ok()) {
    LOG(ERROR) << "Failed to back up the LocalAggregateStore with error code: "
               << status.error_code()
               << "\nError message: " << status.error_message()
               << "\nError details: " << status.error_details();
    return kOther;
  }
  return kOK;
}

Status EventAggregator::BackUpObservationHistory() {
  auto status = obs_history_proto_store_->Write(obs_history_);
  if (!status.ok()) {
    LOG(ERROR) << "Failed to back up the AggregatedObservationHistoryStore. "
                  "::cobalt::util::Status error code: "
               << status.error_code()
               << "\nError message: " << status.error_message()
               << "\nError details: " << status.error_details();
    return kOther;
  }
  return kOK;
}

void EventAggregator::ShutDown() {
  if (worker_thread_.joinable()) {
    {
      auto locked = protected_shutdown_flag_.lock();
      locked->shut_down = true;
      locked->shutdown_notifier.notify_all();
    }
    worker_thread_.join();
  } else {
    protected_shutdown_flag_.lock()->shut_down = true;
  }
}

void EventAggregator::Run() {
  auto current_time = clock_->now();
  // Schedule Observation generation to happen in the first cycle.
  next_generate_obs_ = current_time;
  // Schedule garbage collection to happen |gc_interval_| seconds from now.
  next_gc_ = current_time + gc_interval_;
  // Acquire the mutex protecting the shutdown flag and condition variable.
  auto locked = protected_shutdown_flag_.lock();
  while (true) {
    // If shutdown has been requested, back up the LocalAggregateStore and
    // exit.
    if (locked->shut_down) {
      BackUpLocalAggregateStore();
      return;
    }
    // Sleep until the next scheduled backup of the LocalAggregateStore or
    // until notified of shutdown. Back up the LocalAggregateStore after
    // waking.
    auto shutdown_requested = locked.wait_for_with(
        &(locked->shutdown_notifier), aggregate_backup_interval_,
        [&locked]() { return locked->shut_down; });
    BackUpLocalAggregateStore();
    // If the worker thread was woken up by a shutdown request, exit.
    // Otherwise, complete any scheduled Observation generation and garbage
    // collection.
    if (shutdown_requested) {
      return;
    }
    // Check whether it is time to generate Observations or to garbage-collect
    // the LocalAggregate store. If so, do that task and schedule the next
    // occurrence.
    DoScheduledTasks(clock_->now());
  }
}

void EventAggregator::DoScheduledTasks(
    std::chrono::system_clock::time_point current_time) {
  auto current_time_t = std::chrono::system_clock::to_time_t(current_time);
  auto current_day_index_utc =
      TimeToDayIndex(current_time_t, MetricDefinition::UTC);
  auto current_day_index_local =
      TimeToDayIndex(current_time_t, MetricDefinition::LOCAL);
  // Skip the tasks (but do schedule a retry) if either day index is too small.
  uint32_t min_allowed_day_index =
      kMaxAllowedAggregationWindowSize + backfill_days_ + 1;
  bool skip_tasks = (current_day_index_utc < min_allowed_day_index ||
                     current_day_index_local < min_allowed_day_index);

  if (current_time >= next_generate_obs_) {
    next_generate_obs_ += generate_obs_interval_;
    if (skip_tasks) {
      LOG_FIRST_N(ERROR, 10)
          << "EventAggregator is skipping Observation generation because the "
             "current day index is too small.";
    } else {
      auto obs_status = GenerateObservations(current_day_index_utc - 1,
                                             current_day_index_local - 1);
      if (obs_status == kOK) {
        BackUpObservationHistory();
      } else {
        LOG(ERROR) << "GenerateObservations failed with status: " << obs_status;
      }
    }
  }
  if (current_time >= next_gc_) {
    next_gc_ += gc_interval_;
    if (skip_tasks) {
      LOG_FIRST_N(ERROR, 10)
          << "EventAggregator is skipping garbage collection because the "
             "current day index is too small.";
    } else {
      auto gc_status = GarbageCollect(current_day_index_utc - 1,
                                      current_day_index_local - 1);
      if (gc_status == kOK) {
        BackUpLocalAggregateStore();
      } else {
        LOG(ERROR) << "GarbageCollect failed with status: " << gc_status;
      }
    }
  }
}

////////////////////// GarbageCollect and helper functions /////////////////////

namespace {

void GarbageCollectUniqueActivesReportAggregates(
    uint32_t day_index, uint32_t max_window_size, uint32_t backfill_days,
    UniqueActivesReportAggregates* report_aggregates) {
  auto map_by_event_code = report_aggregates->mutable_by_event_code();
  for (auto event_code = map_by_event_code->begin();
       event_code != map_by_event_code->end();) {
    auto map_by_day = event_code->second.mutable_by_day_index();
    for (auto day = map_by_day->begin(); day != map_by_day->end();) {
      if (day->first <= day_index - backfill_days - max_window_size) {
        day = map_by_day->erase(day);
      } else {
        ++day;
      }
    }
    if (map_by_day->empty()) {
      event_code = map_by_event_code->erase(event_code);
    } else {
      ++event_code;
    }
  }
}

void GarbageCollectNumericReportAggregates(
    uint32_t day_index, uint32_t max_window_size, uint32_t backfill_days,
    PerDeviceNumericAggregates* report_aggregates) {
  auto map_by_component = report_aggregates->mutable_by_component();
  for (auto component = map_by_component->begin();
       component != map_by_component->end();) {
    auto map_by_event_code = component->second.mutable_by_event_code();
    for (auto event_code = map_by_event_code->begin();
         event_code != map_by_event_code->end();) {
      auto map_by_day = event_code->second.mutable_by_day_index();
      for (auto day = map_by_day->begin(); day != map_by_day->end();) {
        if (day->first <= day_index - backfill_days - max_window_size) {
          day = map_by_day->erase(day);
        } else {
          ++day;
        }
      }
      if (map_by_day->empty()) {
        event_code = map_by_event_code->erase(event_code);
      } else {
        ++event_code;
      }
    }
    if (map_by_event_code->empty()) {
      component = map_by_component->erase(component);
    } else {
      ++component;
    }
  }
}

}  // namespace

Status EventAggregator::GarbageCollect(uint32_t day_index_utc,
                                       uint32_t day_index_local) {
  if (day_index_local == 0u) {
    day_index_local = day_index_utc;
  }
  CHECK_LT(day_index_utc, UINT32_MAX);
  CHECK_LT(day_index_local, UINT32_MAX);
  CHECK_GE(day_index_utc, kMaxAllowedAggregationWindowSize + backfill_days_);
  CHECK_GE(day_index_local, kMaxAllowedAggregationWindowSize + backfill_days_);

  auto locked = protected_aggregate_store_.lock();
  for (auto pair : locked->local_aggregate_store.by_report_key()) {
    uint32_t day_index;
    switch (pair.second.aggregation_config().metric().time_zone_policy()) {
      case MetricDefinition::UTC: {
        day_index = day_index_utc;
        break;
      }
      case MetricDefinition::LOCAL: {
        day_index = day_index_local;
        break;
      }
      default:
        LOG(ERROR) << "The TimeZonePolicy of this MetricDefinition is invalid.";
        return kInvalidConfig;
    }
    uint32_t max_window_size = pair.second.aggregation_config().window_size(
        pair.second.aggregation_config().window_size_size() - 1);
    // For each ReportAggregates, descend to and iterate over the sub-map of
    // local aggregates keyed by day index. Keep buckets with day indices
    // greater than |day_index| - |backfill_days_| - |max_window_size|, and
    // remove all buckets with smaller day indices.
    switch (pair.second.type_case()) {
      case ReportAggregates::kUniqueActivesAggregates: {
        GarbageCollectUniqueActivesReportAggregates(
            day_index, max_window_size, backfill_days_,
            locked->local_aggregate_store.mutable_by_report_key()
                ->at(pair.first)
                .mutable_unique_actives_aggregates());
        break;
      }
      case ReportAggregates::kNumericAggregates: {
        GarbageCollectNumericReportAggregates(
            day_index, max_window_size, backfill_days_,
            locked->local_aggregate_store.mutable_by_report_key()
                ->at(pair.first)
                .mutable_numeric_aggregates());
        break;
      }
      default:
        continue;
    }
  }
  return kOK;
}

Status EventAggregator::GenerateObservations(uint32_t final_day_index_utc,
                                             uint32_t final_day_index_local) {
  if (final_day_index_local == 0u) {
    final_day_index_local = final_day_index_utc;
  }
  CHECK_LT(final_day_index_utc, UINT32_MAX);
  CHECK_LT(final_day_index_local, UINT32_MAX);
  CHECK_GE(final_day_index_utc,
           kMaxAllowedAggregationWindowSize + backfill_days_);
  CHECK_GE(final_day_index_local,
           kMaxAllowedAggregationWindowSize + backfill_days_);

  // Lock, copy the LocalAggregateStore, and release the lock. Use the copy to
  // generate observations.
  auto local_aggregate_store = CopyLocalAggregateStore();
  for (auto pair : local_aggregate_store.by_report_key()) {
    const auto& config = pair.second.aggregation_config();

    const auto& metric = config.metric();
    auto metric_ref = MetricRef(&config.project(), &metric);
    uint32_t final_day_index;
    switch (metric.time_zone_policy()) {
      case MetricDefinition::UTC: {
        final_day_index = final_day_index_utc;
        break;
      }
      case MetricDefinition::LOCAL: {
        final_day_index = final_day_index_local;
        break;
      }
      default:
        LOG(ERROR) << "The TimeZonePolicy of this MetricDefinition is invalid.";
        return kInvalidConfig;
    }

    const auto& report = config.report();
    // PopulateReportAggregates has checked that there is at least one window
    // size and that all window sizes are positive and <=
    // kMaxAllowedAggregationWindowSize, and has sorted the elements of
    // config.window_size() in increasing order.
    CHECK_GT(config.window_size_size(), 0);
    auto max_window_size = config.window_size(config.window_size_size() - 1);
    CHECK_GT(max_window_size, 0);
    CHECK_GE(final_day_index, max_window_size);

    switch (metric.metric_type()) {
      case MetricDefinition::EVENT_OCCURRED: {
        auto num_event_codes =
            RapporConfigHelper::BasicRapporNumCategories(metric);

        switch (report.report_type()) {
          case ReportDefinition::UNIQUE_N_DAY_ACTIVES: {
            auto status = GenerateUniqueActivesObservations(
                metric_ref, pair.first, pair.second, num_event_codes,
                final_day_index);
            if (status != kOK) {
              return status;
            }
            break;
          }
          default:
            continue;
        }
        break;
      }
      case MetricDefinition::EVENT_COUNT:
      case MetricDefinition::ELAPSED_TIME:
      case MetricDefinition::FRAME_RATE:
      case MetricDefinition::MEMORY_USAGE: {
        switch (report.report_type()) {
          case ReportDefinition::PER_DEVICE_NUMERIC_STATS: {
            auto status = GeneratePerDeviceNumericObservations(
                metric_ref, pair.first, pair.second, final_day_index);
            if (status != kOK) {
              return status;
            }
            break;
          }
          default:
            continue;
        }
        break;
      }
      default:
        continue;
    }
  }
  return kOK;
}

////////// GenerateUniqueActivesObservations and helper methods ////////////////

namespace {

// Given the set of daily aggregates for a fixed event code, and the size and
// end date of an aggregation window, returns the first day index within that
// window on which the event code occurred. Returns 0 if the event code did
// not occur within the window.
uint32_t FirstActiveDayIndexInWindow(const DailyAggregates& daily_aggregates,
                                     uint32_t obs_day_index,
                                     uint32_t window_size) {
  for (uint32_t day_index = obs_day_index - window_size + 1;
       day_index <= obs_day_index; day_index++) {
    auto day_aggregate = daily_aggregates.by_day_index().find(day_index);
    if (day_aggregate != daily_aggregates.by_day_index().end() &&
        day_aggregate->second.activity_daily_aggregate().activity_indicator() ==
            true) {
      return day_index;
    }
  }
  return 0u;
}

// Given the day index of an event occurrence and the size and end date
// of an aggregation window, returns true if the occurrence falls within
// the window and false if not.
bool IsActivityInWindow(uint32_t active_day_index, uint32_t obs_day_index,
                        uint32_t window_size) {
  return (active_day_index <= obs_day_index &&
          active_day_index > obs_day_index - window_size);
}

}  // namespace

uint32_t EventAggregator::UniqueActivesLastGeneratedDayIndex(
    const std::string& report_key, uint32_t event_code,
    uint32_t window_size) const {
  auto report_history = obs_history_.by_report_key().find(report_key);
  if (report_history == obs_history_.by_report_key().end()) {
    return 0u;
  }
  auto event_code_history =
      report_history->second.unique_actives_history().by_event_code().find(
          event_code);
  if (event_code_history ==
      report_history->second.unique_actives_history().by_event_code().end()) {
    return 0u;
  }
  auto window_size_history =
      event_code_history->second.by_window_size().find(window_size);
  if (window_size_history ==
      event_code_history->second.by_window_size().end()) {
    return 0u;
  }
  return window_size_history->second;
}

Status EventAggregator::GenerateSingleUniqueActivesObservation(
    const MetricRef metric_ref, const ReportDefinition* report,
    uint32_t obs_day_index, uint32_t event_code, uint32_t window_size,
    bool was_active) const {
  auto encoder_result = encoder_->EncodeUniqueActivesObservation(
      metric_ref, report, obs_day_index, event_code, was_active, window_size);
  if (encoder_result.status != kOK) {
    return encoder_result.status;
  }
  if (encoder_result.observation == nullptr ||
      encoder_result.metadata == nullptr) {
    LOG(ERROR) << "Failed to encode UniqueActivesObservation";
    return kOther;
  }

  auto writer_status = observation_writer_->WriteObservation(
      *encoder_result.observation, std::move(encoder_result.metadata));
  if (writer_status != kOK) {
    return writer_status;
  }
  return kOK;
}

Status EventAggregator::GenerateUniqueActivesObservations(
    const MetricRef metric_ref, const std::string& report_key,
    const ReportAggregates& report_aggregates, uint32_t num_event_codes,
    uint32_t final_day_index) {
  CHECK_GT(final_day_index, backfill_days_);
  // The earliest day index for which we might need to generate an
  // Observation.
  auto backfill_period_start = uint32_t(final_day_index - backfill_days_);

  for (uint32_t event_code = 0; event_code < num_event_codes; event_code++) {
    auto daily_aggregates =
        report_aggregates.unique_actives_aggregates().by_event_code().find(
            event_code);
    // Have any events ever been logged for this report and event code?
    bool found_event_code =
        (daily_aggregates !=
         report_aggregates.unique_actives_aggregates().by_event_code().end());
    for (uint32_t window_size :
         report_aggregates.aggregation_config().report().window_size()) {
      // Skip any window size larger than
      // kMaxAllowedAggregationWindowSize.
      if (window_size > kMaxAllowedAggregationWindowSize) {
        LOG(WARNING) << "GenerateUniqueActivesObservations ignoring a window "
                        "size exceeding the maximum allowed value";
        continue;
      }
      // Find the earliest day index for which an Observation has not yet
      // been generated for this report, event code, and window size. If
      // that day index is later than |final_day_index|, no Observation is
      // generated on this invocation.
      auto last_gen = UniqueActivesLastGeneratedDayIndex(report_key, event_code,
                                                         window_size);
      auto first_day_index = std::max(last_gen + 1, backfill_period_start);
      // The latest day index on which |event_type| is known to have
      // occurred, so far. This value will be updated as we search
      // forward from the earliest day index belonging to a window of
      // interest.
      uint32_t active_day_index = 0u;
      // Iterate over the day indices |obs_day_index| for which we need
      // to generate Observations. On each iteration, generate an
      // Observation for the window of size |window_size| ending on
      // |obs_day_index|.
      for (uint32_t obs_day_index = first_day_index;
           obs_day_index <= final_day_index; obs_day_index++) {
        bool was_active = false;
        if (found_event_code) {
          // If the current value of |active_day_index| falls within the
          // window, generate an Observation of activity. If not, search
          // forward in the window, update |active_day_index|, and generate an
          // Observation of activity or inactivity depending on the result of
          // the search.
          if (IsActivityInWindow(active_day_index, obs_day_index,
                                 window_size)) {
            was_active = true;
          } else {
            active_day_index = FirstActiveDayIndexInWindow(
                daily_aggregates->second, obs_day_index, window_size);
            was_active = IsActivityInWindow(active_day_index, obs_day_index,
                                            window_size);
          }
        }
        auto status = GenerateSingleUniqueActivesObservation(
            metric_ref, &report_aggregates.aggregation_config().report(),
            obs_day_index, event_code, window_size, was_active);
        if (status != kOK) {
          return status;
        }
        // Update |obs_history_| with the latest date of Observation
        // generation for this report, event code, and window size.
        (*(*(*obs_history_.mutable_by_report_key())[report_key]
                .mutable_unique_actives_history()
                ->mutable_by_event_code())[event_code]
              .mutable_by_window_size())[window_size] = obs_day_index;
      }
    }
  }
  return kOK;
}

////////// GeneratePerDeviceNumericObservations and helper methods /////////////

uint32_t EventAggregator::PerDeviceNumericLastGeneratedDayIndex(
    const std::string& report_key, const std::string& component,
    uint32_t event_code, uint32_t window_size) const {
  const auto& report_history = obs_history_.by_report_key().find(report_key);
  if (report_history == obs_history_.by_report_key().end()) {
    return 0u;
  }
  if (!report_history->second.has_per_device_numeric_history()) {
    return 0u;
  }
  const auto& component_history =
      report_history->second.per_device_numeric_history().by_component().find(
          component);
  if (component_history == report_history->second.per_device_numeric_history()
                               .by_component()
                               .end()) {
    return 0u;
  }
  const auto& event_code_history =
      component_history->second.by_event_code().find(event_code);
  if (event_code_history == component_history->second.by_event_code().end()) {
    return 0u;
  }
  const auto& window_size_history =
      event_code_history->second.by_window_size().find(window_size);
  if (window_size_history ==
      event_code_history->second.by_window_size().end()) {
    return 0u;
  }
  return window_size_history->second;
}

uint32_t EventAggregator::ReportParticipationLastGeneratedDayIndex(
    const std::string& report_key) const {
  const auto& report_history = obs_history_.by_report_key().find(report_key);
  if (report_history == obs_history_.by_report_key().end()) {
    return 0u;
  }
  return report_history->second.report_participation_history().last_generated();
}

Status EventAggregator::GenerateSinglePerDeviceNumericObservation(
    const MetricRef metric_ref, const ReportDefinition* report,
    uint32_t obs_day_index, const std::string& component, uint32_t event_code,
    uint32_t window_size, int64_t value) const {
  auto encoder_result = encoder_->EncodePerDeviceNumericObservation(
      metric_ref, report, obs_day_index, component,
      UnpackEventCodesProto(event_code), value, window_size);
  if (encoder_result.status != kOK) {
    return encoder_result.status;
  }
  if (encoder_result.observation == nullptr ||
      encoder_result.metadata == nullptr) {
    LOG(ERROR) << "Failed to encode PerDeviceNumericObservation";
    return kOther;
  }

  const auto& writer_status = observation_writer_->WriteObservation(
      *encoder_result.observation, std::move(encoder_result.metadata));
  if (writer_status != kOK) {
    return writer_status;
  }
  return kOK;
}

Status EventAggregator::GenerateSingleReportParticipationObservation(
    const MetricRef metric_ref, const ReportDefinition* report,
    uint32_t obs_day_index) const {
  auto encoder_result = encoder_->EncodeReportParticipationObservation(
      metric_ref, report, obs_day_index);
  if (encoder_result.status != kOK) {
    return encoder_result.status;
  }
  if (encoder_result.observation == nullptr ||
      encoder_result.metadata == nullptr) {
    LOG(ERROR) << "Failed to encode ReportParticipationObservation";
    return kOther;
  }

  const auto& writer_status = observation_writer_->WriteObservation(
      *encoder_result.observation, std::move(encoder_result.metadata));
  if (writer_status != kOK) {
    return writer_status;
  }
  return kOK;
}

Status EventAggregator::GeneratePerDeviceNumericObservations(
    const MetricRef metric_ref, const std::string& report_key,
    const ReportAggregates& report_aggregates, uint32_t final_day_index) {
  CHECK_GT(final_day_index, backfill_days_);
  // The first day index for which we might have to generate an Observation.
  auto backfill_period_start = uint32_t(final_day_index - backfill_days_);

  // Generate any necessary PerDeviceNumericObservations for this report.
  for (const auto& component_pair :
       report_aggregates.numeric_aggregates().by_component()) {
    const auto& component = component_pair.first;
    for (const auto& event_code_pair : component_pair.second.by_event_code()) {
      auto event_code = event_code_pair.first;
      const auto& event_code_aggregates = event_code_pair.second;
      // Populate a helper map keyed by day indices which belong to the range
      // [|backfill_period_start|, |final_day_index|]. The value at a day
      // index is the list of window sizes, in increasing order, for which an
      // Observation should be generated for that day index.
      std::map<uint32_t, std::vector<uint32_t>> window_sizes_by_obs_day;
      for (auto window_size :
           report_aggregates.aggregation_config().window_size()) {
        auto last_gen = PerDeviceNumericLastGeneratedDayIndex(
            report_key, component, event_code, window_size);
        auto first_day_index = std::max(last_gen + 1, backfill_period_start);
        for (auto obs_day_index = first_day_index;
             obs_day_index <= final_day_index; obs_day_index++) {
          window_sizes_by_obs_day[obs_day_index].push_back(window_size);
        }
      }
      // Iterate over the day indices |obs_day_index| for which we might need
      // to generate an Observation. For each day index, generate an
      // Observation for each needed window size.
      //
      // More precisely, for each needed window size, iterate over the days
      // within that window. If at least one numeric event was logged during the
      // window, compute the aggregate of the numeric values over the window and
      // generate a PerDeviceNumericObservation. Whether or not a numeric event
      // was found, update the AggregatedObservationHistory for this report,
      // component, event code, and window size with |obs_day_index| as the most
      // recent date of Observation generation. This reflects the fact that all
      // needed Observations were generated for the window ending on that date.
      for (auto obs_day_index = backfill_period_start;
           obs_day_index <= final_day_index; obs_day_index++) {
        const auto& window_sizes = window_sizes_by_obs_day.find(obs_day_index);
        if (window_sizes == window_sizes_by_obs_day.end()) {
          continue;
        }
        bool found_value_for_window = false;
        int64_t window_aggregate = 0;
        uint32_t num_days = 0;
        for (auto window_size : window_sizes->second) {
          while (num_days < window_size) {
            bool found_value_for_day = false;
            const auto& day_aggregates =
                event_code_aggregates.by_day_index().find(obs_day_index -
                                                          num_days);
            if (day_aggregates != event_code_aggregates.by_day_index().end()) {
              found_value_for_day = true;
            }
            const auto& aggregation_type =
                report_aggregates.aggregation_config()
                    .report()
                    .aggregation_type();
            switch (aggregation_type) {
              case ReportDefinition::SUM:
                if (found_value_for_day) {
                  window_aggregate +=
                      day_aggregates->second.numeric_daily_aggregate().value();
                  found_value_for_window = true;
                }
                break;
              case ReportDefinition::MAX:
                if (found_value_for_day) {
                  window_aggregate = std::max(
                      window_aggregate,
                      day_aggregates->second.numeric_daily_aggregate().value());
                  found_value_for_window = true;
                }
                break;
              case ReportDefinition::MIN:
                if (found_value_for_day && !found_value_for_window) {
                  window_aggregate =
                      day_aggregates->second.numeric_daily_aggregate().value();
                  found_value_for_window = true;
                } else if (found_value_for_day) {
                  window_aggregate = std::min(
                      window_aggregate,
                      day_aggregates->second.numeric_daily_aggregate().value());
                }
                break;
              default:
                LOG(ERROR) << "Unexpected aggregation type "
                           << aggregation_type;
                return kInvalidArguments;
            }
            num_days++;
          }
          if (found_value_for_window) {
            auto status = GenerateSinglePerDeviceNumericObservation(
                metric_ref, &report_aggregates.aggregation_config().report(),
                obs_day_index, component, event_code, window_size,
                window_aggregate);
            if (status != kOK) {
              return status;
            }
          }
          // Update |obs_history_| with the latest date of Observation
          // generation for this report, component, event code, and window
          // size.
          (*(*(*(*obs_history_.mutable_by_report_key())[report_key]
                    .mutable_per_device_numeric_history()
                    ->mutable_by_component())[component]
                  .mutable_by_event_code())[event_code]
                .mutable_by_window_size())[window_size] = obs_day_index;
        }
      }
    }
  }
  // Generate any necessary ReportParticipationObservations for this report.
  auto participation_last_gen =
      ReportParticipationLastGeneratedDayIndex(report_key);
  auto participation_first_day_index =
      std::max(participation_last_gen + 1, backfill_period_start);
  for (auto obs_day_index = participation_first_day_index;
       obs_day_index <= final_day_index; obs_day_index++) {
    GenerateSingleReportParticipationObservation(
        metric_ref, &report_aggregates.aggregation_config().report(),
        obs_day_index);
    (*obs_history_.mutable_by_report_key())[report_key]
        .mutable_report_participation_history()
        ->set_last_generated(obs_day_index);
  }
  return kOK;
}

}  // namespace logger
}  // namespace cobalt