feat: Implement authorizers to use data from ProjectAuth graph (#1688)

* Allow filter for project-auth service

* security finder using project-auth data

* Move authorizers to kg module

* auth finder for dataset id and sameAs

acceptance-tests/src/test/scala/io/renku/graph/acceptancetests/db/EventLog.scala

@@ -60,15 +60,19 @@ object EventLog extends TypeSerializers {
       session.prepare(query).flatMap(_.stream(projectId, 32).compile.toList)
   }
 
-  def findSyncEvents(projectId: GitLabId)(implicit ioRuntime: IORuntime): List[CategoryName] = execute { session =>
-    val query: Query[projects.GitLabId, CategoryName] = sql"""
+  def findSyncEventsIO(projectId: GitLabId): IO[List[CategoryName]] =
+    sessionResource.flatMap(_.session).use { session =>
+      val query: Query[projects.GitLabId, CategoryName] = sql"""
           SELECT category_name
           FROM subscription_category_sync_time
           WHERE project_id = $projectIdEncoder"""
-      .query(varchar)
-      .map(category => CategoryName(category))
-    session.prepare(query).flatMap(_.stream(projectId, 32).compile.toList)
-  }
+        .query(varchar)
+        .map(category => CategoryName(category))
+      session.prepare(query).flatMap(_.stream(projectId, 32).compile.toList)
+    }
+
+  def findSyncEvents(projectId: GitLabId)(implicit ioRuntime: IORuntime): List[CategoryName] =
+    findSyncEventsIO(projectId).unsafeRunSync()
 
   def forceCategoryEventTriggering(categoryName: CategoryName, projectId: projects.GitLabId)(implicit
       ioRuntime: IORuntime

acceptance-tests/src/test/scala/io/renku/graph/acceptancetests/db/TriplesStore.scala

@@ -22,6 +22,8 @@ import cats.effect.{IO, Resource, Temporal}
 import cats.{Applicative, Monad}
 import eu.timepit.refined.auto._
 import io.renku.db.DBConfigProvider
+import io.renku.graph.model.{RenkuUrl, projects}
+import io.renku.projectauth.{ProjectAuthData, QueryFilter}
 import io.renku.triplesgenerator.TgLockDB.SessionResource
 import io.renku.triplesgenerator.{TgLockDB, TgLockDbConfigProvider}
 import io.renku.triplesstore._
@@ -54,4 +56,10 @@ object TriplesStore extends InMemoryJena with ProjectsDataset with MigrationsDat
   private def waitForReadiness(implicit logger: Logger[IO]): IO[Unit] =
     Monad[IO].whileM_(IO(!isRunning))(logger.info("Waiting for TS") >> (Temporal[IO] sleep (500 millis)))
 
+  def findProjectAuth(
+      slug: projects.Slug
+  )(implicit renkuUrl: RenkuUrl, sqtr: SparqlQueryTimeRecorder[IO], L: Logger[IO]): IO[Option[ProjectAuthData]] =
+    ProjectSparqlClient[IO](projectsDSConnectionInfo)
+      .map(_.asProjectAuthService)
+      .use(_.getAll(QueryFilter.all.withSlug(slug)).compile.last)
 }

acceptance-tests/src/test/scala/io/renku/graph/acceptancetests/flows/TSProvisioning.scala

@@ -16,31 +16,35 @@
  * limitations under the License.
  */
 
-package io.renku.graph.acceptancetests.flows
+package io.renku.graph.acceptancetests
+package flows
 
 import cats.data.NonEmptyList
 import cats.effect.IO
 import cats.effect.unsafe.IORuntime
+import cats.syntax.all._
 import fs2.Stream
+import io.renku.eventlog.events.producers.membersync.{categoryName => memberSyncCategory}
+import io.renku.eventlog.events.producers.minprojectinfo.{categoryName => minProjectInfoCategory}
 import io.renku.events.CategoryName
 import io.renku.graph.acceptancetests.data
-import io.renku.graph.acceptancetests.db.EventLog
+import io.renku.graph.acceptancetests.db.{EventLog, TriplesStore}
 import io.renku.graph.acceptancetests.testing.AcceptanceTestPatience
-import io.renku.graph.acceptancetests.tooling.EventLogClient.ProjectEvent
-import io.renku.graph.acceptancetests.tooling.{AcceptanceSpec, ApplicationServices, ModelImplicits}
 import io.renku.graph.model.events.{CommitId, EventId, EventStatus, EventStatusProgress}
 import io.renku.graph.model.projects
 import io.renku.http.client.AccessToken
+import io.renku.logging.TestSparqlQueryTimeRecorder
 import io.renku.testtools.IOSpec
+import io.renku.triplesstore.SparqlQueryTimeRecorder
 import io.renku.webhookservice.model.HookToken
 import org.http4s.Status._
 import org.scalatest.concurrent.Eventually
 import org.scalatest.matchers.should
 import org.scalatest.{Assertion, EitherValues}
 import org.typelevel.log4cats.Logger
+import tooling.EventLogClient.ProjectEvent
+import tooling.{AcceptanceSpec, ApplicationServices, ModelImplicits}
 
-import java.lang.Thread.sleep
-import scala.annotation.tailrec
 import scala.concurrent.duration._
 
 trait TSProvisioning
@@ -77,6 +81,8 @@ trait TSProvisioning
     // commitId is the eventId
     val condition = commitIds.map(e => EventId(e.value)).toList.map(_ -> EventStatus.TriplesStore)
     waitForAllEvents(project.id, condition: _*)
+    waitForSyncEvents(project.id, memberSyncCategory)
+    waitForProjectAuthData(project.slug)
   }
 
   private def projectEvents(projectId: projects.GitLabId): Stream[IO, List[ProjectEvent]] = {
@@ -119,24 +125,57 @@ trait TSProvisioning
     lastValue.forall(_ == EventStatusProgress.Stage.Final) shouldBe true
   }
 
-  def `check hook cannot be found`(projectId: projects.GitLabId, accessToken: AccessToken): Assertion = eventually {
-    webhookServiceClient.`GET projects/:id/events/status`(projectId, accessToken).status shouldBe NotFound
+  def getSyncEvents(projectId: projects.GitLabId) = {
+    val getSyncEvents = EventLog.findSyncEventsIO(projectId)
+
+    val waitTimes = Stream.iterate(1d)(_ * 1.5).map(_.seconds).covary[IO].evalMap(IO.sleep)
+    Stream
+      .repeatEval(getSyncEvents)
+      .zip(waitTimes)
+      .map(_._1)
   }
 
-  def `wait for the Fast Tract event`(projectId: projects.GitLabId)(implicit ioRuntime: IORuntime): Unit = eventually {
+  def waitForSyncEvents(projectId: projects.GitLabId, category1: CategoryName, categoryN: CategoryName*) = {
+    val expected = categoryN.toSet + category1
 
-    val sleepTime = 1 second
+    val tries =
+      getSyncEvents(projectId)
+        .evalTap(l => Logger[IO].info(s"Sync events for project $projectId: ${l.mkString(", ")}"))
+        .takeThrough(evs => expected.intersect(evs.toSet) != expected)
+        .take(13)
 
-    @tailrec
-    def checkIfWasSent(categoryName: CategoryName, attempt: Int = 1): Unit = {
-      if (attempt > 20) fail(s"'$categoryName' event wasn't sent after ${(sleepTime * attempt).toSeconds}")
+    val lastValue = tries.compile.lastOrError.unsafeRunSync()
+    expected.intersect(lastValue.toSet) shouldBe expected
+  }
 
-      if (!EventLog.findSyncEvents(projectId).contains(categoryName)) {
-        sleep(sleepTime.toMillis)
-        checkIfWasSent(categoryName)
-      }
-    }
+  def getProjectAuthData(slug: projects.Slug) = {
+    implicit val sqtr: SparqlQueryTimeRecorder[IO] = TestSparqlQueryTimeRecorder.createUnsafe
+    val waitTimes = Stream.iterate(1d)(_ * 1.5).map(_.seconds).covary[IO].evalMap(IO.sleep)
 
-    checkIfWasSent(CategoryName("ADD_MIN_PROJECT_INFO"))
+    Stream
+      .repeatEval(TriplesStore.findProjectAuth(slug))
+      .zip(waitTimes)
+      .map(_._1)
   }
+
+  def waitForProjectAuthData(slug: projects.Slug) = {
+    val tries =
+      getProjectAuthData(slug)
+        .evalTap {
+          case None           => Logger[IO].info(show"auth data not ready for $slug")
+          case Some(authData) => Logger[IO].info(show"auth data ready $authData")
+        }
+        .takeThrough(_.isEmpty)
+        .take(15)
+
+    val lastValue = tries.compile.lastOrError.unsafeRunSync()
+    lastValue.isDefined shouldBe true
+  }
+
+  def `check hook cannot be found`(projectId: projects.GitLabId, accessToken: AccessToken): Assertion = eventually {
+    webhookServiceClient.`GET projects/:id/events/status`(projectId, accessToken).status shouldBe NotFound
+  }
+
+  def `wait for the Fast Tract event`(projectId: projects.GitLabId) =
+    waitForSyncEvents(projectId, minProjectInfoCategory)
 }

acceptance-tests/src/test/scala/io/renku/graph/acceptancetests/knowledgegraph/DatasetsResourcesSpec.scala

@@ -23,17 +23,17 @@ import eu.timepit.refined.auto._
 import io.circe.Json
 import io.circe.literal._
 import io.renku.generators.CommonGraphGenerators.authUsers
-import io.renku.generators.Generators._
 import io.renku.generators.Generators.Implicits._
+import io.renku.generators.Generators._
 import io.renku.graph.acceptancetests.data
 import io.renku.graph.acceptancetests.data._
 import io.renku.graph.acceptancetests.flows.TSProvisioning
-import io.renku.graph.acceptancetests.tooling.{AcceptanceSpec, ApplicationServices}
 import io.renku.graph.acceptancetests.tooling.TestReadabilityTools._
-import io.renku.graph.model._
+import io.renku.graph.acceptancetests.tooling.{AcceptanceSpec, ApplicationServices}
 import io.renku.graph.model.EventsGenerators.commitIds
-import io.renku.graph.model.testentities.{::~, creatorUsernameUpdaterInternal}
+import io.renku.graph.model._
 import io.renku.graph.model.testentities.generators.EntitiesGenerators._
+import io.renku.graph.model.testentities.{::~, creatorUsernameUpdaterInternal}
 import io.renku.http.client.UrlEncoder.urlEncode
 import io.renku.http.rest.Links.Rel
 import io.renku.http.server.EndpointTester._
@@ -57,23 +57,22 @@ class DatasetsResourcesSpec
 
   Feature("GET knowledge-graph/projects/<namespace>/<name>/datasets to find project's datasets") {
 
-    val (dataset1 -> dataset2 -> dataset2Modified, testProject) =
-      renkuProjectEntities(visibilityPublic, creatorGen = cliShapedPersons)
-        .modify(removeMembers())
-        .addDataset(datasetEntities(provenanceInternal(cliShapedPersons)))
-        .addDatasetAndModification(
-          datasetEntities(provenanceInternal(cliShapedPersons)),
-          creatorGen = cliShapedPersons
-        )
-        .generateOne
-    val creatorPerson = cliShapedPersons.generateOne
-    val project =
-      dataProjects(testProject)
-        .map(replaceCreatorFrom(creatorPerson, creator.id))
-        .map(addMemberFrom(creatorPerson, creator.id) >>> addMemberWithId(user.id))
-        .generateOne
-
     Scenario("As a user I would like to find project's datasets by calling a REST endpoint") {
+      val (dataset1 -> dataset2 -> dataset2Modified, testProject) =
+        renkuProjectEntities(visibilityPublic, creatorGen = cliShapedPersons)
+          .modify(removeMembers())
+          .addDataset(datasetEntities(provenanceInternal(cliShapedPersons)))
+          .addDatasetAndModification(
+            datasetEntities(provenanceInternal(cliShapedPersons)),
+            creatorGen = cliShapedPersons
+          )
+          .generateOne
+      val creatorPerson = cliShapedPersons.generateOne
+      val project =
+        dataProjects(testProject)
+          .map(replaceCreatorFrom(creatorPerson, creator.id))
+          .map(addMemberFrom(creatorPerson, creator.id) >>> addMemberWithId(user.id))
+          .generateOne
 
       Given("some data in the Triples Store")
       gitLabStub.addAuthenticated(creator)

build.sbt

@@ -108,7 +108,10 @@ lazy val graphCommons = project
   .in(file("graph-commons"))
   .withId("graph-commons")
   .settings(commonSettings)
-  .dependsOn(renkuModel % "compile->compile; test->test")
+  .dependsOn(
+    renkuModel  % "compile->compile; test->test",
+    projectAuth % "compile->compile; test->test"
+  )
   .enablePlugins(AutomateHeaderPlugin)
 
 lazy val eventLogApi = project

graph-commons/src/main/scala/io/renku/graph/http/server/security/Authorizer.scala

@@ -40,7 +40,10 @@ object Authorizer {
                                   projectSlug:    projects.Slug,
                                   allowedPersons: Set[persons.GitLabId]
   )
-  trait SecurityRecordFinder[F[_], Key] extends ((Key, Option[AuthUser]) => F[List[SecurityRecord]])
+  trait SecurityRecordFinder[F[_], Key] extends ((Key, Option[AuthUser]) => F[List[SecurityRecord]]) {
+    def asAuthorizer(implicit F: MonadThrow[F]): Authorizer[F, Key] =
+      Authorizer.of(this)
+  }
 
   final case class AuthContext[Key](maybeAuthUser: Option[AuthUser], key: Key, allowedProjects: Set[projects.Slug]) {
     def addAllowedProject(slug: projects.Slug): AuthContext[Key] = copy(allowedProjects = allowedProjects + slug)
@@ -52,6 +55,9 @@ object Authorizer {
       AuthContext(None, key, allowedProjects)
   }
 
+  def of[F[_]: MonadThrow, K](securityRecordFinder: SecurityRecordFinder[F, K]): Authorizer[F, K] =
+    new AuthorizerImpl[F, K](securityRecordFinder)
+
   def using[F[_]: Async: Logger, Key](
       securityRecordsFinderFactory: F[SecurityRecordFinder[F, Key]]
   ): F[Authorizer[F, Key]] = securityRecordsFinderFactory.map(new AuthorizerImpl[F, Key](_))