WIP: change event processing.

Author: isoos

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 'package:pub_dev/admin/models.dart';
+import 'package:pub_dev/service/change_event/change_event.dart';
 
 import '../audit/models.dart';
 import '../frontend/request_context.dart';
@@ -462,7 +463,7 @@ class AccountBackend {
     if (user == null || user.isModerated || user.isDeleted) {
       throw AuthenticationException.failed();
     }
-    final data = await withRetryTransaction(_db, (tx) async {
+    final (data, change) = await withRetryTransaction(_db, (tx) async {
       final session = await tx.userSessions.lookupOrNull(sessionId);
       if (session == null || session.isExpired()) {
         throw AuthenticationException.failed('Session has been expired.');
@@ -473,7 +474,8 @@ class AccountBackend {
           oldUserId != user.userId) {
         // expire old session
         tx.delete(session.key);
-        await cache.userSessionData(sessionId).purgeAndRepeat();
+        final change =
+            CapturedChange(ChangeAction.delete, UserSession, [sessionId]);
 
         // create a new session
         final newSession = UserSession.init()
@@ -487,7 +489,7 @@ class AccountBackend {
           ..authenticatedAt = now
           ..expires = now.add(_sessionDuration);
         tx.insert(newSession);
-        return SessionData.fromModel(newSession);
+        return (SessionData.fromModel(newSession), change);
       } else {
         // only update the current one
         session
@@ -500,10 +502,13 @@ class AccountBackend {
           ..authenticatedAt = now
           ..expires = now.add(_sessionDuration);
         tx.insert(session);
-        return SessionData.fromModel(session);
+        return (SessionData.fromModel(session), null);
       }
     });
     await cache.userSessionData(data.sessionId).set(data);
+    if (change != null) {
+      changeEventAggregator.addChange(change);
+    }
     return data;
   }
 
@@ -698,7 +703,8 @@ class _UserSessionDataAccess {
     } on Exception catch (_) {
       // ignore if the entity has been already deleted concurrently
     }
-    await cache.userSessionData(sessionId).purge();
+    changeEventAggregator.addChange(
+        CapturedChange(ChangeAction.delete, UserSession, [sessionId]));
   }
 
   /// Removes the session data that has expiry before [ts].
@@ -719,3 +725,12 @@ class _UserSessionTransactionDataAcccess {
     return await _tx.lookupOrNull<UserSession>(key);
   }
 }
+
+Iterable<TriggeredEvent> processUserSessionChange(CapturedChange change) sync* {
+  if (change.entity == UserSession && change.action == ChangeAction.delete) {
+    yield TriggeredEvent(
+      [UserSession, ...change.keys],
+      () => cache.userSessionData(change.keys.single as String).purge(),
+    );
+  }
+}

app/lib/package/backend.dart

@@ -20,6 +20,7 @@ import 'package:pub_dev/package/api_export/api_exporter.dart';
 import 'package:pub_dev/package/tarball_storage.dart';
 import 'package:pub_dev/scorecard/backend.dart';
 import 'package:pub_dev/service/async_queue/async_queue.dart';
+import 'package:pub_dev/service/change_event/change_event.dart';
 import 'package:pub_dev/service/rate_limit/rate_limit.dart';
 import 'package:pub_dev/shared/versions.dart';
 import 'package:pub_dev/task/backend.dart';
@@ -480,9 +481,8 @@ class PackageBackend {
         options: optionsChanges,
       ));
     });
-    await purgePackageCache(package);
-    await taskBackend.trackPackage(package);
-    await apiExporter.synchronizePackage(package);
+    changeEventAggregator
+        .addChange(CapturedChange(ChangeAction.update, Package, [package]));
   }
 
   /// Updates [options] on [package]/[version], assuming the current user
@@ -2096,3 +2096,22 @@ class _VersionTransactionDataAcccess {
     return await _tx.query<PackageVersion>(pkgKey).run().toList();
   }
 }
+
+Iterable<TriggeredEvent> processPackageChange(CapturedChange change) sync* {
+  if (change.entity != Package) return;
+  final package = change.keys.single as String;
+  if (change.action == ChangeAction.update) {
+    yield change.toTriggeredEvent(
+      ['purge-cache'],
+      () => purgePackageCache(package),
+    );
+    yield change.toTriggeredEvent(
+      ['track-task'],
+      () => taskBackend.trackPackage(package),
+    );
+    yield change.toTriggeredEvent(
+      ['export-package'],
+      () => apiExporter.synchronizePackage(package),
+    );
+  }
+}

app/lib/service/change_event/change_event.dart

@@ -0,0 +1,104 @@
+// Copyright (c) 2025, 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 'package:gcloud/service_scope.dart' as ss;
+
+import 'package:pub_dev/service/async_queue/async_queue.dart';
+
+/// Sets the [ChangeEventAggregator] service in the scope.
+void registerChangeEventAggregator(ChangeEventAggregator service) =>
+    ss.register(#_changeEventAggregator, service);
+
+/// The active [ChangeEventAggregator] service.
+ChangeEventAggregator get changeEventAggregator =>
+    ss.lookup(#_changeEventAggregator) as ChangeEventAggregator;
+
+/// Creates events that need to be run after given [change].
+typedef ChangeProcessorFn = Iterable<TriggeredEvent> Function(
+    CapturedChange change);
+
+/// Processes entity changes that could trigger further events (e.g. cache invalidation
+/// or re-running export to CDN bucket).
+///
+/// The events are aggregated (duplicates are filtered) and run in batches, off-request.
+class ChangeEventAggregator {
+  final List<ChangeProcessorFn> _changeProcessorFns;
+  final _events = <String, TriggeredEvent>{};
+  bool _scheduled = false;
+
+  ChangeEventAggregator(this._changeProcessorFns);
+
+  /// Digests the [change] and may trigger new event(s).
+  ///
+  /// The events are scheduled via [AsyncQueue], off-request.
+  void addChange(CapturedChange change) {
+    for (final fn in _changeProcessorFns) {
+      for (final event in fn(change)) {
+        if (_events.containsKey(event.deduplicateKey)) {
+          continue;
+        }
+        _events[event.deduplicateKey] = event;
+      }
+    }
+    _scheduleProcessing();
+  }
+
+  void _scheduleProcessing() {
+    if (_scheduled) return;
+    _scheduled = true;
+    asyncQueue.addAsyncFn(() async {
+      _scheduled = false;
+      await executeEvents();
+      if (_events.isNotEmpty) {
+        _scheduleProcessing();
+      }
+    });
+  }
+
+  /// Executes the scheduled events.
+  Future<void> executeEvents() async {
+    if (_events.isEmpty) {
+      return;
+    }
+    // Clears event map to prevent concurrent updates while the events are processed.
+    final fns = _events.values.map((e) => e.fn).toList();
+    _events.clear();
+
+    // Processing the events without any batching.
+    // TODO: consider concurrency for events that are cheap (e.g. cache invalidation).
+    for (final fn in fns) {
+      await fn();
+    }
+  }
+}
+
+/// Describes the entity being changed with [keys] and other relevant [fields].
+class CapturedChange {
+  final ChangeAction action;
+  final Type entity;
+  final List<Object> keys;
+  final Map<String, Object?>? fields;
+
+  CapturedChange(this.action, this.entity, this.keys, [this.fields]);
+
+  TriggeredEvent toTriggeredEvent(Iterable<Object>? parameters, AsyncFn fn) =>
+      TriggeredEvent([entity, action, ...keys, ...?parameters], fn);
+}
+
+enum ChangeAction {
+  create,
+  update,
+  delete,
+}
+
+/// Describes the event that was triggered.
+class TriggeredEvent {
+  final List<Object?> parameters;
+  final AsyncFn fn;
+
+  TriggeredEvent(this.parameters, this.fn);
+
+  late final deduplicateKey =
+      Uri(pathSegments: parameters.map((p) => p.toString())).toString();
+}

app/lib/service/services.dart

@@ -16,9 +16,6 @@ import 'package:googleapis_auth/auth_io.dart' as auth;
 import 'package:logging/logging.dart';
 import 'package:pub_dev/package/api_export/api_exporter.dart';
 import 'package:pub_dev/search/handlers.dart';
-import 'package:pub_dev/service/async_queue/async_queue.dart';
-import 'package:pub_dev/service/download_counts/backend.dart';
-import 'package:pub_dev/service/security_advisories/backend.dart';
 import 'package:shelf/shelf.dart' as shelf;
 import 'package:shelf/shelf_io.dart';
 
@@ -47,7 +44,11 @@ import '../search/backend.dart';
 import '../search/search_client.dart';
 import '../search/top_packages.dart';
 import '../search/updater.dart';
+import '../service/async_queue/async_queue.dart';
+import '../service/change_event/change_event.dart';
+import '../service/download_counts/backend.dart';
 import '../service/email/backend.dart';
+import '../service/security_advisories/backend.dart';
 import '../service/youtube/backend.dart';
 import '../shared/configuration.dart';
 import '../shared/datastore.dart';
@@ -250,6 +251,10 @@ Future<R> _withPubServices<R>(FutureOr<R> Function() fn) async {
     ));
     registerAsyncQueue(AsyncQueue());
     registerAuditBackend(AuditBackend(dbService));
+    registerChangeEventAggregator(ChangeEventAggregator([
+      processPackageChange,
+      processUserSessionChange,
+    ]));
     registerConsentBackend(ConsentBackend(dbService));
     registerEmailBackend(EmailBackend(dbService));
     registerLikeBackend(LikeBackend(dbService));