Initial migration of task backend to typed_sql

Author: jonasfj

app/build.yaml

@@ -6,6 +6,7 @@ targets:
       - '$package$'
       - 'lib/$lib$'
       - 'lib/account/models.dart'
+      - 'lib/database/**'
       - 'lib/admin/models.dart'
       - 'lib/dartdoc/models.dart'
       - 'lib/frontend/handlers/pubapi.dart'

app/lib/admin/tools/delete_all_staging.dart

@@ -55,6 +55,7 @@ Future<String> executeDeleteAllStaging(List<String> args) async {
     dbService.query<ModeratedPackage>(): 500,
     dbService.query<NeatTaskStatus>(): 500,
     dbService.query<Secret>(): 500,
+    // ignore: deprecated_member_use_from_same_package
     dbService.query<PackageState>(): 100,
   };
 

app/lib/search/backend.dart

@@ -19,6 +19,7 @@ import 'package:meta/meta.dart';
 import 'package:pana/src/dartdoc/pub_dartdoc_data.dart';
 import 'package:path/path.dart' as p;
 import 'package:pool/pool.dart';
+import 'package:pub_dev/database/model.dart';
 import 'package:pub_dev/shared/monitoring.dart';
 import 'package:retry/retry.dart';
 
@@ -40,7 +41,6 @@ import '../shared/storage.dart';
 import '../shared/versions.dart';
 import '../task/backend.dart';
 import '../task/global_lock.dart';
-import '../task/models.dart';
 
 import 'dart_sdk_mem_index.dart';
 import 'flutter_sdk_mem_index.dart';
@@ -85,10 +85,11 @@ void registerSearchIndex(SearchIndex index) =>
 
 /// Datastore-related access methods for the search service
 class SearchBackend {
+  final Database<PrimaryDatabase> db;
   final DatastoreDB _db;
   final VersionedJsonStorage _snapshotStorage;
 
-  SearchBackend(this._db, Bucket snapshotBucket)
+  SearchBackend(this.db, this._db, Bucket snapshotBucket)
       : _snapshotStorage = VersionedJsonStorage(snapshotBucket, 'snapshot/');
 
   /// Runs a forever loop and tries to get a global lock.
@@ -255,11 +256,14 @@ class SearchBackend {
       addResult(p.name!, p.updated!);
     }
 
-    final q3 = _db.query<PackageState>()
-      ..filter('finished >=', updatedThreshold)
-      ..order('-finished');
-    await for (final s in q3.run()) {
-      addResult(s.package, s.finished);
+    final q2 = db.tasks
+        .where((task) =>
+            task.runtimeVersion.equalsValue(runtimeVersion) &
+            task.finished.isAfterValue(updatedThreshold))
+        .orderBy((task) => [(task.finished, Order.descending)])
+        .select((task) => (task.package, task.finished));
+    await for (final (package, finished) in q2.stream()) {
+      addResult(package, finished);
     }
 
     return results;

app/lib/service/services.dart

@@ -3,6 +3,7 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async' show FutureOr, Zone;
+import 'dart:io' show Platform;
 
 import 'package:_pub_shared/utils/http.dart';
 import 'package:appengine/appengine.dart';
@@ -14,6 +15,10 @@ import 'package:gcloud/service_scope.dart';
 import 'package:gcloud/storage.dart';
 import 'package:googleapis_auth/auth_io.dart' as auth;
 import 'package:logging/logging.dart';
+import 'package:postgres/postgres.dart' show Pool, Endpoint, PoolSettings;
+import 'package:pub_dev/database/model.dart';
+import 'package:pub_dev/fake/backend/fake_database.dart'
+    show createFakeDatabaseAdaptor;
 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';
@@ -87,6 +92,29 @@ Future<void> withServices(FutureOr<void> Function() fn) async {
       final retryingAuthClient = httpRetryClient(innerClient: authClient);
       registerScopeExitCallback(() async => retryingAuthClient.close());
 
+      final databaseAdapter = DatabaseAdapter.postgres(
+        Pool.withEndpoints(
+          [
+            Endpoint(
+              host: Platform.environment['PGHOST'] ?? '127.0.0.1',
+              port: int.tryParse(Platform.environment['PGPORT'] ?? '') ?? 5432,
+              database: Platform.environment['PGDATABASE'] ?? 'postgres',
+              username: Platform.environment['PGUSER'] ?? 'postgres',
+              password: Platform.environment['PGPASSWORD'] ?? 'postgres',
+            ),
+          ],
+          settings: PoolSettings(
+            applicationName: 'pub-dev',
+            maxConnectionCount: 10,
+          ),
+        ),
+      );
+      registerDatabase(Database<PrimaryDatabase>(
+        databaseAdapter,
+        SqlDialect.postgres(),
+      ));
+      registerScopeExitCallback(() => databaseAdapter.close());
+
       // override storageService with retrying http client
       registerStorageService(
           Storage(retryingAuthClient, activeConfiguration.projectId));
@@ -179,6 +207,14 @@ Future<R> withFakeServices<R>({
     }
 
     // register fake services that would have external dependencies
+    final databaseAdapter = createFakeDatabaseAdaptor();
+    registerDatabase(Database<PrimaryDatabase>(
+      databaseAdapter,
+      SqlDialect.postgres(),
+    ));
+    await database.createTables();
+    registerScopeExitCallback(() => databaseAdapter.close());
+
     registerSecretBackend(FakeSecretBackend({}));
     registerAuthProvider(FakeAuthProvider());
     registerScopeExitCallback(authProvider.close);
@@ -261,7 +297,7 @@ Future<R> _withPubServices<R>(FutureOr<R> Function() fn) async {
     registerIndexUpdater(IndexUpdater(dbService));
     registerPublisherBackend(PublisherBackend(dbService));
     registerScoreCardBackend(ScoreCardBackend(dbService));
-    registerSearchBackend(SearchBackend(dbService,
+    registerSearchBackend(SearchBackend(database, dbService,
         storageService.bucket(activeConfiguration.searchSnapshotBucketName!)));
     registerSearchClient(SearchClient());
     registerSearchAdapter(SearchAdapter());
@@ -280,6 +316,7 @@ Future<R> _withPubServices<R>(FutureOr<R> Function() fn) async {
       storageService.bucket(activeConfiguration.publicPackagesBucketName!),
     ));
     registerTaskBackend(TaskBackend(
+      database,
       dbService,
       storageService.bucket(activeConfiguration.taskResultBucketName!),
     ));