Tool for making a report of unique uploaders per month (#6289)

Author: sigurdm

app/config/dartlang-pub-dev.yaml

@@ -42,6 +42,16 @@ admins:
   - manageRetraction
   - removePackage
   - removeUsers
+- oauthUserId: '109830288482976007810'
+  email: [email protected]
+  permissions:
+  - executeTool
+  - listUsers
+  - manageAssignedTags
+  - managePackageOwnership
+  - manageRetraction
+  - removePackage
+  - removeUsers
 - oauthUserId: '117672289743137340098'
   email: [email protected]
   permissions:

app/lib/account/backend.dart

@@ -13,6 +13,7 @@ import 'package:meta/meta.dart';
 // ignore: import_of_legacy_library_into_null_safe
 import 'package:neat_cache/neat_cache.dart';
 
+import '../audit/models.dart';
 import '../service/openid/gcp_openid.dart';
 import '../service/openid/github_openid.dart';
 import '../shared/configuration.dart';
@@ -571,6 +572,20 @@ class AccountBackend {
     }
   }
 
+  /// Retrieves a list of all uploader events that happened between [begin] and
+  /// [end].
+  Stream<AuditLogRecord> getUploadEvents({DateTime? begin, DateTime? end}) {
+    final query = _db.query<AuditLogRecord>();
+    query.filter('kind =', AuditLogRecordKind.packagePublished);
+    if (begin != null) {
+      query.filter('created >=', begin);
+    }
+    if (end != null) {
+      query.filter('created <', end);
+    }
+    return query.run();
+  }
+
   // expire all sessions of a given user from datastore and cache
   Future<void> _expireAllSessions(String userId) async {
     final query = _db.query<UserSession>()..filter('userId =', userId);

app/lib/admin/backend.dart

@@ -43,6 +43,7 @@ import 'tools/recent_uploaders.dart';
 import 'tools/set_package_blocked.dart';
 import 'tools/set_secret.dart';
 import 'tools/set_user_blocked.dart';
+import 'tools/uploader_count_report.dart';
 import 'tools/user_merger.dart';
 
 final _logger = Logger('pub.admin.backend');
@@ -72,6 +73,7 @@ final Map<String, Tool> availableTools = {
   'set-user-blocked': executeSetUserBlocked,
   'user-merger': executeUserMergerTool,
   'list-tools': executeListTools,
+  'uploader-count-report': executeCountUploaderReport,
 };
 
 /// Represents the backend for the admin handling and authentication.

app/lib/admin/tools/uploader_count_report.dart

@@ -0,0 +1,52 @@
+// Copyright (c) 2022, 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 'package:clock/clock.dart';
+import 'package:pub_dev/account/backend.dart';
+
+const _monthNames = {
+  1: 'January',
+  2: 'February',
+  3: 'March',
+  4: 'April',
+  5: 'May',
+  6: 'June',
+  7: 'July',
+  8: 'August',
+  9: 'September',
+  10: 'October',
+  11: 'November',
+  12: 'December',
+};
+
+Future<String> executeCountUploaderReport(List<String> args) async {
+  final buckets = <DateTime, Set<String>>{};
+  final now = clock.now();
+
+  await for (final record in accountBackend.getUploadEvents(
+    begin: DateTime(now.year, now.month - 12),
+  )) {
+    final created = record.created;
+    final users = record.users;
+    if (created == null) continue;
+    if (users == null) continue;
+    final bucket =
+        buckets.putIfAbsent(DateTime(created.year, created.month), () => {});
+    for (String user in users) {
+      bucket.add(user);
+    }
+  }
+  final buffer = StringBuffer();
+  buffer.writeln('Monthly unique uploading users:');
+
+  for (int i = 11; i >= 0; i--) {
+    final month = DateTime(now.year, now.month - i);
+    final bucket = buckets[month] ?? {};
+    buffer
+        .writeln('${_monthNames[month.month]} ${month.year}: ${bucket.length}');
+  }
+  return buffer.toString();
+}

app/lib/service/services.dart

@@ -5,12 +5,14 @@
 import 'dart:async' show FutureOr, Zone;
 
 import 'package:appengine/appengine.dart';
+import 'package:clock/clock.dart';
 import 'package:fake_gcloud/mem_datastore.dart';
 import 'package:fake_gcloud/mem_storage.dart';
 import 'package:gcloud/service_scope.dart';
 import 'package:gcloud/storage.dart';
 import 'package:googleapis_auth/auth_io.dart' as auth;
 import 'package:logging/logging.dart';
+import 'package:shelf/shelf.dart' as shelf;
 import 'package:shelf/shelf_io.dart';
 
 import '../account/backend.dart';
@@ -188,8 +190,9 @@ Future<R> withFakeServices<R>({
       await topPackages.start();
       await youtubeBackend.start();
       if (frontendServer != null) {
-        final handler =
-            wrapHandler(_logger, createAppHandler(), sanitize: true);
+        final handler = wrapHandler(
+            _logger, _fakeClockWrapper(createAppHandler()),
+            sanitize: true);
         final fsSubscription = frontendServer.server.listen((rq) async {
           await fork(() => handleRequest(rq, handler));
         });
@@ -200,6 +203,21 @@ Future<R> withFakeServices<R>({
   }) as R;
 }
 
+const fakeClockHeaderName = '_fake_clock';
+
+/// In the fake server a request can send a '_fake_clock' header to specify at
+/// what timestamp the request should be handled.
+shelf.Handler _fakeClockWrapper(shelf.Handler handler) {
+  return (shelf.Request request) {
+    final t = request.headers[fakeClockHeaderName];
+    if (t == null) {
+      return handler(request);
+    } else {
+      return withClock(Clock.fixed(DateTime.parse(t)), () => handler(request));
+    }
+  };
+}
+
 /// Run [fn] with pub services that are shared between server instances, CLI
 /// tools and integration tests.
 Future<R> _withPubServices<R>(FutureOr<R> Function() fn) async {

app/lib/tool/utils/pub_api_client.dart

@@ -2,14 +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:convert';
+import 'dart:typed_data';
+
 import 'package:_pub_shared/data/package_api.dart';
+import 'package:clock/clock.dart';
 import 'package:gcloud/service_scope.dart';
 import 'package:http/http.dart' as http;
 import 'package:meta/meta.dart';
 
 import '../../frontend/handlers/pubapi.client.dart';
+import '../../service/services.dart';
 import '../../shared/configuration.dart';
-
 import 'http.dart';
 
 /// Creates an API client with [authToken] that uses the configured HTTP endpoints.
@@ -22,10 +26,60 @@ PubApiClient createPubApiClient({String? authToken}) {
   registerScopeExitCallback(() async => httpClient.close());
   return PubApiClient(
     activeConfiguration.primaryApiUri!.toString(),
-    client: httpClient,
+    client: _FakeTimeClient(httpClient),
   );
 }
 
+class _FakeTimeClient implements http.Client {
+  final http.Client _client;
+  _FakeTimeClient(this._client);
+
+  @override
+  Future<http.StreamedResponse> send(http.BaseRequest request) async {
+    request.headers[fakeClockHeaderName] = clock.now().toIso8601String();
+    return _client.send(request);
+  }
+
+  @override
+  void close() => _client.close();
+
+  @override
+  Future<http.Response> delete(Uri url,
+          {Map<String, String>? headers, Object? body, Encoding? encoding}) =>
+      _client.delete(url, headers: headers, body: body, encoding: encoding);
+
+  @override
+  Future<http.Response> get(Uri url, {Map<String, String>? headers}) =>
+      _client.get(url, headers: headers);
+
+  @override
+  Future<http.Response> head(Uri url, {Map<String, String>? headers}) =>
+      _client.head(url, headers: headers);
+
+  @override
+  Future<http.Response> patch(Uri url,
+          {Map<String, String>? headers, Object? body, Encoding? encoding}) =>
+      _client.patch(url, headers: headers, body: body, encoding: encoding);
+
+  @override
+  Future<http.Response> post(Uri url,
+          {Map<String, String>? headers, Object? body, Encoding? encoding}) =>
+      _client.post(url, headers: headers, body: body, encoding: encoding);
+
+  @override
+  Future<http.Response> put(Uri url,
+          {Map<String, String>? headers, Object? body, Encoding? encoding}) =>
+      _client.put(url, headers: headers, body: body, encoding: encoding);
+
+  @override
+  Future<String> read(Uri url, {Map<String, String>? headers}) =>
+      _client.read(url, headers: headers);
+
+  @override
+  Future<Uint8List> readBytes(Uri url, {Map<String, String>? headers}) =>
+      _client.readBytes(url, headers: headers);
+}
+
 /// Creates a pub.dev API client and executes [fn], making sure that the HTTP
 /// resources are freed after the callback finishes.
 ///
@@ -55,6 +109,7 @@ extension PubApiClientExt on PubApiClient {
     final uploadInfo = await getPackageUploadUrl();
 
     final request = http.MultipartRequest('POST', Uri.parse(uploadInfo.url))
+      ..headers[fakeClockHeaderName] = clock.now().toIso8601String()
       ..fields.addAll(uploadInfo.fields!)
       ..files.add(http.MultipartFile.fromBytes('file', bytes))
       ..followRedirects = false;

app/test/tool/utils/uploader_count_report_test.dart

@@ -0,0 +1,80 @@
+// Copyright (c) 2022, 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 'package:clock/clock.dart';
+import 'package:pub_dev/admin/tools/uploader_count_report.dart';
+import 'package:pub_dev/fake/backend/fake_auth_provider.dart';
+import 'package:pub_dev/tool/test_profile/models.dart';
+import 'package:test/test.dart';
+
+import '../../package/backend_test_utils.dart';
+import '../../shared/test_services.dart';
+
+class UserAndTime {
+  final String uploaderEmail;
+  final DateTime publishingTime;
+  UserAndTime(this.uploaderEmail, this.publishingTime);
+}
+
+void main() {
+  withClock(Clock.fixed(DateTime(2022, 12, 13)), () {
+    testWithProfile('uploader count report',
+        testProfile: TestProfile.fromJson(
+            {'defaultUser': '[email protected]', 'packages': []}), fn: () async {
+      Future<void> uploadAtTime(UserAndTime userAndTime, int cnt) async {
+        await withClock(Clock.fixed(userAndTime.publishingTime), () async {
+          final token = createFakeAuthTokenForEmail(userAndTime.uploaderEmail,
+              audience: 'fake-client-audience');
+          final pubspecContent = '''
+name: 'new_package_$cnt'
+version: '1.2.$cnt'
+author: 'Hans Juergen <[email protected]>'
+description: 'my package description'
+environment:
+  sdk: '>=2.10.0 <3.0.0'
+''';
+          await createPubApiClient(authToken: token).uploadPackageBytes(
+            await packageArchiveBytes(pubspecContent: pubspecContent),
+          );
+        });
+      }
+
+      var i = 0;
+      for (final t in [
+        UserAndTime('[email protected]',
+            DateTime(2021, 12, 2)), // Before range, should not be counted.
+        UserAndTime('[email protected]', DateTime(2022, 1, 5)),
+        UserAndTime('[email protected]',
+            DateTime(2022, 1, 6)), // Same user, should not be counted.
+        UserAndTime('[email protected]', DateTime(2022, 2, 26)),
+        UserAndTime('[email protected]', DateTime(2022, 2, 21)),
+        UserAndTime('[email protected]', DateTime(2022, 2, 22)),
+        UserAndTime('[email protected]', DateTime(2022, 12, 11)),
+      ]) {
+        await uploadAtTime(t, i);
+        i++;
+      }
+
+      final s = await executeCountUploaderReport([]);
+
+      expect(s, '''
+Monthly unique uploading users:
+January 2022: 1
+February 2022: 3
+March 2022: 0
+April 2022: 0
+May 2022: 0
+June 2022: 0
+July 2022: 0
+August 2022: 0
+September 2022: 0
+October 2022: 0
+November 2022: 0
+December 2022: 1
+''');
+    });
+  });
+}

index.yaml

@@ -18,6 +18,11 @@ indexes:
   - name: created
     direction: desc
 
+- kind: AuditLogRecord
+  properties:
+  - name: kind
+  - name: created
+
 - kind: AuditLogRecord
   properties:
   - name: publishers