compute and upload 30 day total download counts

Author: szakarias

app/lib/service/download_counts/backend.dart

@@ -2,11 +2,18 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:math';
+
+import 'package:basics/basics.dart';
 import 'package:gcloud/service_scope.dart' as ss;
+import 'package:gcloud/storage.dart';
 import 'package:pub_dev/service/download_counts/download_counts.dart';
 import 'package:pub_dev/service/download_counts/models.dart';
+import 'package:pub_dev/shared/configuration.dart';
 import 'package:pub_dev/shared/datastore.dart';
 import 'package:pub_dev/shared/redis_cache.dart';
+import 'package:pub_dev/shared/storage.dart';
+import 'package:pub_dev/shared/utils.dart';
 
 /// Sets the download counts backend service.
 void registerDownloadCountsBackend(DownloadCountsBackend backend) =>
@@ -29,6 +36,11 @@ class DownloadCountsBackend {
     }));
   }
 
+  Future<Stream<DownloadCounts>> listAllDownloadCounts() async {
+    final query = _db.query<DownloadCounts>();
+    return query.run();
+  }
+
   Future<DownloadCounts> updateDownloadCounts(
     String pkg,
     Map<String, int> dayCounts,

app/lib/service/download_counts/compute_30_days_total_counts.dart

@@ -0,0 +1,44 @@
+// Copyright (c) 2024, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+import 'dart:async';
+import 'dart:math';
+
+import 'package:gcloud/storage.dart';
+import 'package:pub_dev/service/download_counts/backend.dart';
+import 'package:pub_dev/service/download_counts/models.dart';
+import 'package:pub_dev/shared/configuration.dart';
+import 'package:pub_dev/shared/storage.dart';
+import 'package:pub_dev/shared/utils.dart';
+
+Future<void> compute30DaysTotalTask() async {
+  final allDownloadCounts = await downloadCountsBackend.listAllDownloadCounts();
+  final totals = await compute30DayTotals(allDownloadCounts);
+  await updload30DaysTotal(totals);
+}
+
+Future<Map<String, int>> compute30DayTotals(
+    Stream<DownloadCounts> downloadCounts) async {
+  final res = <String, int>{};
+  await for (final dc in downloadCounts) {
+    res[dc.package] = compute30DayTotal(dc);
+  }
+
+  return res;
+}
+
+int compute30DayTotal(DownloadCounts downloadCounts) {
+  final totals = downloadCounts.countData.totalCounts;
+  return totals
+      .take(30)
+      .fold(0, (previousValue, element) => previousValue + max(0, element));
+}
+
+final downloadCounts30DaysTotalsFileName = 'download-counts-30-days-total.json';
+
+Future<void> updload30DaysTotal(Map<String, int> counts) async {
+  final reportsBucket =
+      storageService.bucket(activeConfiguration.reportsBucketName!);
+  await uploadBytesWithRetry(reportsBucket, downloadCounts30DaysTotalsFileName,
+      jsonUtf8Encoder.convert(counts));
+}

app/lib/tool/neat_task/pub_dev_tasks.dart

@@ -8,6 +8,7 @@ import 'dart:io';
 import 'package:gcloud/service_scope.dart' as ss;
 import 'package:logging/logging.dart';
 import 'package:neat_periodic_task/neat_periodic_task.dart';
+import 'package:pub_dev/service/download_counts/compute_30_days_total_counts.dart';
 
 import '../../account/backend.dart';
 import '../../account/consent_backend.dart';
@@ -187,6 +188,11 @@ void _setupGenericPeriodicTasks() {
       isRuntimeVersioned: false,
       task: syncDownloadCounts);
 
+  _daily(
+      name: 'compute-download-counts-30-days-totals',
+      isRuntimeVersioned: false,
+      task: compute30DaysTotalTask);
+
   _daily(name: 'count-topics', isRuntimeVersioned: false, task: countTopics);
 
   _daily(

app/test/service/download_counts/compute_total_download_counts_test.dart

@@ -0,0 +1,102 @@
+// Copyright (c) 2024, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+import 'dart:convert';
+
+import 'package:basics/basics.dart';
+import 'package:gcloud/storage.dart';
+import 'package:pub_dev/service/download_counts/backend.dart';
+import 'package:pub_dev/service/download_counts/compute_30_days_total_counts.dart';
+import 'package:pub_dev/shared/configuration.dart';
+import 'package:test/test.dart';
+
+import '../../shared/test_services.dart';
+
+void main() {
+  group('', () {
+    testWithProfile('compute download counts 30 day totals', fn: () async {
+      final pkg = 'foo';
+      final versionsCounts = {
+        '1.0.1': 2,
+        '2.0.0-alpha': 2,
+        '2.0.0': 2,
+        '2.1.0': 2,
+        '3.1.0': 2,
+        '4.0.0-0': 2,
+        '6.1.0': 2,
+      };
+      final date = DateTime.parse('1986-02-16');
+      var downloadCounts1 = await downloadCountsBackend.updateDownloadCounts(
+          pkg, versionsCounts, date);
+      for (var i = 1; i < 5; i++) {
+        downloadCounts1 = await downloadCountsBackend.updateDownloadCounts(
+            pkg, versionsCounts, date.addCalendarDays(i));
+      }
+
+      expect(compute30DayTotal(downloadCounts1), 70);
+
+      final pkg2 = 'bar';
+      final versionsCounts2 = {
+        '1.0.1': 3,
+        '2.0.0-alpha': 3,
+        '2.0.0': 3,
+        '2.1.0': 3,
+        '3.1.0': 3,
+        '4.0.0-0': 3,
+        '6.1.0': 3,
+      };
+      var downloadCounts2 = await downloadCountsBackend.updateDownloadCounts(
+          pkg2, versionsCounts2, date);
+      for (var i = 1; i < 5; i++) {
+        downloadCounts2 = await downloadCountsBackend.updateDownloadCounts(
+            pkg2, versionsCounts2, date.addCalendarDays(i));
+      }
+
+      expect(compute30DayTotal(downloadCounts2), 105);
+
+      final pkg3 = 'baz';
+      final versionsCounts3 = {
+        '1.0.1': 4,
+        '2.0.0-alpha': 4,
+        '2.0.0': 4,
+        '2.1.0': 4,
+        '3.1.0': 4,
+        '4.0.0-0': 4,
+        '6.1.0': 4,
+      };
+      var downloadCounts3 = await downloadCountsBackend.updateDownloadCounts(
+          pkg3, versionsCounts3, date);
+      for (var i = 1; i < 5; i++) {
+        downloadCounts3 = await downloadCountsBackend.updateDownloadCounts(
+            pkg3, versionsCounts3, date.addCalendarDays(i));
+      }
+      expect(compute30DayTotal(downloadCounts3), 140);
+
+      final downloadCounts = [
+        downloadCounts1,
+        downloadCounts2,
+        downloadCounts3
+      ];
+
+      final res = await compute30DayTotals(Stream.fromIterable(downloadCounts));
+
+      expect(
+        res,
+        {'foo': 70, 'bar': 105, 'baz': 140},
+      );
+    });
+
+    testWithProfile('succesful 30 day totals upload', fn: () async {
+      await updload30DaysTotal({'foo': 70, 'bar': 105, 'baz': 140});
+
+      final data = await storageService
+          .bucket(activeConfiguration.reportsBucketName!)
+          .read(downloadCounts30DaysTotalsFileName)
+          .transform(utf8.decoder)
+          .transform(json.decoder)
+          .single;
+
+      expect(data, {'foo': 70, 'bar': 105, 'baz': 140});
+    });
+  });
+}