content/browser/conversions/conversion_reporter_impl.cc

// Copyright 2020 The Chromium 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 "content/browser/conversions/conversion_reporter_impl.h"

#include "base/bind.h"
#include "base/callback.h"
#include "base/rand_util.h"
#include "base/time/clock.h"
#include "content/browser/conversions/conversion_manager.h"
#include "content/browser/conversions/conversion_network_sender_impl.h"

namespace content {

ConversionReporterImpl::ConversionReporterImpl(
    StoragePartition* storage_partition,
    ConversionManager* conversion_manager,
    const base::Clock* clock)
    : conversion_manager_(conversion_manager),
      clock_(clock),
      network_sender_(
          std::make_unique<ConversionNetworkSenderImpl>(storage_partition)) {
  DCHECK(conversion_manager_);
}

ConversionReporterImpl::~ConversionReporterImpl() = default;

void ConversionReporterImpl::AddReportsToQueue(
    std::vector<ConversionReport> reports) {
  DCHECK(!reports.empty());

  std::vector<std::unique_ptr<ConversionReport>> swappable_reports;
  for (ConversionReport& report : reports) {
    swappable_reports.push_back(
        std::make_unique<ConversionReport>(std::move(report)));
  }

  // Shuffle new reports to provide plausible deniability on the ordering of
  // reports that share the same |report_time|. This is important because
  // multiple conversions for the same impression share the same report time if
  // they are within the same reporting window, and we do not want to allow
  // ordering on their conversion metadata bits.
  base::RandomShuffle(swappable_reports.begin(), swappable_reports.end());

  for (std::unique_ptr<ConversionReport>& report : swappable_reports) {
    // If the given report is already being processed, ignore it.
    bool inserted =
        conversion_ids_being_processed_.insert(*(report->conversion_id)).second;
    if (inserted)
      report_queue_.push(std::move(report));
  }
  MaybeScheduleNextReport();
}

void ConversionReporterImpl::SetNetworkSenderForTesting(
    std::unique_ptr<NetworkSender> network_sender) {
  network_sender_ = std::move(network_sender);
}

void ConversionReporterImpl::SendNextReport() {
  // Send the next report and remove it from the queue. Bind the conversion id
  // to the sent callback so we know which conversion report has finished
  // sending.
  network_sender_->SendReport(
      report_queue_.top().get(),
      base::BindOnce(&ConversionReporterImpl::OnReportSent,
                     base::Unretained(this),
                     *report_queue_.top()->conversion_id));
  report_queue_.pop();
  MaybeScheduleNextReport();
}

void ConversionReporterImpl::MaybeScheduleNextReport() {
  if (report_queue_.empty())
    return;

  send_report_timer_.Stop();
  base::Time current_time = clock_->Now();
  base::Time report_time = report_queue_.top()->report_time;

  // Start a timer to wait until the next report is ready to be sent. This
  // purposefully yields the thread for every report that gets scheduled.
  // Unretained is safe because the task should never actually be posted if the
  // timer itself is destroyed
  send_report_timer_.Start(
      FROM_HERE, report_time - current_time,
      base::BindOnce(&ConversionReporterImpl::SendNextReport,
                     base::Unretained(this)));
}

void ConversionReporterImpl::OnReportSent(int64_t conversion_id) {
  conversion_ids_being_processed_.erase(conversion_id);
  conversion_manager_->HandleSentReport(conversion_id);
}

bool ConversionReporterImpl::ReportComparator::operator()(
    const std::unique_ptr<ConversionReport>& a,
    const std::unique_ptr<ConversionReport>& b) const {
  // Returns whether a should appear before b in ordering. Because
  // std::priority_queue is max priority queue, we used greater then to make a
  // min priority queue.
  return a->report_time > b->report_time;
}

}  // namespace content