Cached list of owned packages (#9131)

Author: isoos

app/lib/account/backend.dart

@@ -662,6 +662,7 @@ Future<void> purgeAccountCache({required String userId}) async {
   await Future.wait([
     cache.userPackageLikes(userId).purgeAndRepeat(),
     cache.publisherPage(userId).purgeAndRepeat(),
+    cache.userUploaderOfPackages(userId).purgeAndRepeat(),
   ]);
 }
 

app/lib/package/backend.dart

@@ -250,8 +250,11 @@ class PackageBackend {
   }
 
   /// Streams package names where the [userId] is an uploader.
-  Stream<String> streamPackagesWhereUserIsUploader(String userId) async* {
-    var page = await listPackagesForUser(userId);
+  Stream<String> streamPackagesWhereUserIsUploader(
+    String userId, {
+    int pageSize = 100,
+  }) async* {
+    var page = await listPackagesForUser(userId, limit: pageSize);
     while (page.packages.isNotEmpty) {
       yield* Stream.fromIterable(page.packages);
       if (page.nextPackage == null) {
@@ -262,6 +265,17 @@ class PackageBackend {
     }
   }
 
+  /// Returns the cached list of package names, where the [userId] is an uploader
+  /// (package is not under a publisher).
+  Future<List<String>> cachedPackagesWhereUserIsUploader(String userId) async {
+    final list = await cache.userUploaderOfPackages(userId).get(() async {
+      return await streamPackagesWhereUserIsUploader(
+        userId,
+      ).take(1000).toList();
+    });
+    return list as List<String>;
+  }
+
   /// Returns the latest releases info of a package.
   Future<LatestReleases> latestReleases(Package package) async {
     // TODO: implement runtimeVersion-specific release calculation
@@ -1522,6 +1536,11 @@ class PackageBackend {
     asyncQueue.addAsyncFn(
       () => _postUploadTasks(package, newVersion, outgoingEmail),
     );
+    if (isNew && agent is AuthenticatedUser) {
+      asyncQueue.addAsyncFn(
+        () => cache.userUploaderOfPackages(agent.userId).purge(),
+      );
+    }
 
     _logger.info('Post-upload tasks completed in ${sw.elapsed}.');
     return (pv, uploadMessages);
@@ -1865,7 +1884,7 @@ class PackageBackend {
     User uploader, {
     required String consentRequestFromAgent,
   }) async {
-    await withRetryTransaction(db, (tx) async {
+    final uploaderUserId = await withRetryTransaction(db, (tx) async {
       final packageKey = db.emptyKey.append(Package, id: packageName);
       final package = (await tx.lookup([packageKey])).first as Package;
 
@@ -1878,7 +1897,7 @@ class PackageBackend {
       }
       if (package.containsUploader(uploader.userId)) {
         // The requested uploaderEmail is already part of the uploaders.
-        return;
+        return uploader.userId;
       }
 
       // Add [uploaderEmail] to uploaders and commit.
@@ -1892,7 +1911,9 @@ class PackageBackend {
           package: packageName,
         ),
       );
+      return uploader.userId;
     });
+    await purgeAccountCache(userId: uploaderUserId);
     triggerPackagePostUpdates(
       packageName,
       skipReanalysis: true,
@@ -1923,7 +1944,7 @@ class PackageBackend {
     uploaderEmail = uploaderEmail.toLowerCase();
     final authenticatedUser = await requireAuthenticatedWebUser();
     final user = authenticatedUser.user;
-    await withRetryTransaction(db, (tx) async {
+    final uploaderUserId = await withRetryTransaction(db, (tx) async {
       final packageKey = db.emptyKey.append(Package, id: packageName);
       final package = await tx.lookupOrNull<Package>(packageKey);
       if (package == null) {
@@ -1976,7 +1997,9 @@ class PackageBackend {
           uploaderUser: uploader,
         ),
       );
+      return uploader.userId;
     });
+    await purgeAccountCache(userId: uploaderUserId);
     triggerPackagePostUpdates(
       packageName,
       skipReanalysis: true,

app/lib/shared/redis_cache.dart

@@ -298,6 +298,18 @@ class CachePatterns {
         ),
       )[userId];
 
+  Entry<List<String>> userUploaderOfPackages(String userId) => _cache
+      .withPrefix('user-uploader-of-packages/')
+      .withTTL(Duration(minutes: 60))
+      .withCodec(utf8)
+      .withCodec(json)
+      .withCodec(
+        wrapAsCodec(
+          encode: (List<String> l) => l,
+          decode: (d) => (d as List).cast<String>(),
+        ),
+      )[userId];
+
   Entry<String> secretValue(String secretId) => _cache
       .withPrefix('secret-value/')
       .withTTL(Duration(minutes: 60))

app/test/package/upload_test.dart

@@ -110,6 +110,12 @@ void main() {
       testWithProfile(
         'successful new package',
         fn: () async {
+          final user = await accountBackend.lookupUserByEmail('user@pub.dev');
+          expect(
+            await packageBackend.cachedPackagesWhereUserIsUploader(user.userId),
+            isEmpty,
+          );
+
           final dateBeforeTest = clock.now().toUtc();
           final pubspecContent = generatePubspecYaml('new_package', '1.2.3');
           final message = await createPubApiClient(authToken: userClientToken)
@@ -121,8 +127,6 @@ void main() {
           expect(message.success.message, contains('1.2.3'));
 
           // verify state
-          final user = await accountBackend.lookupUserByEmail('user@pub.dev');
-
           final pkgKey = dbService.emptyKey.append(Package, id: 'new_package');
           final package = (await dbService.lookup<Package>([pkgKey])).single!;
           expect(package.name, 'new_package');
@@ -143,6 +147,11 @@ void main() {
           expect(pv.publisherId, isNull);
 
           await asyncQueue.ongoingProcessing;
+          expect(
+            await packageBackend.cachedPackagesWhereUserIsUploader(user.userId),
+            ['new_package'],
+          );
+
           expect(fakeEmailSender.sentMessages, hasLength(1));
           final email = fakeEmailSender.sentMessages.single;
           expect(email.recipients.single.email, user.email);