Mirror ModerationCase entities to SQL.

CHANGELOG.md

@@ -4,6 +4,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.
+ * Note: `ModerationCase` is now mirrored to SQL.
 
 ## `20261014t070500-all`
  * Bump runtimeVersion to `2026.10.13`.

app/lib/account/backend.dart

@@ -12,6 +12,7 @@ import 'package:logging/logging.dart';
 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/backend.dart';
 import 'package:pub_dev/admin/models.dart';
 import 'package:typed_sql/typed_sql.dart' hide AuthenticationException;
 
@@ -722,6 +723,9 @@ class AccountBackend {
       return user;
     });
     await mirrorUserToSql(user);
+    if (refCaseKey != null) {
+      await adminBackend.mirrorModerationCaseToSqlById(refCaseKey.id as String);
+    }
     await _expireSessionsForUserId(userId);
     await purgeAccountCache(userId: userId);
   }

app/lib/admin/actions/migrate_dry_run.dart

@@ -22,19 +22,18 @@ ALTER TABLE "consents"
 ALTER TABLE "consents"
   ADD CONSTRAINT "consents_unique_dedup_id" UNIQUE ("dedup_id");
 ''',
-  // TODO: re-enable after ModerationCase migration lands
-  //   // Fails if a case was resolved before it was opened.
-  //   'moderation-cases-chk-resolved': '''
-  // ALTER TABLE "moderation_cases"
-  //   ADD CONSTRAINT "moderation_cases_chk_resolved"
-  //   CHECK (resolved_at IS NULL OR resolved_at >= opened_at);
-  // ''',
-  //   // Fails if any appealed_case_id has no matching moderation_cases row.
-  //   'moderation-cases-fk-appealed': '''
-  // ALTER TABLE "moderation_cases"
-  //   ADD CONSTRAINT "moderation_cases_fk_appealed"
-  //   FOREIGN KEY ("appealed_case_id") REFERENCES "moderation_cases" ("case_id");
-  // ''',
+  // Fails if a case was resolved before it was opened.
+  'moderation-cases-chk-resolved': '''
+ALTER TABLE "moderation_cases"
+  ADD CONSTRAINT "moderation_cases_chk_resolved"
+  CHECK (resolved_at IS NULL OR resolved_at >= opened_at);
+''',
+  // Fails if any appealed_case_id has no matching moderation_cases row.
+  'moderation-cases-fk-appealed': '''
+ALTER TABLE "moderation_cases"
+  ADD CONSTRAINT "moderation_cases_fk_appealed"
+  FOREIGN KEY ("appealed_case_id") REFERENCES "moderation_cases" ("case_id");
+''',
   // Fails if any package_version_assets.package has no matching packages row.
   'package-version-assets-fk-package': '''
 ALTER TABLE "package_version_assets"

app/lib/admin/actions/moderate_package.dart

@@ -38,7 +38,7 @@ Note: the action may take a longer time to complete as the public archive bucket
     final refCase = await adminBackend
         .loadAndVerifyModerationCaseForAdminAction(caseId);
 
-    return await adminMarkPackageVisibility(
+    final result = await adminMarkPackageVisibility(
       package,
       state: state,
       whenUpdating: (tx, p, valueToSet) async {
@@ -59,6 +59,10 @@ Note: the action may take a longer time to complete as the public archive bucket
         'moderatedAt': p.moderatedAt?.toIso8601String(),
       },
     );
+    if (refCase != null) {
+      await adminBackend.mirrorModerationCaseToSqlById(refCase.caseId);
+    }
+    return result;
   },
 );
 

app/lib/admin/actions/moderate_package_versions.dart

@@ -43,7 +43,7 @@ Set the moderated flag on a package version (updating the flag and the timestamp
     final refCase = await adminBackend
         .loadAndVerifyModerationCaseForAdminAction(caseId);
 
-    return await adminMarkPackageVersionVisibility(
+    final result = await adminMarkPackageVersionVisibility(
       package,
       version,
       state: state,
@@ -65,6 +65,10 @@ Set the moderated flag on a package version (updating the flag and the timestamp
         'moderatedAt': v.moderatedAt?.toIso8601String(),
       },
     );
+    if (refCase != null) {
+      await adminBackend.mirrorModerationCaseToSqlById(refCase.caseId);
+    }
+    return result;
   },
 );
 

app/lib/admin/actions/moderate_publisher.dart

@@ -78,6 +78,9 @@ can't be updated, administrators must not be able to update publisher options.
         return p;
       });
       await publisherBackend.mirrorPublisherToSql(publisherId);
+      if (refCase != null) {
+        await adminBackend.mirrorModerationCaseToSqlById(refCase.caseId);
+      }
       await purgePublisherCache(publisherId);
     }
 

app/lib/admin/actions/moderation_case_create.dart

@@ -3,6 +3,7 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'package:pub_dev/account/agent.dart';
+import 'package:pub_dev/admin/backend.dart';
 import 'package:pub_dev/admin/models.dart';
 import 'package:pub_dev/frontend/handlers/report.dart';
 import 'package:pub_dev/service/email/email_templates.dart';
@@ -72,6 +73,7 @@ Returns the fields on the newly created moderation case.
       tx.insert(mc);
       return mc;
     });
+    await adminBackend.mirrorModerationCaseToSql(mc);
 
     return mc.toDebugInfo();
   },

app/lib/admin/actions/moderation_case_delete.dart

@@ -2,6 +2,7 @@
 // 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:pub_dev/admin/backend.dart';
 import 'package:pub_dev/admin/models.dart';
 import 'package:pub_dev/shared/datastore.dart';
 
@@ -31,6 +32,7 @@ Deletes a moderation case.
         throw NotFoundException.resource(caseId!);
       }
     });
+    await adminBackend.deleteModerationCaseFromSql(caseId!);
 
     return {'deleted': true};
   },

app/lib/admin/actions/moderation_case_resolve.dart

@@ -3,6 +3,7 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'package:clock/clock.dart';
+import 'package:pub_dev/admin/backend.dart';
 import 'package:pub_dev/admin/models.dart';
 import 'package:pub_dev/shared/datastore.dart';
 import 'package:pub_dev/shared/utils.dart';
@@ -130,6 +131,7 @@ Closes the moderation case and updates the status based on the actions logged on
       tx.insert(mc);
       return mc;
     });
+    await adminBackend.mirrorModerationCaseToSqlById(mc.caseId);
 
     return mc.toDebugInfo();
   },

app/lib/admin/actions/moderation_case_update.dart

@@ -76,7 +76,8 @@ Returns the fields of the updated moderation case.
       tx.insert(mc);
     });
     final mc = await adminBackend.lookupModerationCase(caseId);
+    await adminBackend.mirrorModerationCaseToSql(mc!);
 
-    return mc!.toDebugInfo();
+    return mc.toDebugInfo();
   },
 );

app/lib/admin/backend.dart

@@ -15,6 +15,8 @@ import 'package:gcloud/service_scope.dart' as ss;
 import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
 import 'package:pool/pool.dart';
+import 'package:typed_sql/typed_sql.dart';
+import 'package:ulid/ulid.dart';
 
 import '../account/backend.dart';
 import '../account/consent_backend.dart';
@@ -23,6 +25,8 @@ import '../account/models.dart';
 import '../admin/models.dart';
 import '../audit/backend.dart';
 import '../audit/models.dart';
+import '../database/database.dart';
+import '../database/schema.dart';
 import '../package/backend.dart'
     show checkPackageVersionParams, packageBackend, triggerPackagePostUpdates;
 import '../package/models.dart';
@@ -905,6 +909,146 @@ class AdminBackend {
     );
   }
 
+  /// Upserts [mc] and its action log into SQL.
+  Future<void> mirrorModerationCaseToSql(ModerationCase mc) async {
+    await primaryDatabase.transactWithRetry((db) async {
+      await db.moderationCases
+          .upsertValue(
+            caseId: mc.caseId,
+            reporterEmail: mc.reporterEmail,
+            openedAt: mc.opened,
+            resolvedAt: mc.resolved,
+            source: mc.source,
+            kind: mc.kind,
+            subject: mc.subject,
+            url: mc.url,
+            appealedCaseId: mc.appealedCaseId,
+            isSubjectOwner: mc.isSubjectOwner,
+            status: mc.status!,
+            grounds: mc.grounds,
+            violation: mc.violation,
+            reason: mc.reason,
+          )
+          .execute();
+      await db.moderationActionLogs
+          .where((l) => l.caseId.equalsValue(mc.caseId))
+          .delete()
+          .execute();
+      final entries = mc.getActionLog().entries.sorted(
+        (a, b) => a.timestamp.compareTo(b.timestamp),
+      );
+      final ulidFactory = UlidFactory(monotonic: true);
+      for (final e in entries) {
+        await db.moderationActionLogs
+            .upsertValue(
+              caseId: mc.caseId,
+              logId: ulidFactory
+                  .next(millis: e.timestamp.millisecondsSinceEpoch)
+                  .toUuid(),
+              createdAt: e.timestamp,
+              subject: e.subject,
+              action: e.moderationAction.name,
+              note: e.note,
+            )
+            .execute();
+      }
+    });
+  }
+
+  /// Looks up the [ModerationCase] with [caseId] in Datastore and mirrors it
+  /// into SQL.
+  Future<void> mirrorModerationCaseToSqlById(String caseId) async {
+    final mc = await lookupModerationCase(caseId);
+    if (mc != null) {
+      await mirrorModerationCaseToSql(mc);
+    }
+  }
+
+  /// Deletes the SQL rows of the case with [caseId] (including its action log).
+  Future<void> deleteModerationCaseFromSql(String caseId) async {
+    await primaryDatabase.withRetry(
+      (db) => db.moderationCases.delete(caseId).execute(),
+    );
+  }
+
+  /// Syncs SQL with the [ModerationCase] entities: upserts the missing or
+  /// changed cases, and removes the rows without a matching entity.
+  ///
+  /// Returns the number of cases that were changed in SQL.
+  Future<int> backfillModerationCasesSqlFromDatastore() async {
+    var count = 0;
+    final sqlCaseIds = await primaryDatabase.withRetry(
+      (db) => db.moderationCases.select((c) => (c.caseId,)).fetch(),
+    );
+    final remainingCaseIds = sqlCaseIds.toSet();
+
+    await for (final mc in _db.query<ModerationCase>().run()) {
+      remainingCaseIds.remove(mc.caseId);
+      if (await _isModerationCaseInSql(mc)) {
+        continue;
+      }
+      await mirrorModerationCaseToSql(mc);
+      count++;
+    }
+    // Rows left over seem to have no Datastore entity: double-check before
+    // deleting, in case the entity was created after the scan.
+    for (final caseId in remainingCaseIds) {
+      if (await lookupModerationCase(caseId) != null) {
+        continue;
+      }
+      await deleteModerationCaseFromSql(caseId);
+      count++;
+    }
+    return count;
+  }
+
+  Future<bool> _isModerationCaseInSql(ModerationCase mc) async {
+    final row = await primaryDatabase.withRetry(
+      (db) => db.moderationCases.byKey(mc.caseId).fetch(),
+    );
+    if (row == null ||
+        row.reporterEmail != mc.reporterEmail ||
+        !row.openedAt.isAtSameMomentAs(mc.opened) ||
+        !_isSameMomentOrBothNull(row.resolvedAt, mc.resolved) ||
+        row.source != mc.source ||
+        row.kind != mc.kind ||
+        row.subject != mc.subject ||
+        row.url != mc.url ||
+        row.appealedCaseId != mc.appealedCaseId ||
+        row.isSubjectOwner != mc.isSubjectOwner ||
+        row.status != mc.status ||
+        row.grounds != mc.grounds ||
+        row.violation != mc.violation ||
+        row.reason != mc.reason) {
+      return false;
+    }
+    final logs = await primaryDatabase.withRetry(
+      (db) => db.moderationActionLogs
+          .where((l) => l.caseId.equalsValue(mc.caseId))
+          .orderBy((l) => [(l.logId, Order.ascending)])
+          .fetch(),
+    );
+    final entries = mc.getActionLog().entries;
+    if (logs.length != entries.length) {
+      return false;
+    }
+    final sorted = entries.sorted((a, b) => a.timestamp.compareTo(b.timestamp));
+    for (var i = 0; i < logs.length; i++) {
+      final l = logs[i];
+      final e = sorted[i];
+      if (!l.createdAt.isAtSameMomentAs(e.timestamp) ||
+          l.subject != e.subject ||
+          l.action != e.moderationAction.name ||
+          l.note != e.note) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  static bool _isSameMomentOrBothNull(DateTime? a, DateTime? b) =>
+      a == null ? b == null : (b != null && a.isAtSameMomentAs(b));
+
   /// Returns a valid [ModerationCase] if it exists.
   /// Returns `null` if [caseId] is `none`.
   ///
@@ -944,6 +1088,7 @@ class AdminBackend {
       // delete the entity
       _logger.info('Deleting ModerationCase: ${mc.caseId}');
       await withRetryTransaction(_db, (tx) async => tx.delete(mc.key));
+      await deleteModerationCaseFromSql(mc.caseId);
       _logger.info('Deleted ModerationCase: ${mc.caseId}');
     }
   }

app/lib/database/schema.dart

@@ -16,6 +16,7 @@ part 'schema.audit.dart';
 part 'schema.consent.dart';
 part 'schema.email.dart';
 part 'schema.global_lock.dart';
+part 'schema.moderation.dart';
 part 'schema.neat_task.dart';
 part 'schema.package.dart';
 part 'schema.publisher.dart';
@@ -50,6 +51,12 @@ abstract final class PrimarySchema extends Schema {
 
   Table<GlobalLockState> get globalLockStates;
 
+  // moderation tables
+
+  Table<ModerationCaseRow> get moderationCases;
+
+  Table<ModerationActionLog> get moderationActionLogs;
+
   // neat periodic task status table
 
   Table<NeatTaskStatusRow> get neatTaskStatuses;