base::ListValue reports_value;

for (const ReportingReport* report : reports) {

std::unique_ptr<base::DictionaryValue> report_value =

std::make_unique<base::DictionaryValue>();

report_value->SetInteger("age", (now - report->queued).InMilliseconds());

report_value->SetString("type", report->type);

report_value->SetString("url", report->url.spec());

report_value->SetString("user_agent", report->user_agent);

report_value->SetKey("body", report->body->Clone());

reports_value.Append(std::move(report_value));

}

std::string json_out;

bool json_written = base::JSONWriter::Write(reports_value, &json_out);

DCHECK(json_written);

return json_out;

public:

// The target of a delivery. All reports uploaded together must share the

// same values for these parameters.

// Note that |origin| here (which matches the report's |origin|) is not

// necessarily the same as the |origin| of the ReportingEndpoint's group key

// (if the endpoint is configured to include subdomains). Reports with

// different group keys can be in the same delivery, as long as the NIK,

// report origin and reporting source are the same, and they all get assigned

// to the same endpoint URL.

// |isolation_info| is the IsolationInfo struct associated with the reporting

// endpoint, and is used to determine appropriate credentials for the upload.

// |network_isolation_key| is the NIK from the ReportingEndpoint, which may

// have been cleared in the ReportingService if reports are not being

// partitioned by NIK. (This is why a separate parameter is used here, rather

// than simply using the computed NIK from |isolation_info|.)

struct Target {

Target(const IsolationInfo& isolation_info,

const NetworkIsolationKey& network_isolation_key,

const url::Origin& origin,

const GURL& endpoint_url,

const absl::optional<base::UnguessableToken> reporting_source)

: isolation_info(isolation_info),

network_isolation_key(network_isolation_key),

origin(origin),

endpoint_url(endpoint_url),

reporting_source(reporting_source) {

DCHECK(network_isolation_key.IsEmpty() ||

network_isolation_key == isolation_info.network_isolation_key());

}

~Target() = default;

bool operator<(const Target& other) const {

// Note that sorting by NIK here is required for V0 reports; V1 reports

// should not need this (but it doesn't hurt). We can remove that as a

// comparison key when V0 reporting endpoints are removed.

return std::tie(network_isolation_key, origin, endpoint_url,

reporting_source) <

std::tie(other.network_isolation_key, other.origin,

other.endpoint_url, other.reporting_source);

}

IsolationInfo isolation_info;

NetworkIsolationKey network_isolation_key;

url::Origin origin;

GURL endpoint_url;

absl::optional<base::UnguessableToken> reporting_source;

};

explicit Delivery(const Target& target) : target_(target) {}

~Delivery() = default;

// Add the reports in [reports_begin, reports_end) into this delivery.

// Modify the report counter for the |endpoint| to which this delivery is

// destined.

void AddReports(const ReportingEndpoint& endpoint,

const ReportList::const_iterator reports_begin,

const ReportList::const_iterator reports_end) {

DCHECK(reports_begin != reports_end);

DCHECK_EQ(endpoint.group_key.network_isolation_key,

network_isolation_key());

DCHECK(IsSubdomainOf(target_.origin.host() /* subdomain */,

endpoint.group_key.origin.host() /* superdomain */));

for (auto it = reports_begin; it != reports_end; ++it) {

DCHECK_EQ((*reports_begin)->GetGroupKey(), (*it)->GetGroupKey());

DCHECK_EQ((*it)->network_isolation_key, network_isolation_key());

DCHECK_EQ(url::Origin::Create((*it)->url), target_.origin);

DCHECK_EQ((*it)->group, endpoint.group_key.group_name);

// Report origin is equal to, or a subdomain of, the endpoint

// configuration's origin.

DCHECK(IsSubdomainOf((*it)->url.host_piece() /* subdomain */,

endpoint.group_key.origin.host() /* superdomain */));

}

reports_per_group_[endpoint.group_key] +=

std::distance(reports_begin, reports_end);

reports_.insert(reports_.end(), reports_begin, reports_end);

}

// Records statistics for reports after an upload has completed.

// Either removes successfully delivered reports, or increments the failure

// counter if delivery was unsuccessful.

void ProcessOutcome(ReportingCache* cache, bool success) {

for (const auto& group_name_and_count : reports_per_group_) {

cache->IncrementEndpointDeliveries(group_name_and_count.first,

target_.endpoint_url,

group_name_and_count.second, success);

}

if (success) {

ReportingUploadHeaderType upload_type =

target_.reporting_source.has_value()

? ReportingUploadHeaderType::kReportingEndpoints

: ReportingUploadHeaderType::kReportTo;

for (size_t i = 0; i < reports_.size(); ++i) {

RecordReportingUploadHeaderType(upload_type);

}

cache->RemoveReports(reports_, /* delivery_success */ true);

} else {

cache->IncrementReportsAttempts(reports_);

}

}

const NetworkIsolationKey& network_isolation_key() const {

return target_.network_isolation_key;

}

const GURL& endpoint_url() const { return target_.endpoint_url; }

const ReportList& reports() const { return reports_; }

private:

const Target target_;

ReportList reports_;

// Used to track statistics for each ReportingEndpoint.

// The endpoint is uniquely identified by the key in conjunction with

// |target_.endpoint_url|. See ProcessOutcome().

std::map<ReportingEndpointGroupKey, int> reports_per_group_;

public ReportingCacheObserver {

public:

ReportingDeliveryAgentImpl(ReportingContext* context,

const RandIntCallback& rand_callback)

: context_(context),

timer_(std::make_unique<base::OneShotTimer>()),

endpoint_manager_(

ReportingEndpointManager::Create(&context->policy(),

&context->tick_clock(),

context->delegate(),

context->cache(),

rand_callback)) {

context_->AddCacheObserver(this);

}

ReportingDeliveryAgentImpl(const ReportingDeliveryAgentImpl&) = delete;

ReportingDeliveryAgentImpl& operator=(const ReportingDeliveryAgentImpl&) =

delete;

// ReportingDeliveryAgent implementation:

~ReportingDeliveryAgentImpl() override {

context_->RemoveCacheObserver(this);

}

void SetTimerForTesting(std::unique_ptr<base::OneShotTimer> timer) override {

DCHECK(!timer_->IsRunning());

timer_ = std::move(timer);

}

void SendReportsForSource(base::UnguessableToken reporting_source) override {

DCHECK(!reporting_source.is_empty());

ReportList reports =

cache()->GetReportsToDeliverForSource(reporting_source);

if (reports.empty())

return;

DoSendReports(std::move(reports));

}

// ReportingCacheObserver implementation:

void OnReportsUpdated() override {

if (CacheHasReports() && !timer_->IsRunning()) {

SendReports();

StartTimer();

}

}

private:

bool CacheHasReports() {

ReportList reports;

context_->cache()->GetReports(&reports);

return !reports.empty();

}

void StartTimer() {

timer_->Start(FROM_HERE, policy().delivery_interval,

base::BindOnce(&ReportingDeliveryAgentImpl::OnTimerFired,

base::Unretained(this)));

}

void OnTimerFired() {

if (CacheHasReports()) {

SendReports();

StartTimer();

}

}

void SendReports() {

ReportList reports = cache()->GetReportsToDeliver();

if (reports.empty())

return;

DoSendReports(std::move(reports));

}

void DoSendReports(ReportList reports) {

// First determine which origins we're allowed to upload reports about.

std::set<url::Origin> report_origins;

for (const ReportingReport* report : reports) {

report_origins.insert(url::Origin::Create(report->url));

}

delegate()->CanSendReports(

std::move(report_origins),

base::BindOnce(&ReportingDeliveryAgentImpl::OnSendPermissionsChecked,

weak_factory_.GetWeakPtr(), std::move(reports)));

}

void OnSendPermissionsChecked(ReportList reports,

std::set<url::Origin> allowed_report_origins) {

DCHECK(!reports.empty());

std::map<Delivery::Target, std::unique_ptr<Delivery>> deliveries;

// Sort by group key

std::sort(reports.begin(), reports.end(), &CompareReportGroupKeys);

// Iterate over "buckets" of reports with the same group key.

for (auto bucket_it = reports.begin(); bucket_it != reports.end();) {

auto bucket_start = bucket_it;

// Set the iterator to the beginning of the next group bucket.

bucket_it = std::upper_bound(bucket_it, reports.end(), *bucket_it,

&CompareReportGroupKeys);

// Skip this group if we don't have origin permissions for this origin.

const ReportingEndpointGroupKey& report_group_key =

(*bucket_start)->GetGroupKey();

if (!base::Contains(allowed_report_origins, report_group_key.origin))

continue;

// Skip this group if there is already a pending upload for it.

// We don't allow multiple concurrent uploads for the same group.

if (base::Contains(pending_groups_, report_group_key))

continue;

// Find an endpoint to deliver these reports to.

const ReportingEndpoint endpoint =

endpoint_manager_->FindEndpointForDelivery(report_group_key);

// TODO(chlily): Remove reports for which there are no valid delivery

// endpoints.

if (!endpoint)

continue;

pending_groups_.insert(report_group_key);

IsolationInfo isolation_info =

cache()->GetIsolationInfoForEndpoint(endpoint);

// Add the reports to the appropriate delivery.

Delivery::Target target(isolation_info,

report_group_key.network_isolation_key,

report_group_key.origin, endpoint.info.url,

endpoint.group_key.reporting_source);

auto delivery_it = deliveries.find(target);

if (delivery_it == deliveries.end()) {

bool inserted;

auto new_delivery = std::make_unique<Delivery>(target);

std::tie(delivery_it, inserted) = deliveries.insert(

std::make_pair(std::move(target), std::move(new_delivery)));

DCHECK(inserted);

}

delivery_it->second->AddReports(endpoint, bucket_start, bucket_it);

}

// Keep track of which of these reports we don't queue for delivery; we'll

// need to mark them as not-pending.

std::set<const ReportingReport*> undelivered_reports(reports.begin(),

reports.end());

// Start an upload for each delivery.

for (auto& target_and_delivery : deliveries) {

const Delivery::Target& target = target_and_delivery.first;

std::unique_ptr<Delivery>& delivery = target_and_delivery.second;

int max_depth = 0;

for (const ReportingReport* report : delivery->reports()) {

undelivered_reports.erase(report);

max_depth = std::max(report->depth, max_depth);

}

std::string upload_data =

SerializeReports(delivery->reports(), tick_clock().NowTicks());

// TODO: Calculate actual max depth.

uploader()->StartUpload(

target.origin, target.endpoint_url, target.isolation_info,

upload_data, max_depth,

/*eligible_for_credentials=*/target.reporting_source.has_value(),

base::BindOnce(&ReportingDeliveryAgentImpl::OnUploadComplete,

weak_factory_.GetWeakPtr(), std::move(delivery)));

}

cache()->ClearReportsPending(

{undelivered_reports.begin(), undelivered_reports.end()});

}

void OnUploadComplete(std::unique_ptr<Delivery> delivery,

ReportingUploader::Outcome outcome) {

bool success = outcome == ReportingUploader::Outcome::SUCCESS;

delivery->ProcessOutcome(cache(), success);

endpoint_manager_->InformOfEndpointRequest(

delivery->network_isolation_key(), delivery->endpoint_url(), success);

// TODO(chlily): This leaks information across NIKs. If the endpoint URL is

// configured for both NIK1 and NIK2, and it responds with a 410 on a NIK1

// connection, then the change in configuration will be detectable on a NIK2

// connection.

// TODO(rodneyding): Handle Remove endpoint for Reporting-Endpoints header.

if (outcome == ReportingUploader::Outcome::REMOVE_ENDPOINT)

cache()->RemoveEndpointsForUrl(delivery->endpoint_url());

for (const ReportingReport* report : delivery->reports()) {

pending_groups_.erase(report->GetGroupKey());

}

cache()->ClearReportsPending(delivery->reports());

}

const ReportingPolicy& policy() const { return context_->policy(); }

const base::TickClock& tick_clock() const { return context_->tick_clock(); }

ReportingDelegate* delegate() { return context_->delegate(); }

ReportingCache* cache() { return context_->cache(); }

ReportingUploader* uploader() { return context_->uploader(); }

raw_ptr<ReportingContext> context_;

std::unique_ptr<base::OneShotTimer> timer_;

// Tracks endpoint groups for which there is a pending delivery running.

std::set<ReportingEndpointGroupKey> pending_groups_;

std::unique_ptr<ReportingEndpointManager> endpoint_manager_;

base::WeakPtrFactory<ReportingDeliveryAgentImpl> weak_factory_{this};

ReportingContext* context,

const RandIntCallback& rand_callback) {

return std::make_unique<ReportingDeliveryAgentImpl>(context, rand_callback);