<commit_before>// Copyright (c) 2014, 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.

library pub_dartlang_org.backend;

import 'dart:async';
import 'dart:io';
import 'dart:math';

import 'package:gcloud/db.dart';
import 'package:gcloud/service_scope.dart' as ss;
import 'package:gcloud/storage.dart';
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:pub_server/repository.dart';
import 'package:uuid/uuid.dart';

import '../account/backend.dart';
import '../history/backend.dart';
import '../history/models.dart';
import '../shared/analyzer_client.dart';
import '../shared/configuration.dart';
import '../shared/dartdoc_client.dart';
import '../shared/email.dart';
import '../shared/package_memcache.dart';
import '../shared/urls.dart' as urls;
import '../shared/utils.dart';
import 'email_sender.dart';
import 'model_properties.dart';
import 'models.dart' as models;
import 'name_tracker.dart';
import 'upload_signer_service.dart';

final Logger _logger = Logger('pub.cloud_repository');
final _random = Random.secure();
// The maximum stored length of `README.md` and other user-provided file content
// that is stored separately in the database.
final _maxStoredLength = 128 * 1024;

/// Sets the active tarball storage
void registerTarballStorage(TarballStorage ts) =>
    ss.register(#_tarball_storage, ts);

/// The active tarball storage.
TarballStorage get tarballStorage =>
    ss.lookup(#_tarball_storage) as TarballStorage;

/// Sets the backend service.
void registerBackend(Backend backend) => ss.register(#_backend, backend);

/// The active backend service.
Backend get backend => ss.lookup(#_backend) as Backend;

/// Represents the backend for the pub site.
class Backend {
  final DatastoreDB db;
  final GCloudPackageRepository repository;
  final UIPackageCache uiPackageCache;

  Backend(DatastoreDB db, TarballStorage storage, {UIPackageCache cache})
      : db = db,
        repository = GCloudPackageRepository(db, storage, cache: cache),
        uiPackageCache = cache;

  /// Retrieves packages ordered by their created date.
  Future<List<models.Package>> newestPackages({int offset, int limit}) {
    final query = db.query<models.Package>()
      ..order('-created')
      ..offset(offset)
      ..limit(limit);
    return query.run().toList();
  }

  /// Retrieves packages ordered by their latest version date.
  Future<List<models.Package>> latestPackages({int offset, int limit}) {
    final query = db.query<models.Package>()
      ..order('-updated')
      ..offset(offset)
      ..limit(limit);
    return query.run().toList();
  }

  /// Retrieves the names of all packages, ordered by name.
  Stream<String> allPackageNames(
      {DateTime updatedSince, bool excludeDiscontinued = false}) {
    final query = db.query<models.Package>();

    if (updatedSince != null) {
      query.filter('updated >', updatedSince);
    }

    bool isExcluded(models.Package p) =>
        // isDiscontinued may be null
        excludeDiscontinued && p.isDiscontinued == true;

    return query.run().where((p) => !isExcluded(p)).map((p) => p.name);
  }

  /// Retrieves package versions ordered by their latest version date.
  Future<List<models.PackageVersion>> latestPackageVersions(
      {int offset, int limit, bool devVersions = false}) async {
    final packages = await latestPackages(offset: offset, limit: limit);
    return lookupLatestVersions(packages, devVersions: devVersions);
  }

  /// Looks up a package by name.
  Future<models.Package> lookupPackage(String packageName) async {
    final packageKey = db.emptyKey.append(models.Package, id: packageName);
    return (await db.lookup([packageKey])).first as models.Package;
  }

  /// Looks up a package by name.
  Future<List<models.Package>> lookupPackages(
      Iterable<String> packageNames) async {
    return (await db.lookup(packageNames
            .map((p) => db.emptyKey.append(models.Package, id: p))
            .toList()))
        .cast();
  }

  /// Looks up a specific package version.
  Future<models.PackageVersion> lookupPackageVersion(
      String package, String version) async {
    version = canonicalizeVersion(version);
    final packageVersionKey = db.emptyKey
        .append(models.Package, id: package)
        .append(models.PackageVersion, id: version);
    return (await db.lookup([packageVersionKey])).first
        as models.PackageVersion;
  }

  /// Looks up the latest versions of a list of packages.
  Future<List<models.PackageVersion>> lookupLatestVersions(
      List<models.Package> packages,
      {bool devVersions = false}) async {
    final keys = packages.map((models.Package p) {
      if (devVersions) {
        return p.latestDevVersionKey ?? p.latestVersionKey;
      }
      return p.latestVersionKey;
    }).toList();
    return (await db.lookup(keys)).cast();
  }

  /// Looks up all versions of a package.
  Future<List<models.PackageVersion>> versionsOfPackage(
      String packageName) async {
    final packageKey = db.emptyKey.append(models.Package, id: packageName);
    final query = db.query<models.PackageVersion>(ancestorKey: packageKey);
    return await query.run().toList();
  }

  /// Get a [Uri] which can be used to download a tarball of the pub package.
  Future<Uri> downloadUrl(String package, String version) async {
    version = canonicalizeVersion(version);
    assert(repository.supportsDownloadUrl);
    return repository.downloadUrl(package, version);
  }

  /// Stores a [models.PackageInvite] entry in the Datastore and returns its
  /// current status. When set, the status.urlNonce can be used in
  /// client-communication, e.g. sending via e-mail.
  Future<InviteStatus> updatePackageInvite({
    @required String packageName,
    @required String type,
    @required String recipientEmail,
    @required String fromUserId,
    @required String fromEmail,
  }) async {
    final now = DateTime.now().toUtc();
    final inviteId = models.PackageInvite.createId(type, recipientEmail);
    final pkgKey = db.emptyKey.append(models.Package, id: packageName);
    final inviteKey = pkgKey.append(models.PackageInvite, id: inviteId);
    return await db.withTransaction((tx) async {
      final list = await tx.lookup([inviteKey]);
      models.PackageInvite invite = list.single as models.PackageInvite;

      // Existing and active invite with throttled notification.
      if (invite != null && !invite.isExpired() && !invite.shouldNotify()) {
        await tx.rollback();
        return InviteStatus(nextNotification: invite.nextNotification);
      }

      // Existing and active invite with notification enabled.
      if (invite != null && !invite.isExpired()) {
        invite.lastNotified = now;
        invite.notificationCount++;
        tx.queueMutations(inserts: [invite]);
        await tx.commit();
        return InviteStatus(urlNonce: invite.urlNonce);
      }

      // Reset old or create new invite.
      invite ??= models.PackageInvite()
        ..parentKey = pkgKey
        ..id = inviteId
        ..type = type
        ..recipientEmail = recipientEmail;
      final urlNonce =
          List.generate(25, (i) => _random.nextInt(36).toRadixString(36))
              .join();
      invite
        ..urlNonce = urlNonce
        ..fromUserId = fromUserId
        ..fromEmail = fromEmail
        ..created = now
        ..expires = now.add(Duration(days: 1))
        ..notificationCount = 1
        ..lastNotified = now;

      final inserts = <Model>[invite];
      if (historyBackend.isEnabled) {
        final history = History.entry(UploaderInvited(
          packageName: packageName,
          currentUserId: fromUserId,
          currentUserEmail: fromEmail,
          uploaderUserEmail: recipientEmail,
        ));
        inserts.add(history);
      }

      tx.queueMutations(inserts: inserts);
      await tx.commit();
      return InviteStatus(urlNonce: invite.urlNonce);
    }) as InviteStatus;
  }

  /// Get the invite or return null if it does not exist or is not valid anymore.
  Future<models.PackageInvite> getPackageInvite({
    @required String packageName,
    @required String type,
    @required String recipientEmail,
    @required String urlNonce,
  }) async {
    final inviteId = models.PackageInvite.createId(type, recipientEmail);
    final pkgKey = db.emptyKey.append(models.Package, id: packageName);
    final inviteKey = pkgKey.append(models.PackageInvite, id: inviteId);
    final invite = (await db.lookup<models.PackageInvite>([inviteKey])).single;

    if (invite == null) {
      return null;
    }

    if (invite.isValid(recipientEmail: recipientEmail, urlNonce: urlNonce)) {
      return invite;
    }
    return null;
  }

  /// Delete the invite and clear package cache.
  Future confirmPackageInvite(models.PackageInvite invite) async {
    await db.commit(deletes: [invite.key]);
    await uiPackageCache.invalidateUIPackagePage(invite.packageName);
  }

  /// Removes obsolete/expired invites from Datastore.
  Future deleteObsoleteInvites() async {
    final query = db.query<models.PackageInvite>()
      ..filter('expires <', DateTime.now().toUtc());
    await for (var invite in query.run()) {
      try {
        await db.commit(deletes: [invite.key]);
      } catch (e) {
        _logger.info(
            'PackageInvite delete failed: '
            '${invite.packageName} ${invite.type} ${invite.recipientEmail}',
            e);
      }
    }
  }
}

/// The status of an invite after being created or updated.
class InviteStatus {
  final String urlNonce;
  final DateTime nextNotification;

  InviteStatus({this.urlNonce, this.nextNotification});

  bool get isActive => urlNonce != null;

  bool get isDelayed => nextNotification != null;
}

/// A read-only implementation of [PackageRepository] using the Cloud Datastore
/// for metadata and Cloud Storage for tarball storage.
class GCloudPackageRepository extends PackageRepository {
  final Uuid uuid = Uuid();
  final DatastoreDB db;
  final TarballStorage storage;
  final UIPackageCache cache;

  GCloudPackageRepository(this.db, this.storage, {this.cache});

  // Metadata support.

  @override
  Stream<PackageVersion> versions(String package) {
    final packageKey = db.emptyKey.append(models.Package, id: package);
    final query = db.query<models.PackageVersion>(ancestorKey: packageKey);
    return query.run().map((model) =>
        PackageVersion(package, model.version, model.pubspec.jsonString));
  }

  @override
  Future<PackageVersion> lookupVersion(String package, String version) async {
    version = canonicalizeVersion(version);

    final packageVersionKey = db.emptyKey
        .append(models.Package, id: package)
        .append(models.PackageVersion, id: version);

    final pv =
        (await db.lookup([packageVersionKey])).first as models.PackageVersion;
    if (pv == null) return null;
    return PackageVersion(package, version, pv.pubspec.jsonString);
  }

  // Download support.

  @override
  Future<Stream<List<int>>> download(String package, String version) async {
    // TODO: Should we first test for existence?
    // Maybe with a cache?
    version = canonicalizeVersion(version);
    return storage.download(package, version);
  }

  @override
  bool get supportsDownloadUrl => true;

  @override
  Future<Uri> downloadUrl(String package, String version) async {
    version = canonicalizeVersion(version);
    return storage.downloadUrl(package, version);
  }

  // Upload support.

  @override
  bool get supportsUpload => true;

  @override
  Future<PackageVersion> upload(Stream<List<int>> data) {
    _logger.info('Starting upload.');
    return withAuthenticatedUser((AuthenticatedUser user) {
      _logger.info('User: ${user.userId} / ${user.email}.');

      return withTempDirectory((Directory dir) async {
        final filename = '${dir.absolute.path}/tarball.tar.gz';
        await _saveTarballToFS(data, filename);
        return _performTarballUpload(user, filename, (package, version) {
          return storage.upload(package, version, File(filename).openRead());
        });
      });
    });
  }

  @override
  bool get supportsAsyncUpload => true;

  @override
  Future<AsyncUploadInfo> startAsyncUpload(Uri redirectUrl) async {
    _logger.info('Starting async upload.');
    // NOTE: We use a authenticated user scope here to ensure the uploading
    // user is authenticated. But we're not validating anything at this point
    // because we don't even know which package or version is going to be
    // uploaded.
    return withAuthenticatedUser((AuthenticatedUser user) {
      _logger.info('User: ${user.email}.');

      final guid = uuid.v4().toString();
      final String object = storage.tempObjectName(guid);
      final String bucket = storage.bucket.bucketName;
      final Duration lifetime = const Duration(minutes: 10);

      final url = redirectUrl.resolve('?upload_id=$guid');

      _logger
          .info('Redirecting pub client to google cloud storage (uuid: $guid)');
      return uploadSigner.buildUpload(bucket, object, lifetime, '$url');
    });
  }

  /// Finishes the upload of a package.
  @override
  Future<PackageVersion> finishAsyncUpload(Uri uri) {
    return withAuthenticatedUser((AuthenticatedUser user) async {
      final guid = uri.queryParameters['upload_id'];
      _logger.info('Finishing async upload (uuid: $guid)');
      _logger.info('Reading tarball from cloud storage.');

      return withTempDirectory((Directory dir) async {
        final filename = '${dir.absolute.path}/tarball.tar.gz';
        // TODO: check why this is flaky https://github.com/dart-lang/pub-dartlang-dart/issues/1680
        await retryAsync(
            () => _saveTarballToFS(storage.readTempObject(guid), filename));
        return _performTarballUpload(user, filename, (package, version) {
          return storage.uploadViaTempObject(guid, package, version);
        }).whenComplete(() async {
          _logger.info('Removing temporary object $guid.');
          await storage.removeTempObject(guid);
        });
      });<commit_msg>Remove retry that is no longer necessary. (#2370)<commit_after>// Copyright (c) 2014, 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.

library pub_dartlang_org.backend;

import 'dart:async';
import 'dart:io';
import 'dart:math';

import 'package:gcloud/db.dart';
import 'package:gcloud/service_scope.dart' as ss;
import 'package:gcloud/storage.dart';
import 'package:logging/logging.dart';
import 'package:meta/meta.dart';
import 'package:pub_server/repository.dart';
import 'package:uuid/uuid.dart';

import '../account/backend.dart';
import '../history/backend.dart';
import '../history/models.dart';
import '../shared/analyzer_client.dart';
import '../shared/configuration.dart';
import '../shared/dartdoc_client.dart';
import '../shared/email.dart';
import '../shared/package_memcache.dart';
import '../shared/urls.dart' as urls;
import '../shared/utils.dart';
import 'email_sender.dart';
import 'model_properties.dart';
import 'models.dart' as models;
import 'name_tracker.dart';
import 'upload_signer_service.dart';

final Logger _logger = Logger('pub.cloud_repository');
final _random = Random.secure();
// The maximum stored length of `README.md` and other user-provided file content
// that is stored separately in the database.
final _maxStoredLength = 128 * 1024;

/// Sets the active tarball storage
void registerTarballStorage(TarballStorage ts) =>
    ss.register(#_tarball_storage, ts);

/// The active tarball storage.
TarballStorage get tarballStorage =>
    ss.lookup(#_tarball_storage) as TarballStorage;

/// Sets the backend service.
void registerBackend(Backend backend) => ss.register(#_backend, backend);

/// The active backend service.
Backend get backend => ss.lookup(#_backend) as Backend;

/// Represents the backend for the pub site.
class Backend {
  final DatastoreDB db;
  final GCloudPackageRepository repository;
  final UIPackageCache uiPackageCache;

  Backend(DatastoreDB db, TarballStorage storage, {UIPackageCache cache})
      : db = db,
        repository = GCloudPackageRepository(db, storage, cache: cache),
        uiPackageCache = cache;

  /// Retrieves packages ordered by their created date.
  Future<List<models.Package>> newestPackages({int offset, int limit}) {
    final query = db.query<models.Package>()
      ..order('-created')
      ..offset(offset)
      ..limit(limit);
    return query.run().toList();
  }

  /// Retrieves packages ordered by their latest version date.
  Future<List<models.Package>> latestPackages({int offset, int limit}) {
    final query = db.query<models.Package>()
      ..order('-updated')
      ..offset(offset)
      ..limit(limit);
    return query.run().toList();
  }

  /// Retrieves the names of all packages, ordered by name.
  Stream<String> allPackageNames(
      {DateTime updatedSince, bool excludeDiscontinued = false}) {
    final query = db.query<models.Package>();

    if (updatedSince != null) {
      query.filter('updated >', updatedSince);
    }

    bool isExcluded(models.Package p) =>
        // isDiscontinued may be null
        excludeDiscontinued && p.isDiscontinued == true;

    return query.run().where((p) => !isExcluded(p)).map((p) => p.name);
  }

  /// Retrieves package versions ordered by their latest version date.
  Future<List<models.PackageVersion>> latestPackageVersions(
      {int offset, int limit, bool devVersions = false}) async {
    final packages = await latestPackages(offset: offset, limit: limit);
    return lookupLatestVersions(packages, devVersions: devVersions);
  }

  /// Looks up a package by name.
  Future<models.Package> lookupPackage(String packageName) async {
    final packageKey = db.emptyKey.append(models.Package, id: packageName);
    return (await db.lookup([packageKey])).first as models.Package;
  }

  /// Looks up a package by name.
  Future<List<models.Package>> lookupPackages(
      Iterable<String> packageNames) async {
    return (await db.lookup(packageNames
            .map((p) => db.emptyKey.append(models.Package, id: p))
            .toList()))
        .cast();
  }

  /// Looks up a specific package version.
  Future<models.PackageVersion> lookupPackageVersion(
      String package, String version) async {
    version = canonicalizeVersion(version);
    final packageVersionKey = db.emptyKey
        .append(models.Package, id: package)
        .append(models.PackageVersion, id: version);
    return (await db.lookup([packageVersionKey])).first
        as models.PackageVersion;
  }

  /// Looks up the latest versions of a list of packages.
  Future<List<models.PackageVersion>> lookupLatestVersions(
      List<models.Package> packages,
      {bool devVersions = false}) async {
    final keys = packages.map((models.Package p) {
      if (devVersions) {
        return p.latestDevVersionKey ?? p.latestVersionKey;
      }
      return p.latestVersionKey;
    }).toList();
    return (await db.lookup(keys)).cast();
  }

  /// Looks up all versions of a package.
  Future<List<models.PackageVersion>> versionsOfPackage(
      String packageName) async {
    final packageKey = db.emptyKey.append(models.Package, id: packageName);
    final query = db.query<models.PackageVersion>(ancestorKey: packageKey);
    return await query.run().toList();
  }

  /// Get a [Uri] which can be used to download a tarball of the pub package.
  Future<Uri> downloadUrl(String package, String version) async {
    version = canonicalizeVersion(version);
    assert(repository.supportsDownloadUrl);
    return repository.downloadUrl(package, version);
  }

  /// Stores a [models.PackageInvite] entry in the Datastore and returns its
  /// current status. When set, the status.urlNonce can be used in
  /// client-communication, e.g. sending via e-mail.
  Future<InviteStatus> updatePackageInvite({
    @required String packageName,
    @required String type,
    @required String recipientEmail,
    @required String fromUserId,
    @required String fromEmail,
  }) async {
    final now = DateTime.now().toUtc();
    final inviteId = models.PackageInvite.createId(type, recipientEmail);
    final pkgKey = db.emptyKey.append(models.Package, id: packageName);
    final inviteKey = pkgKey.append(models.PackageInvite, id: inviteId);
    return await db.withTransaction((tx) async {
      final list = await tx.lookup([inviteKey]);
      models.PackageInvite invite = list.single as models.PackageInvite;

      // Existing and active invite with throttled notification.
      if (invite != null && !invite.isExpired() && !invite.shouldNotify()) {
        await tx.rollback();
        return InviteStatus(nextNotification: invite.nextNotification);
      }

      // Existing and active invite with notification enabled.
      if (invite != null && !invite.isExpired()) {
        invite.lastNotified = now;
        invite.notificationCount++;
        tx.queueMutations(inserts: [invite]);
        await tx.commit();
        return InviteStatus(urlNonce: invite.urlNonce);
      }

      // Reset old or create new invite.
      invite ??= models.PackageInvite()
        ..parentKey = pkgKey
        ..id = inviteId
        ..type = type
        ..recipientEmail = recipientEmail;
      final urlNonce =
          List.generate(25, (i) => _random.nextInt(36).toRadixString(36))
              .join();
      invite
        ..urlNonce = urlNonce
        ..fromUserId = fromUserId
        ..fromEmail = fromEmail
        ..created = now
        ..expires = now.add(Duration(days: 1))
        ..notificationCount = 1
        ..lastNotified = now;

      final inserts = <Model>[invite];
      if (historyBackend.isEnabled) {
        final history = History.entry(UploaderInvited(
          packageName: packageName,
          currentUserId: fromUserId,
          currentUserEmail: fromEmail,
          uploaderUserEmail: recipientEmail,
        ));
        inserts.add(history);
      }

      tx.queueMutations(inserts: inserts);
      await tx.commit();
      return InviteStatus(urlNonce: invite.urlNonce);
    }) as InviteStatus;
  }

  /// Get the invite or return null if it does not exist or is not valid anymore.
  Future<models.PackageInvite> getPackageInvite({
    @required String packageName,
    @required String type,
    @required String recipientEmail,
    @required String urlNonce,
  }) async {
    final inviteId = models.PackageInvite.createId(type, recipientEmail);
    final pkgKey = db.emptyKey.append(models.Package, id: packageName);
    final inviteKey = pkgKey.append(models.PackageInvite, id: inviteId);
    final invite = (await db.lookup<models.PackageInvite>([inviteKey])).single;

    if (invite == null) {
      return null;
    }

    if (invite.isValid(recipientEmail: recipientEmail, urlNonce: urlNonce)) {
      return invite;
    }
    return null;
  }

  /// Delete the invite and clear package cache.
  Future confirmPackageInvite(models.PackageInvite invite) async {
    await db.commit(deletes: [invite.key]);
    await uiPackageCache.invalidateUIPackagePage(invite.packageName);
  }

  /// Removes obsolete/expired invites from Datastore.
  Future deleteObsoleteInvites() async {
    final query = db.query<models.PackageInvite>()
      ..filter('expires <', DateTime.now().toUtc());
    await for (var invite in query.run()) {
      try {
        await db.commit(deletes: [invite.key]);
      } catch (e) {
        _logger.info(
            'PackageInvite delete failed: '
            '${invite.packageName} ${invite.type} ${invite.recipientEmail}',
            e);
      }
    }
  }
}

/// The status of an invite after being created or updated.
class InviteStatus {
  final String urlNonce;
  final DateTime nextNotification;

  InviteStatus({this.urlNonce, this.nextNotification});

  bool get isActive => urlNonce != null;

  bool get isDelayed => nextNotification != null;
}

/// A read-only implementation of [PackageRepository] using the Cloud Datastore
/// for metadata and Cloud Storage for tarball storage.
class GCloudPackageRepository extends PackageRepository {
  final Uuid uuid = Uuid();
  final DatastoreDB db;
  final TarballStorage storage;
  final UIPackageCache cache;

  GCloudPackageRepository(this.db, this.storage, {this.cache});

  // Metadata support.

  @override
  Stream<PackageVersion> versions(String package) {
    final packageKey = db.emptyKey.append(models.Package, id: package);
    final query = db.query<models.PackageVersion>(ancestorKey: packageKey);
    return query.run().map((model) =>
        PackageVersion(package, model.version, model.pubspec.jsonString));
  }

  @override
  Future<PackageVersion> lookupVersion(String package, String version) async {
    version = canonicalizeVersion(version);

    final packageVersionKey = db.emptyKey
        .append(models.Package, id: package)
        .append(models.PackageVersion, id: version);

    final pv =
        (await db.lookup([packageVersionKey])).first as models.PackageVersion;
    if (pv == null) return null;
    return PackageVersion(package, version, pv.pubspec.jsonString);
  }

  // Download support.

  @override
  Future<Stream<List<int>>> download(String package, String version) async {
    // TODO: Should we first test for existence?
    // Maybe with a cache?
    version = canonicalizeVersion(version);
    return storage.download(package, version);
  }

  @override
  bool get supportsDownloadUrl => true;

  @override
  Future<Uri> downloadUrl(String package, String version) async {
    version = canonicalizeVersion(version);
    return storage.downloadUrl(package, version);
  }

  // Upload support.

  @override
  bool get supportsUpload => true;

  @override
  Future<PackageVersion> upload(Stream<List<int>> data) {
    _logger.info('Starting upload.');
    return withAuthenticatedUser((AuthenticatedUser user) {
      _logger.info('User: ${user.userId} / ${user.email}.');

      return withTempDirectory((Directory dir) async {
        final filename = '${dir.absolute.path}/tarball.tar.gz';
        await _saveTarballToFS(data, filename);
        return _performTarballUpload(user, filename, (package, version) {
          return storage.upload(package, version, File(filename).openRead());
        });
      });
    });
  }

  @override
  bool get supportsAsyncUpload => true;

  @override
  Future<AsyncUploadInfo> startAsyncUpload(Uri redirectUrl) async {
    _logger.info('Starting async upload.');
    // NOTE: We use a authenticated user scope here to ensure the uploading
    // user is authenticated. But we're not validating anything at this point
    // because we don't even know which package or version is going to be
    // uploaded.
    return withAuthenticatedUser((AuthenticatedUser user) {
      _logger.info('User: ${user.email}.');

      final guid = uuid.v4().toString();
      final String object = storage.tempObjectName(guid);
      final String bucket = storage.bucket.bucketName;
      final Duration lifetime = const Duration(minutes: 10);

      final url = redirectUrl.resolve('?upload_id=$guid');

      _logger
          .info('Redirecting pub client to google cloud storage (uuid: $guid)');
      return uploadSigner.buildUpload(bucket, object, lifetime, '$url');
    });
  }

  /// Finishes the upload of a package.
  @override
  Future<PackageVersion> finishAsyncUpload(Uri uri) {
    return withAuthenticatedUser((AuthenticatedUser user) async {
      final guid = uri.queryParameters['upload_id'];
      _logger.info('Finishing async upload (uuid: $guid)');
      _logger.info('Reading tarball from cloud storage.');

      return withTempDirectory((Directory dir) async {
        final filename = '${dir.absolute.path}/tarball.tar.gz';
        await _saveTarballToFS(storage.readTempObject(guid), filename);
        return _performTarballUpload(user, filename, (package, version) {
          return storage.uploadViaTempObject(guid, package, version);
        }).whenComplete(() async {
          _logger.info('Removing temporary object $guid.');
          await storage.removeTempObject(guid);
        });
      });
{'fraction_non_alphanumeric': 0.1163110562511021, 'fraction_numerical': 0.0018338917298536413, 'mean_word_length': 3.3504142374961643, 'pattern_counts': {'":': 0, '<': 83, '<?xml version=': 0, '>': 114, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2619150', 'n_tokens_mistral': 8155, 'n_tokens_neox': 8008, 'n_words': 2193}