Return all users from GET /v1/users/, paginating 1k at a time (#12766)

* Return all users from `GET /v1/users/`, paginating 1k at a time

The test runs overall for ~30 seconds, mostly due to the time required to
create 20k users.

changelog_begin
[HTTP-JSON API] Previously, a 10k limit was in place on the number of users returned by
`GET /v1/users/`. This limit has been removed and users are retrieved in chunks from the ledger
changelog_end

As of now, this builds the response in-memory, which is likely not what we want to
support long-term.

* Relax test condition due to interference with other test setups

ledger-service/http-json/src/it/scala/http/HttpServiceIntegrationTestUserManagement.scala

@@ -18,7 +18,7 @@ import com.daml.ledger.api.v1.{value => v}
 import com.daml.lf.data.Ref
 import com.daml.platform.sandbox.{SandboxRequiringAuthorization, SandboxRequiringAuthorizationFuns}
 import com.typesafe.scalalogging.StrictLogging
-import org.scalatest.{AsyncTestSuite, Inside}
+import org.scalatest.{Assertion, AsyncTestSuite, Inside}
 import org.scalatest.freespec.AsyncFreeSpec
 import org.scalatest.matchers.should.Matchers
 import scalaz.NonEmptyList
@@ -579,6 +579,71 @@ class HttpServiceIntegrationTestUserManagementNoAuth
       }
     } yield assertion
   }
+
+  "creating and listing 20K users should be possible" in withHttpServiceAndClient(
+    participantAdminJwt
+  ) { (uri, _, _, _, _) =>
+    import spray.json._
+    import spray.json.DefaultJsonProtocol._
+
+    val createdUsers = 20000
+
+    val createUserRequests: List[domain.CreateUserRequest] =
+      List.tabulate(createdUsers) { sequenceNumber =>
+        {
+          val p = getUniqueParty(f"p$sequenceNumber%05d")
+          domain.CreateUserRequest(
+            p.unwrap,
+            Some(p.unwrap),
+            Some(
+              List[domain.UserRight](
+                domain.CanActAs(p),
+                domain.ParticipantAdmin,
+              )
+            ),
+          )
+        }
+      }
+
+    // Create users in chunks to avoid overloading the server
+    // https://doc.akka.io/docs/akka-http/current/client-side/pool-overflow.html
+    def createUsers(
+        createUserRequests: Seq[domain.CreateUserRequest],
+        chunkSize: Int = 20,
+    ): Future[Assertion] = {
+      createUserRequests.splitAt(chunkSize) match {
+        case (Nil, _) => Future.successful(succeed)
+        case (next, remainingRequests) =>
+          Future
+            .sequence {
+              next.map { request =>
+                postRequest(
+                  uri.withPath(Uri.Path("/v1/user/create")),
+                  request.toJson,
+                  headers = authorizationHeader(participantAdminJwt),
+                ).map(_._1)
+              }
+            }
+            .flatMap { statusCodes =>
+              all(statusCodes) shouldBe StatusCodes.OK
+              createUsers(remainingRequests)
+            }
+      }
+    }
+
+    for {
+      _ <- createUsers(createUserRequests)
+      (status, output) <- getRequest(
+        uri.withPath(Uri.Path("/v1/users")),
+        headers = authorizationHeader(participantAdminJwt),
+      )
+    } yield {
+      status shouldBe StatusCodes.OK
+      val userIds = getResult(output).convertTo[List[UserDetails]].map(_.userId)
+      val expectedUserIds = "participant_admin" :: createUserRequests.map(_.userId)
+      userIds should contain allElementsOf expectedUserIds
+    }
+  }
 }
 
 class HttpServiceIntegrationTestUserManagement

ledger-service/http-json/src/main/scala/com/digitalasset/http/Endpoints.scala

@@ -492,17 +492,22 @@ class Endpoints(
       } yield emptyObjectResponse
     }(req)
 
+  private def aggregateListUserPages(
+      token: Option[String],
+      pageToken: String = "",
+      pageSize: Int = 1000, // TODO could be made configurable in the future
+  ): Future[Seq[User]] =
+    userManagementClient.listUsers(token, pageToken, pageSize).flatMap {
+      case (users, "") => Future.successful(users)
+      case (users, pageToken) => aggregateListUserPages(token, pageToken, pageSize).map(users ++ _)
+    }
+
   def listUsers(req: HttpRequest)(implicit
       lc: LoggingContextOf[InstanceUUID with RequestID]
   ): ET[domain.SyncResponse[List[domain.UserDetails]]] =
     for {
       jwt <- eitherT(input(req)).bimap(identity[Error], _._1)
-      users <- EitherT.rightT(
-        // TODO participant user management: Emulating no-pagination
-        userManagementClient.listUsers(Some(jwt.value), pageToken = "", pageSize = 10000).map {
-          case (users, _) => users
-        }
-      )
+      users <- EitherT.rightT(aggregateListUserPages(Some(jwt.value)))
     } yield domain.OkResponse(users.map(domain.UserDetails.fromUser).toList)
 
   def listUserRights(req: HttpRequest)(implicit