Mirror Like entities to SQL.

CHANGELOG.md

@@ -3,6 +3,7 @@ AppEngine version, listed here to ease deployment and troubleshooting.
 
 ## Next Release (replace with git tag when deployed)
  * Note: `Publisher` and `PublisherMember` are now mirrored to SQL.
+ * Note: `Like` is now mirrored to SQL.
 
 ## `20261014t070500-all`
  * Bump runtimeVersion to `2026.10.13`.

app/lib/account/like_backend.dart

@@ -6,7 +6,11 @@ import 'dart:async';
 
 import 'package:clock/clock.dart';
 import 'package:gcloud/service_scope.dart' as ss;
+import 'package:logging/logging.dart';
+import 'package:typed_sql/typed_sql.dart' hide AuthenticationException;
 
+import '../database/database.dart';
+import '../database/schema.dart';
 import '../package/backend.dart' show packageBackend;
 import '../package/models.dart';
 import '../shared/datastore.dart';
@@ -16,6 +20,8 @@ import '../shared/redis_cache.dart' show cache;
 import 'backend.dart' show purgeAccountCache;
 import 'models.dart';
 
+final _logger = Logger('like_backend');
+
 /// Sets the like backend service.
 void registerLikeBackend(LikeBackend backend) =>
     ss.register(#_likeBackend, backend);
@@ -83,6 +89,7 @@ class LikeBackend {
     });
     if (updatedPackage != null) {
       await packageBackend.mirrorPackageToSql(updatedPackage!);
+      await mirrorToSql(res);
     }
     await cache.packageView(package).purge();
     await purgeAccountCache(userId: user.userId);
@@ -117,8 +124,107 @@ class LikeBackend {
     });
     if (updatedPackage != null) {
       await packageBackend.mirrorPackageToSql(updatedPackage!);
+      await deleteFromSql(user.userId, package);
     }
     await cache.packageView(package).purge();
     await cache.userPackageLikes(user.userId).purge();
   }
+
+  /// Upserts [like] into SQL. The user and the package must exist in SQL.
+  ///
+  /// A [Like] without `created` gets the current time, also stored in Datastore.
+  Future<void> mirrorToSql(Like like) async {
+    final created = like.created ?? await _setMissingCreated(like);
+    await primaryDatabase.transactWithRetry(
+      (db) => db.packageLikes
+          .upsertValue(
+            userId: like.userId,
+            package: like.package,
+            createdAt: created,
+          )
+          .execute(),
+    );
+  }
+
+  /// Sets the missing `created` of [like] in Datastore, unless it was set
+  /// concurrently, and returns the stored value.
+  Future<DateTime> _setMissingCreated(Like like) async {
+    final updated = await withRetryTransaction<Like?>(_db, (tx) async {
+      final current = await tx.lookupOrNull<Like>(like.key);
+      if (current == null) return null;
+      current.created ??= clock.now().toUtc();
+      tx.insert(current);
+      return current;
+    });
+    // If the entity was removed concurrently, the next backfill drops the row.
+    final created = updated?.created ?? clock.now().toUtc();
+    like.created = created;
+    return created;
+  }
+
+  /// Deletes the SQL row of the like of [userId] on [package].
+  Future<void> deleteFromSql(String userId, String package) async {
+    await primaryDatabase.withRetry(
+      (db) => db.packageLikes.delete(userId, package).execute(),
+    );
+  }
+
+  /// Syncs SQL with the [Like] entities, one package at a time: upserts the
+  /// missing or changed likes, and removes the rows without a matching entity.
+  ///
+  /// Likes of users or packages that are missing from SQL are skipped.
+  Future<int> backfillSqlFromDatastore() async {
+    var count = 0;
+    final packageNames = await packageBackend.allPackageNames().toList();
+    for (final package in packageNames) {
+      final remainingSqlRows = <String, DateTime>{};
+      final rows = await primaryDatabase.withRetry(
+        (db) => db.packageLikes
+            .where((l) => l.package.equalsValue(package))
+            .select((l) => (l.userId, l.createdAt))
+            .fetch(),
+      );
+      for (final (userId, createdAt) in rows) {
+        remainingSqlRows[userId] = createdAt;
+      }
+
+      bool? hasPackageInSql;
+      final query = _db.query<Like>()..filter('packageName =', package);
+      await for (final like in query.run()) {
+        final sqlCreated = remainingSqlRows.remove(like.userId);
+        final created = like.created;
+        if (created != null &&
+            sqlCreated != null &&
+            sqlCreated.isAtSameMomentAs(created)) {
+          continue;
+        }
+        hasPackageInSql ??=
+            await primaryDatabase.withRetry(
+              (db) => db.packages.byKey(package).fetch(),
+            ) !=
+            null;
+        final hasUserInSql =
+            await primaryDatabase.withRetry(
+              (db) => db.users.byKey(like.userId).fetch(),
+            ) !=
+            null;
+        if (!hasPackageInSql || !hasUserInSql) {
+          _logger.warning(
+            'Skipping like of "${like.userId}" on "$package": '
+            'user or package is missing in SQL.',
+          );
+          continue;
+        }
+        await mirrorToSql(like);
+        count++;
+      }
+
+      // Rows left over have no Datastore entity.
+      for (final userId in remainingSqlRows.keys) {
+        await deleteFromSql(userId, package);
+        count++;
+      }
+    }
+    return count;
+  }
 }

app/lib/admin/tools/user_merger.dart

@@ -7,6 +7,7 @@ import 'package:clock/clock.dart';
 import 'package:logging/logging.dart';
 import 'package:pool/pool.dart';
 import 'package:pub_dev/account/backend.dart';
+import 'package:pub_dev/account/like_backend.dart';
 import 'package:pub_dev/account/models.dart';
 import 'package:pub_dev/audit/backend.dart';
 import 'package:pub_dev/audit/models.dart';
@@ -216,12 +217,39 @@ class UserMerger {
     await _processConcurrently(_db.query<Like>(ancestorKey: fromUserKey), (
       Like like,
     ) async {
+      final newLike = like.changeParentUser(toUserKey);
+      Package? updatedPackage;
+      var moved = false;
       await withRetryTransaction(_db, (tx) async {
-        tx.queueMutations(
-          inserts: [like.changeParentUser(toUserKey)],
-          deletes: [like.key],
+        updatedPackage = null;
+        moved = false;
+        final existing = await tx.lookupOrNull<Like>(newLike.key);
+        if (existing == null) {
+          tx.queueMutations(inserts: [newLike], deletes: [like.key]);
+          moved = true;
+          return;
+        }
+        // The target user already likes the package: keep their like, drop
+        // the duplicate and decrement the counter to avoid drifting.
+        final p = await tx.lookupOrNull<Package>(
+          _db.emptyKey.append(Package, id: like.package),
         );
+        if (p == null) {
+          tx.queueMutations(deletes: [like.key]);
+          return;
+        }
+        p.likes--;
+        p.updated = clock.now().toUtc();
+        tx.queueMutations(inserts: [p], deletes: [like.key]);
+        updatedPackage = p;
       });
+      if (moved) {
+        await likeBackend.mirrorToSql(newLike);
+      }
+      await likeBackend.deleteFromSql(like.userId, like.package);
+      if (updatedPackage != null) {
+        await packageBackend.mirrorPackageToSql(updatedPackage!);
+      }
     });
 
     // UserSession

app/lib/database/schema.account.dart

@@ -72,3 +72,32 @@ abstract final class UserRow extends Row {
 
   String? get moderatedReason;
 }
+
+/// A user liking a package.
+@PrimaryKey(['userId', 'package'])
+@ForeignKey(
+  ['package'],
+  table: 'packages',
+  fields: ['name'],
+  name: 'packageRow',
+  as: 'likes',
+  onDelete: .cascade,
+  onUpdate: .cascade,
+)
+@ForeignKey(
+  ['userId'],
+  table: 'users',
+  fields: ['userId'],
+  name: 'userRow',
+  as: 'likedPackages',
+  onDelete: .cascade,
+  onUpdate: .cascade,
+)
+@Index(name: 'packageUser', fields: ['package', 'userId'])
+abstract final class PackageLike extends Row {
+  String get userId;
+
+  String get package;
+
+  DateTime get createdAt;
+}

app/lib/database/schema.dart

@@ -30,6 +30,8 @@ abstract final class PrimarySchema extends Schema {
 
   Table<UserSession> get userSessions;
 
+  Table<PackageLike> get packageLikes;
+
   // audit tables
 
   Table<AuditLogRecordRow> get auditLogRecords;