chore(cloud_auth): Improve performance of ListUsers

services/celest_cloud_auth/lib/src/database/schema/cloud_auth_users.drift

@@ -117,6 +117,38 @@ getUser:
     SELECT * FROM cloud_auth_users
     WHERE user_id = :user_id;
 
+listUsers(
+    :start_time AS DATETIME OR NULL,
+    :offset AS INTEGER,
+    :limit AS INTEGER
+):
+    WITH rowed AS(
+        SELECT 
+            ROW_NUMBER() OVER (ORDER BY create_time DESC) AS row_num,
+            user_id
+        FROM cloud_auth_users
+        WHERE
+            cloud_auth_users.create_time < coalesce(:start_time, unixepoch('now', '+1 second', 'subsec'))
+    )
+    SELECT
+        row_num,
+        cloud_auth_users.**,
+        LIST(
+            SELECT *
+            FROM cloud_auth_user_emails
+            WHERE user_id = rowed.user_id
+        ) AS emails,
+        LIST(
+            SELECT *
+            FROM cloud_auth_user_phone_numbers
+            WHERE user_id = rowed.user_id
+        ) AS phone_numbers
+    FROM cloud_auth_users
+    INNER JOIN rowed ON cloud_auth_users.user_id = rowed.user_id
+    WHERE row_num > :offset
+    ORDER BY $order_by
+    LIMIT :limit;
+
 deleteUser:
     DELETE FROM cloud_auth_users
     WHERE user_id = :user_id;

services/celest_cloud_auth/lib/src/database/schema/cloud_auth_users.drift.dart

@@ -1072,6 +1072,52 @@ class CloudAuthUsersDrift extends i4.ModularAccessor {
         }).asyncMap(cloudAuthUsers.mapFromRow);
   }
 
+  i0.Selectable<ListUsersResult> listUsers(
+      {DateTime? startTime,
+      required int offset,
+      ListUsers$orderBy? order_by,
+      required int limit}) {
+    var $arrayStartIndex = 4;
+    final generatedorder_by = $write(
+        order_by?.call(this.cloudAuthUsers) ?? const i0.OrderBy.nothing(),
+        startIndex: $arrayStartIndex);
+    $arrayStartIndex += generatedorder_by.amountOfVariables;
+    return customSelect(
+        'WITH rowed AS (SELECT ROW_NUMBER()OVER (ORDER BY create_time DESC RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW EXCLUDE NO OTHERS) AS row_num, user_id FROM cloud_auth_users WHERE cloud_auth_users.create_time < coalesce(?1, unixepoch(\'now\', \'+1 second\', \'subsec\'))) SELECT row_num,"cloud_auth_users"."user_id" AS "nested_0.user_id", "cloud_auth_users"."given_name" AS "nested_0.given_name", "cloud_auth_users"."family_name" AS "nested_0.family_name", "cloud_auth_users"."time_zone" AS "nested_0.time_zone", "cloud_auth_users"."language_code" AS "nested_0.language_code", "cloud_auth_users"."create_time" AS "nested_0.create_time", "cloud_auth_users"."update_time" AS "nested_0.update_time", rowed.user_id AS "\$n_0", rowed.user_id AS "\$n_1" FROM cloud_auth_users INNER JOIN rowed ON cloud_auth_users.user_id = rowed.user_id WHERE row_num > ?2 ${generatedorder_by.sql} LIMIT ?3',
+        variables: [
+          i0.Variable<DateTime>(startTime),
+          i0.Variable<int>(offset),
+          i0.Variable<int>(limit),
+          ...generatedorder_by.introducedVariables
+        ],
+        readsFrom: {
+          cloudAuthUsers,
+          cloudAuthUserEmails,
+          cloudAuthUserPhoneNumbers,
+          ...generatedorder_by.watchedTables,
+        }).asyncMap((i0.QueryRow row) async => ListUsersResult(
+          rowNum: row.read<int>('row_num'),
+          cloudAuthUsers:
+              await cloudAuthUsers.mapFromRow(row, tablePrefix: 'nested_0'),
+          emails: await customSelect(
+              'SELECT * FROM cloud_auth_user_emails WHERE user_id = ?1',
+              variables: [
+                i0.Variable<String>(row.read('\$n_0'))
+              ],
+              readsFrom: {
+                cloudAuthUserEmails,
+              }).asyncMap(cloudAuthUserEmails.mapFromRow).get(),
+          phoneNumbers: await customSelect(
+              'SELECT * FROM cloud_auth_user_phone_numbers WHERE user_id = ?1',
+              variables: [
+                i0.Variable<String>(row.read('\$n_1'))
+              ],
+              readsFrom: {
+                cloudAuthUserPhoneNumbers,
+              }).asyncMap(cloudAuthUserPhoneNumbers.mapFromRow).get(),
+        ));
+  }
+
   Future<int> deleteUser({required String userId}) {
     return customUpdate(
       'DELETE FROM cloud_auth_users WHERE user_id = ?1',
@@ -1212,6 +1258,22 @@ class CloudAuthUsersDrift extends i4.ModularAccessor {
   i6.CedarDrift get cedarDrift => this.accessor(i6.CedarDrift.new);
 }
 
+class ListUsersResult {
+  final int rowNum;
+  final i1.User cloudAuthUsers;
+  final List<i1.Email> emails;
+  final List<i1.PhoneNumber> phoneNumbers;
+  ListUsersResult({
+    required this.rowNum,
+    required this.cloudAuthUsers,
+    required this.emails,
+    required this.phoneNumbers,
+  });
+}
+
+typedef ListUsers$orderBy = i0.OrderBy Function(
+    i2.CloudAuthUsers cloud_auth_users);
+
 class LookupUserByEmailResult {
   final i1.User cloudAuthUsers;
   final i1.Email cloudAuthUserEmails;

services/celest_cloud_auth/lib/src/users/users_service.dart

@@ -17,7 +17,6 @@ import 'package:celest_cloud_auth/src/model/order_by.dart';
 import 'package:celest_cloud_auth/src/model/page_token.dart';
 import 'package:celest_cloud_auth/src/model/route_map.dart';
 import 'package:celest_core/celest_core.dart';
-import 'package:drift/drift.dart' as drift;
 import 'package:drift/drift.dart';
 import 'package:meta/meta.dart';
 import 'package:shelf/shelf.dart';
@@ -191,74 +190,37 @@ extension type UsersService._(_Deps _deps) implements Object {
     final startTime = pageData?.startTime ??
         DateTime.timestamp().add(const Duration(seconds: 1));
 
-    /*
-    Roughly, we're trying to construct the query:
-
-      WITH rowed AS(
-          SELECT 
-              ROW_NUMBER() OVER (ORDER BY create_time DESC) AS row_num,
-              id,
-              create_time
-          FROM users
-          WHERE create_time <= coalesce(:start_time, unixepoch('now', '+1 second', 'subsec'))
-      )
-      SELECT 
-          row_num, 
-          users.*
-      FROM users
-      JOIN rowed ON users.id = rowed.id
-      WHERE row_num > coalesce(:page_offset, 0)
-      ORDER BY :order_by
-      LIMIT :page_size;
-
-    Since we want a dynamic `ORDER BY` clause, we need to construct the query
-    programmatically. We can't pass `:order_by` as a variable to Drift.
-    */
-
-    const rowNum = CustomExpression<int>(
-      'ROW_NUMBER() OVER (ORDER BY create_time DESC)',
-    );
-    final rowedQuery = Subquery(
-      _db.cloudAuthUsers.selectOnly()
-        ..addColumns([
-          rowNum,
-          _db.cloudAuthUsers.userId,
-          _db.cloudAuthUsers.createTime,
-        ])
-        ..where(
-          _db.cloudAuthUsers.createTime.isSmallerThanValue(startTime),
-        ),
-      'rowed',
-    );
-
-    final query = _db.cloudAuthUsersDrift.select(_db.cloudAuthUsers).join([
-      innerJoin(
-        rowedQuery,
-        _db.cloudAuthUsers.userId
-            .equalsExp(rowedQuery.ref(_db.cloudAuthUsers.userId)),
-        useColumns: false,
-      ),
-    ])
-      ..addColumns([rowedQuery.ref(rowNum)])
-      ..where(
-        rowedQuery.ref(rowNum).isBiggerThanValue(pageOffset),
-      );
-
+    OrderByClause? orderByClause;
     if (orderBy != null) {
-      final clause = OrderByClause.parse(orderBy);
-      query.orderBy(clause.toOrderingTerms(_db.cloudAuthUsers).toList());
+      orderByClause = OrderByClause.parse(orderBy);
     }
 
-    query.limit(pageSize);
-    final rows = await query.get();
+    final rows = await _db.cloudAuthUsersDrift
+        .listUsers(
+          startTime: startTime,
+          offset: pageOffset,
+          limit: pageSize,
+          order_by: (tbl) => switch (orderByClause) {
+            final orderBy? => OrderBy(orderBy.toOrderingTerms(tbl).toList()),
+            _ => OrderBy([
+                OrderingTerm(
+                  expression: tbl.createTime,
+                  mode: OrderingMode.desc,
+                ),
+              ]),
+          },
+        )
+        .get();
     final users = rows
-        .map((user) => user.readTable(_db.cloudAuthUsers).toProto())
+        .map((row) => row.cloudAuthUsers
+            .copyWith(emails: row.emails, phoneNumbers: row.phoneNumbers)
+            .toProto())
         .toList();
     final nextPageToken = rows.isEmpty || rows.length < pageSize
         ? null
         : PageToken(
             startTime: startTime,
-            offset: rows.last.read(rowedQuery.ref(rowNum))!,
+            offset: rows.last.rowNum,
           ).encode();
     return pb.ListUsersResponse(
       users: users,

services/celest_cloud_auth/test/users/users_service_test.dart

@@ -280,6 +280,7 @@ void main() {
                   {
                     'name': 'users/$userId',
                     'userId': userId,
+                    'emails': [user.primaryEmail!.toJson()],
                     'createTime': (Subject<Object?> it) => it
                         .isA<String>()
                         .has(DateTime.parse, 'DateTime')